wrolpi/modules/videos/test/test_video.py

import json
from datetime import datetime, timezone
from http import HTTPStatus

import pytest

from modules.videos.models import Video
from wrolpi.errors import FileGroupIsTagged
from wrolpi.files.models import FileGroup


def test_delete_video_no_channel(test_session, simple_video):
    """A Video can be deleted even if it does not have a channel."""
    assert simple_video.channel
    simple_video.channel_id = None
    test_session.commit()

    paths = list(simple_video.file_group.my_paths())

    assert not simple_video.channel
    simple_video.delete()

    # All files were deleted.
    for path in paths:
        assert not path.exists()

    test_session.commit()

    assert test_session.query(FileGroup).count() == 0
    assert test_session.query(Video).count() == 0


@pytest.mark.asyncio
async def test_delete_video_with_tag(async_client, test_session, video_factory, tag_factory):
    """You cannot delete a video with a tag."""
    video = video_factory(with_video_file=True, with_poster_ext='png')
    tag = await tag_factory()
    video.add_tag(test_session, tag.name)
    test_session.commit()

    with pytest.raises(FileGroupIsTagged):
        video.delete()

    # Video was not deleted.
    test_session.commit()
    assert test_session.query(Video).count() == 1
    assert test_session.query(FileGroup).count() == 1


@pytest.mark.asyncio
async def test_delete_video_api_tagged_requires_force(async_client, test_session, video_factory, tag_factory,
                                                     await_switches):
    """The unified delete endpoint returns 409 for tagged videos until force=true is passed."""
    video = video_factory(with_video_file=True, with_poster_ext='png')
    tag = await tag_factory()
    video.add_tag(test_session, tag.name)
    test_session.commit()
    file_group_id = video.file_group_id
    paths = list(video.file_group.my_paths())

    body = json.dumps({'file_group_ids': [file_group_id]})
    request, response = await async_client.post('/api/files/delete_groups', content=body)
    assert response.status_code == HTTPStatus.CONFLICT
    assert response.json['code'] == 'FILE_GROUP_IS_TAGGED'
    assert len(response.json['file_groups']) == 1
    assert response.json['file_groups'][0]['id'] == file_group_id
    assert tag.name in response.json['file_groups'][0]['tags']
    # Nothing was deleted.
    assert all(p.is_file() for p in paths)
    assert test_session.query(Video).count() == 1

    # With force=true, the Video is deleted.
    body = json.dumps({'file_group_ids': [file_group_id], 'force': True})
    request, response = await async_client.post('/api/files/delete_groups', content=body)
    assert response.status_code == HTTPStatus.NO_CONTENT
    assert all(not p.exists() for p in paths)
    assert test_session.query(Video).count() == 0
    assert test_session.query(FileGroup).count() == 0
    await await_switches()


@pytest.mark.asyncio
async def test_delete_video_api_atomic_when_some_tagged(async_client, test_session, video_factory, tag_factory):
    """When deleting multiple Videos and any are tagged, NONE are deleted until force=true."""
    untagged = video_factory(with_video_file=True)
    tagged = video_factory(with_video_file=True)
    tag = await tag_factory()
    tagged.add_tag(test_session, tag.name)
    test_session.commit()
    untagged_paths = list(untagged.file_group.my_paths())
    tagged_paths = list(tagged.file_group.my_paths())

    body = json.dumps({'file_group_ids': [untagged.file_group_id, tagged.file_group_id]})
    request, response = await async_client.post('/api/files/delete_groups', content=body)
    assert response.status_code == HTTPStatus.CONFLICT
    assert response.json['code'] == 'FILE_GROUP_IS_TAGGED'
    returned_ids = {fg['id'] for fg in response.json['file_groups']}
    assert returned_ids == {tagged.file_group_id}
    # Nothing was deleted.
    assert all(p.is_file() for p in untagged_paths)
    assert all(p.is_file() for p in tagged_paths)
    assert test_session.query(Video).count() == 2


@pytest.mark.asyncio
async def test_video_channel_refresh(async_client, test_session, test_directory, channel_factory, video_factory,
                                     refresh_files):
    """A Video will be associated with a Channel when it's files are in that Channel's directory."""
    # Create a video file in this channel's directory.
    channel = channel_factory()
    video1_path = channel.directory / 'video1.mp4'
    video_factory(channel_id=channel.id, with_video_file=video1_path, with_poster_ext='jpg')
    # Create another video outside the channel's directory.  It should not have a channel.
    video2_path = test_directory / 'video2.mp4'
    video_factory(with_video_file=video2_path)
    # Delete all FileGroups so the channel will be associated again.
    test_session.query(FileGroup).delete()
    test_session.commit()

    await refresh_files()

    assert test_session.query(FileGroup).count() == 2
    assert test_session.query(Video).count() == 2

    # The video is associated with `simple_channel` because its file is in the Channel's directory.
    video1: Video = Video.get_by_path(test_session, video1_path)
    video2: Video = Video.get_by_path(test_session, video2_path)
    assert video1.channel == channel
    assert not video2.channel
    # Serializers do not query: load what Video's `__json__` reads, as the queries that serialize Videos do.
    video1 = test_session.query(Video).options(*Video.json_options()).populate_existing() \
        .filter(Video.id == video1.id).one()
    assert video1.__json__()['video']['channel']
    video_channel = video1.__json__()['video']['channel']
    assert video_channel['id'] == channel.id
    assert video_channel['name'] == channel.name


@pytest.mark.asyncio
async def test_delete_duplicate_video(async_client, test_session, channel_factory, video_factory, tag_factory):
    """If duplicate Video's exist, and everything about the files matches, delete a random one."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_The video title.mp4'

    channel.info_json = {'entries': [{'id': 'ABC123456', 'title': 'The video title'}]}
    entry = channel.info_json['entries'][0]

    tag = await tag_factory()
    vid1 = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title',
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
        with_info_json=True,
        with_caption_file=True,
        tag_names=[tag.name, ]
    )
    vid2 = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title ',  # Video was renamed, trailing space was removed.
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
        with_info_json=True,
        with_caption_file=True,
    )
    vid1.file_group.url = vid2.file_group.url = 'https://example.com/video'

    assert test_session.query(Video).count() == 2

    test_session.commit()

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    assert test_session.query(Video).count() == 1, 'Duplicate video was not deleted.'
    video = test_session.query(Video).one()
    assert video.video_path == video_path, 'Video path does not match'
    assert video.video_path.is_file()
    assert set(channel.directory.iterdir()) == set(video.file_group.my_paths())
    assert video.file_group.tag_names == [tag.name, ]

    # Running again has no effect.
    assert not await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    assert test_session.query(Video).count() == 1, 'Duplicate video was not deleted.'
    video = test_session.query(Video).one()
    assert video.video_path.is_file()
    assert video.video_path == video_path, 'Video path does not match'
    assert set(channel.directory.iterdir()) == set(video.file_group.my_paths())
    assert video.file_group.tag_names == [tag.name, ]


@pytest.mark.asyncio
async def test_delete_duplicate_video_tagged_duplicate(async_client, test_session, channel_factory, video_factory,
                                                       tag_factory):
    """Deleting a duplicate Video should work even when the duplicate being deleted has tags."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_The video title.mp4'

    channel.info_json = {'entries': [{'id': 'ABC123456', 'title': 'The video title'}]}
    entry = channel.info_json['entries'][0]

    tag = await tag_factory()
    vid1 = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title',
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
        with_info_json=True,
        with_caption_file=True,
    )
    vid2 = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title ',
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
        with_info_json=True,
        with_caption_file=True,
        tag_names=[tag.name, ],  # The duplicate being deleted has a tag.
    )
    vid1.file_group.url = vid2.file_group.url = 'https://example.com/video'

    assert test_session.query(Video).count() == 2

    test_session.commit()

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    assert test_session.query(Video).count() == 1, 'Duplicate video was not deleted.'
    video = test_session.query(Video).one()
    assert video.video_path == video_path
    assert video.video_path.is_file()
    # The tag from the deleted duplicate should be consolidated onto the kept video.
    assert video.file_group.tag_names == [tag.name, ]


@pytest.mark.asyncio
async def test_delete_duplicate_video_leaves_no_tags_directory_orphans(async_client, test_session, test_directory,
                                                                       channel_factory, video_factory, tag_factory,
                                                                       await_switches):
    """Deleting a tagged duplicate Video removes its Tags Directory links; only links to existing files remain."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_The video title.mp4'
    tag = await tag_factory()
    for title in (f'{channel.name}_20000101_ABC123456_The video title',
                  f'{channel.name}_20000101_ABC123456_The video title '):
        video = video_factory(channel_id=channel.id, title=title, upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
                              source_id='ABC123456', with_video_file=True, with_info_json=True, tag_names=[tag.name])
        video.file_group.url = 'https://example.com/video'
    test_session.commit()
    await await_switches()
    tags_directory = test_directory / 'tags'
    assert len([i for i in tags_directory.rglob('*.mp4')]) == 2

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', 'ABC123456', video_path)
    test_session.commit()
    await await_switches()

    links = [i for i in tags_directory.rglob('*') if i.is_file() and i.name != 'README.txt']
    assert links, 'The kept Video is still linked'
    assert all(i.stat().st_nlink > 1 for i in links), 'A link was left behind with no source file'


@pytest.mark.asyncio
async def test_delete_duplicate_video_smaller_copy_at_destination(async_client, test_session, channel_factory,
                                                                  video_factory):
    """The kept Video can be renamed onto the path of the smaller duplicate that was just deleted."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_The video title.mp4'

    channel.info_json = {'entries': [{'id': 'ABC123456', 'title': 'The video title'}]}
    entry = channel.info_json['entries'][0]

    # The smaller copy already occupies the destination path.
    small = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title',
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
    )
    large = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title ',  # Trailing space, so a different path.
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
    )
    small.file_group.url = large.file_group.url = 'https://example.com/video'
    large.file_group.size = small.file_group.size + 1
    large_id = large.id
    large_old_path = large.video_path

    test_session.commit()
    assert small.video_path == video_path

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    # The commit must not raise IntegrityError: the deleted duplicate still held `video_path` when the kept Video's
    # `primary_path` was updated to it.
    test_session.commit()

    video = test_session.query(Video).one()
    assert video.id == large_id, 'The larger Video was not the one kept'
    assert video.video_path == video_path
    assert video.video_path.is_file()
    assert not large_old_path.exists(), 'The kept Video was copied to the destination rather than moved'
    assert set(channel.directory.iterdir()) == set(video.file_group.my_paths())


VIDEO_STREAM = {'codec_type': 'video', 'codec_name': 'h264'}
AUDIO_STREAM = {'codec_type': 'audio', 'codec_name': 'aac'}


@pytest.mark.asyncio
@pytest.mark.parametrize('incomplete_streams', [[VIDEO_STREAM], [AUDIO_STREAM]])
async def test_delete_duplicate_video_keeps_complete_copy(async_client, test_session, channel_factory, video_factory,
                                                          incomplete_streams):
    """A larger copy without audio (or without video), like an unmerged yt-dlp format file, is deleted rather than
    the complete copy."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_The video title.mp4'
    channel.info_json = {'entries': [{'id': 'ABC123456', 'title': 'The video title'}]}
    entry = channel.info_json['entries'][0]

    complete = video_factory(channel_id=channel.id, title=f'{channel.name}_20000101_ABC123456_The video title',
                             source_id='ABC123456', with_video_file=True)
    incomplete = video_factory(channel_id=channel.id, title=f'{channel.name}_20000101_ABC123456_The video title.f313',
                               source_id='ABC123456', with_video_file=True)
    complete.ffprobe_json = {'streams': [VIDEO_STREAM, AUDIO_STREAM]}
    incomplete.ffprobe_json = {'streams': incomplete_streams}
    incomplete.file_group.size = complete.file_group.size * 10
    complete_id, incomplete_path = complete.id, incomplete.video_path
    test_session.commit()

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    video = test_session.query(Video).one()
    assert video.id == complete_id, 'The complete Video was not the one kept'
    assert video.video_path == video_path and video.video_path.is_file()
    assert not incomplete_path.exists()


@pytest.mark.asyncio
async def test_delete_duplicate_video_keeps_viewing_history(async_client, test_session, channel_factory,
                                                            video_factory):
    """The kept Video takes the most recent view, and the resume position of a deleted duplicate when it has none."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_The video title.mp4'
    channel.info_json = {'entries': [{'id': 'ABC123456', 'title': 'The video title'}]}
    entry = channel.info_json['entries'][0]

    kept = video_factory(channel_id=channel.id, title=f'{channel.name}_20000101_ABC123456_The video title',
                         source_id='ABC123456', with_video_file=True)
    watched = video_factory(channel_id=channel.id, title=f'{channel.name}_20000101_ABC123456_The video title ',
                            source_id='ABC123456', with_video_file=True)
    older = video_factory(channel_id=channel.id, title=f'{channel.name}_20000101_ABC123456_The video title  ',
                          source_id='ABC123456', with_video_file=True)
    kept.file_group.size = watched.file_group.size * 10
    watched.file_group.viewed = datetime(2026, 2, 1, tzinfo=timezone.utc)
    watched.file_group.progress = 0.5
    watched.file_group.position = {'kind': 'time', 'seconds': 100.0}
    older.file_group.viewed = datetime(2025, 1, 1, tzinfo=timezone.utc)
    older.file_group.progress = 0.9
    older.file_group.position = {'kind': 'time', 'seconds': 900.0}
    kept_id = kept.id
    test_session.commit()

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    video = test_session.query(Video).one()
    assert video.id == kept_id
    assert video.file_group.viewed == datetime(2026, 2, 1, tzinfo=timezone.utc)
    assert video.file_group.progress == 0.5
    assert video.file_group.position == {'kind': 'time', 'seconds': 100.0}

    # The kept Video's own position is not replaced.
    other = video_factory(channel_id=channel.id, title=f'{channel.name}_20000101_ABC123456_The video title ',
                          source_id='ABC123456', with_video_file=True)
    other.file_group.viewed = datetime(2026, 3, 1, tzinfo=timezone.utc)
    other.file_group.progress = 0.2
    other.file_group.position = {'kind': 'time', 'seconds': 20.0}
    test_session.commit()

    assert await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    video = test_session.query(Video).one()
    assert video.id == kept_id
    assert video.file_group.viewed == datetime(2026, 3, 1, tzinfo=timezone.utc)
    assert video.file_group.progress == 0.5
    assert video.file_group.position == {'kind': 'time', 'seconds': 100.0}


@pytest.mark.asyncio
async def test_delete_renamed_video(async_client, test_session, channel_factory, video_factory, tag_factory):
    """If duplicate Video's exist, delete all the videos that do not have the new title."""
    channel = channel_factory(name='Channel Name')
    video_path = channel.directory / f'{channel.name}_20000101_ABC123456_Some new title.mp4'

    channel.info_json = {'entries': [{'id': 'ABC123456', 'title': 'Some new title'}]}
    entry = channel.info_json['entries'][0]

    tag = await tag_factory()
    vid1 = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title',
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
        with_info_json=True,
        with_caption_file=True,
        tag_names=[tag.name, ]
    )
    vid2 = video_factory(
        channel_id=channel.id,
        title=f'{channel.name}_20000101_ABC123456_The video title ',  # Video was renamed, trailing space was removed.
        upload_date=datetime(2000, 1, 1, tzinfo=timezone.utc),
        source_id='ABC123456',
        with_video_file=True,
        with_info_json=True,
        with_caption_file=True,
    )
    vid1.file_group.url = vid2.file_group.url = 'https://example.com/video'

    assert test_session.query(Video).count() == 2

    test_session.commit()

    await Video.delete_duplicate_videos(test_session, 'https://example.com/video', entry['id'], video_path)
    test_session.commit()

    assert test_session.query(Video).count() == 1, 'Duplicate video was not deleted.'
    video = test_session.query(Video).one()
    assert video.video_path == video_path, 'Video path does not match'
    assert video.video_path.is_file()
    assert set(channel.directory.iterdir()) == set(video.file_group.my_paths())
    assert video.file_group.tag_names == [tag.name, ]