rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 02:48:38 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5
wrolpi/modules/videos/test/test_video.py
Roland a20ef4893a
Remove a file's Tags Directory links when WROLPi deletes it 
When WROLPi deleted a tagged file (a forced delete, duplicate Video cleanup, or a poster conversion), the
source was unlinked but its Tags Directory hardlink was not.  The link became the only copy, so the tags sync
refused to delete it and warned on every sync.  The links of a file are now removed before WROLPi removes it,
and only if the link is the same file as its source.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-23 12:50:11 -06:00

445 lines
20 KiB
Python
Raw Blame History

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, ]
Reference in a new issue View git blame Copy permalink