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, ]