mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
9bbef50ca1
* 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> * Remove Tags Directory links on path deletes; keep a valid poster's link The file browser deletes by path (`lib.delete`), which unlinks the sources before any FileGroup is deleted, so its tagged files still left orphaned links. It now removes the links of the files it is about to delete, or of every FileGroup in a deleted directory. A poster is no longer unlinked from the Tags Directory unless it is actually removed: a valid JPEG that is not named .jpg is kept with its link, and a converted poster's links are removed only once the conversion succeeds. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
629 lines
29 KiB
Python
629 lines
29 KiB
Python
import pathlib
|
|
import shutil
|
|
from http import HTTPStatus
|
|
|
|
import pytest
|
|
|
|
from modules.videos import lib
|
|
from modules.videos.downloader import ChannelDownloader, VideoDownloader
|
|
from modules.videos.lib import parse_video_file_name, validate_video, get_statistics, format_videos_destination
|
|
from modules.videos.models import Video, Channel
|
|
from wrolpi.common import get_wrolpi_config
|
|
from wrolpi.downloader import Download, RSSDownloader
|
|
from wrolpi.files.models import FileGroup
|
|
from wrolpi.vars import PROJECT_DIR
|
|
|
|
|
|
@pytest.mark.parametrize('file_name,expected', [
|
|
('channel_20000101_12345678910_ some title.mp4', ('channel', '20000101', '12345678910', 'some title')),
|
|
('channel name_NA_12345678910_ some title.mp4', ('channel name', None, '12345678910', 'some title')),
|
|
('channel name_NA_12345678910_ some title.MP4', ('channel name', None, '12345678910', 'some title')),
|
|
('20000101 foo.mp4', (None, None, None, '20000101 foo')),
|
|
('20000101foo.mp4', (None, None, None, '20000101foo')),
|
|
('something 20000101 foo.mp4', (None, None, None, 'something 20000101 foo')),
|
|
('something_20000101_foo.mp4', ('something', '20000101', None, 'foo')),
|
|
('foo .mp4', (None, None, None, 'foo')), # Has trailing whitespace.
|
|
('NA_20000303_vp91w5_Bob's Pancakes.mp4', (None, '20000303', 'vp91w5', 'Bob's Pancakes')),
|
|
('NA_20000303_Bob's Pancakes.mp4', (None, '20000303', None, 'Bob's Pancakes')),
|
|
('20000303_Bob's Pancakes.mp4', (None, '20000303', None, 'Bob's Pancakes')),
|
|
('NA_NA_vp91w5_Bob's Pancakes.mp4', (None, None, 'vp91w5', 'Bob's Pancakes')),
|
|
('Learning Self-Reliance_20170529_p_MzsCFkUPU_Beekeeping 2017 Part 6 - Merging Hives.mp4',
|
|
('Learning Self-Reliance', '20170529', 'p_MzsCFkUPU', 'Beekeeping 2017 Part 6 - Merging Hives')),
|
|
('Learning Self-Reliance_20170529_p_Mzs_Beekeeping 2017 Part 6 - Merging Hives.mp4',
|
|
('Learning Self-Reliance', '20170529', 'p_Mzs', 'Beekeeping 2017 Part 6 - Merging Hives')),
|
|
('Learning Self-Reliance_20170529_p_Beekeeping 2017 Part 6 - Merging Hives.mp4',
|
|
('Learning Self-Reliance', '20170529', None, 'p_Beekeeping 2017 Part 6 - Merging Hives')),
|
|
('Learning Self-Reliance_20240722_clyx8ark92qk70860vtl997n5_Long Source Id File with _.mp4',
|
|
('Learning Self-Reliance', '20240722', 'clyx8ark92qk70860vtl997n5', 'Long Source Id File with _')),
|
|
# Audio files and other media containers carry the same name format; a suffix missing from
|
|
# the parser leaves the file with no published date, sorted to the end of its Channel.
|
|
('The Tech Prepper_20260820_WRMI S2 Underground Shortwave Interview.flac',
|
|
('The Tech Prepper', '20260820', None, 'WRMI S2 Underground Shortwave Interview')),
|
|
('channel_20000101_12345678910_ some title.mp3', ('channel', '20000101', '12345678910', 'some title')),
|
|
('channel_20000101_12345678910_ some title.mkv', ('channel', '20000101', '12345678910', 'some title')),
|
|
('channel_20000101_some title.opus', ('channel', '20000101', None, 'some title')),
|
|
('channel_20000101_some title.ogg', ('channel', '20000101', None, 'some title')),
|
|
('channel_20000101_some title.m4a', ('channel', '20000101', None, 'some title')),
|
|
('channel_20000101_some title.m4v', ('channel', '20000101', None, 'some title')),
|
|
('channel_20000101_some title.wav', ('channel', '20000101', None, 'some title')),
|
|
])
|
|
def test_parse_video_file_name(file_name, expected):
|
|
"""
|
|
A Video's title can be parsed from the video file.
|
|
"""
|
|
video_path = pathlib.Path(file_name)
|
|
assert parse_video_file_name(video_path) == expected, f'{file_name} != {expected}'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_video_factory(test_session, video_factory, channel_factory):
|
|
"""
|
|
The `video_factory` pytest fixture is used in many video tests. Test all it's functionality.
|
|
"""
|
|
channel = channel_factory()
|
|
video = video_factory(
|
|
channel_id=channel.id,
|
|
with_video_file=True,
|
|
with_info_json={'description': 'hello'},
|
|
with_caption_file=True,
|
|
with_poster_ext='png',
|
|
source_id='some id',
|
|
)
|
|
test_session.commit()
|
|
|
|
assert video.video_path and video.video_path.is_file()
|
|
assert video.caption_paths and all(i.is_file() for i in video.caption_paths)
|
|
assert video.poster_path and video.poster_path.is_file()
|
|
assert video.info_json_path and video.info_json_path.is_file()
|
|
|
|
assert video.file_group.a_text # title
|
|
assert video.file_group.b_text is None
|
|
assert video.file_group.c_text
|
|
assert video.file_group.d_text # captions
|
|
assert video.source_id == 'some id'
|
|
|
|
|
|
def test_validate_video(test_session, test_directory, video_factory, image_bytes_factory, video_file,
|
|
test_wrolpi_config):
|
|
"""A video poster will be generated only if the channel permits."""
|
|
# Disable ffprobe json file creation to avoid unexpected files in test assertions
|
|
# (Directly modify _config to avoid locking issues in sync tests)
|
|
get_wrolpi_config()._config['save_ffprobe_json'] = False
|
|
|
|
video_file = video_file.rename(test_directory / 'Channel Name_20050607_1234567890_The Title.mp4')
|
|
vid1 = video_factory(with_video_file=video_file, with_info_json=True)
|
|
# Clear source_id so source_id in the file will be extracted.
|
|
vid1.source_id = None
|
|
assert not vid1.poster_path
|
|
|
|
validate_video(test_session, vid1, True)
|
|
assert vid1.video_path == video_file
|
|
assert vid1.poster_path, 'Poster was not created'
|
|
assert vid1.poster_path.is_file(), 'Poster path does not exist'
|
|
assert vid1.video_path.stem == vid1.poster_path.stem
|
|
assert vid1.poster_path.suffix == '.jpg'
|
|
# File name date is assumed to be local timezone.
|
|
assert vid1.file_group.published_datetime and vid1.file_group.published_datetime.year == 2005
|
|
assert vid1.file_group.download_datetime
|
|
assert vid1.source_id == '1234567890'
|
|
assert vid1.file_group.title == 'The Title'
|
|
|
|
# Replace info json, FileGroup attributes should be replaced.
|
|
info_json = {
|
|
'channel_id': 'channel id',
|
|
'channel_url': 'https://example.com/example_video_json_channel',
|
|
'duration': 635,
|
|
'epoch': 123456789,
|
|
'fulltitle': 'The full title',
|
|
'id': 'some long id',
|
|
'title': 'The title',
|
|
'upload_date': '20240917',
|
|
'uploader_id': 'uploader id',
|
|
'uploader_url': 'https://example.com/example_video_json_uploader',
|
|
'view_count': 406
|
|
}
|
|
vid1.replace_info_json(info_json)
|
|
validate_video(test_session, vid1, False)
|
|
assert vid1.file_group.published_datetime and vid1.file_group.published_datetime.year == 2024
|
|
assert vid1.file_group.title == 'The full title'
|
|
|
|
# A PNG is converted, but not deleted.
|
|
vid2 = video_factory(with_video_file=True, with_poster_ext='.png')
|
|
assert vid2.poster_path and vid2.poster_path.suffix == '.png', 'Poster was not initialized'
|
|
assert {i.suffix for i in vid2.file_group.my_paths()} == {'.png', '.mp4'}
|
|
validate_video(test_session, vid2, True)
|
|
assert vid2.poster_path.is_file(), 'New poster was not generated'
|
|
assert vid2.poster_path.suffix == '.jpg' and vid2.poster_path.stat().st_size > 0
|
|
assert {i.suffix for i in vid2.file_group.my_paths()} == {'.jpg', '.png', '.mp4'}
|
|
|
|
|
|
def test_validate_video_directory_beats_info_json_channel(test_session, test_directory, channel_factory,
|
|
video_factory):
|
|
"""A Video in a Channel's directory belongs to that Channel, even when its info json names another
|
|
existing Channel.
|
|
|
|
Replicates a user report: videos placed in Channel B's directory did not appear on Channel B's page
|
|
because their info json named Channel A (which exists in the library), and the source_id match
|
|
was preferred over the directory."""
|
|
channel_a = channel_factory(source_id='channel_a_id', url='https://example.com/channel_a')
|
|
channel_b = channel_factory(directory=test_directory / 'videos/ChannelB')
|
|
|
|
video = video_factory(
|
|
with_video_file=channel_b.directory / 'The Title.mp4',
|
|
with_info_json={
|
|
'channel_id': 'channel_a_id',
|
|
'channel_url': 'https://example.com/channel_a',
|
|
'duration': 5,
|
|
'epoch': 12345678,
|
|
'title': 'The Title',
|
|
},
|
|
)
|
|
test_session.commit()
|
|
|
|
assert video.channel_id == channel_b.id, \
|
|
f'Video in ChannelB directory should belong to ChannelB, got channel_id={video.channel_id}'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_statistics(test_session, video_factory, channel_factory):
|
|
# Can get statistics in empty DB.
|
|
await get_statistics(test_session)
|
|
|
|
channel1 = channel_factory()
|
|
channel2 = channel_factory()
|
|
video_factory(channel_id=channel1.id)
|
|
video_factory(channel_id=channel1.id)
|
|
video_factory(channel_id=channel2.id)
|
|
video_factory()
|
|
|
|
result = await get_statistics(test_session)
|
|
assert 'statistics' in result
|
|
assert 'videos' in result['statistics']
|
|
assert 'channels' in result['statistics']
|
|
assert 'historical' in result['statistics']
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orphaned_files(async_client, test_session, make_files_structure, test_directory, video_factory,
|
|
refresh_files, test_wrolpi_config):
|
|
# Disable ffprobe json file creation to avoid unexpected files in test assertions
|
|
get_wrolpi_config().save_ffprobe_json = False
|
|
|
|
# A Video without associated files is not orphaned.
|
|
vid1 = video_factory(title='vid1', with_video_file=True)
|
|
# The video files will be removed...
|
|
vid2 = video_factory(title='vid2', with_video_file=True, with_caption_file=True, with_poster_ext='jpeg',
|
|
with_info_json=True)
|
|
vid3 = video_factory(title='vid3', with_video_file=True, with_poster_ext='jpg')
|
|
# This will be ignored because it is not in the "videos" subdirectory.
|
|
shutil.copy(PROJECT_DIR / 'test/example1.en.vtt', test_directory / 'vid4.en.vtt')
|
|
test_session.commit()
|
|
|
|
# Remove vid2 video. Caption, poster, info_json are now orphaned.
|
|
vid2.video_path.unlink()
|
|
vid2_caption_paths, vid2_poster_path, vid2_info_json_path = vid2.caption_paths, vid2.poster_path, vid2.info_json_path
|
|
# Remove vid3 video. Poster is now orphaned.
|
|
vid3.video_path.unlink()
|
|
vid3_poster_path = vid3.poster_path
|
|
await refresh_files()
|
|
test_session.commit()
|
|
|
|
# 6 files when two video files are deleted.
|
|
# (vid1, vid2[caption,poster,json], vid3 poster, vid4)
|
|
assert test_session.query(FileGroup).count() == 4
|
|
assert test_session.query(Video).count() == 1
|
|
|
|
videos_directory = test_directory / 'videos'
|
|
results = lib.find_orphaned_video_files(videos_directory)
|
|
|
|
assert sorted(list(results)) == sorted([
|
|
vid2_caption_paths[0],
|
|
vid2_info_json_path,
|
|
vid2_poster_path,
|
|
vid3_poster_path,
|
|
])
|
|
|
|
|
|
def test_link_channel_and_downloads(test_session, channel_factory, test_download_manager):
|
|
channel = channel_factory(
|
|
url='https://www.youtube.com/c/LearningSelfReliance/videos',
|
|
source_id='UCng5u6ASda3LNRXJN0JCQJA',
|
|
)
|
|
download1 = Download(url=channel.get_rss_url(), downloader=RSSDownloader.name)
|
|
download2 = Download(url='https://example.com/videos', downloader=ChannelDownloader.name, frequency=99,
|
|
settings=dict(destination=str(channel.directory)))
|
|
download3 = Download(url='https://example.com/video/1', downloader=RSSDownloader.name,
|
|
settings=dict(destination=str(channel.directory)))
|
|
test_session.add_all([download1, download2, download3])
|
|
test_session.commit()
|
|
assert test_session.query(Download).count() == 3
|
|
assert not any(i.collection_id for i in test_download_manager.get_downloads(test_session))
|
|
|
|
# `link_channel_and_downloads` links Downloads to Collections.
|
|
lib.link_channel_and_downloads(test_session)
|
|
assert test_session.query(Download).count() == 3
|
|
assert all(i.collection_id for i in test_download_manager.get_recurring_downloads(test_session))
|
|
assert not any(i.collection_id for i in test_download_manager.get_once_downloads(test_session))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
def test_link_channel_and_downloads_migration(async_client, test_session, channel_factory, test_download_manager):
|
|
"""Test the 8d0d81bc9c34_channel_channel_downloads.py migration."""
|
|
# A simple Channel which has a download for its URL.
|
|
channel1 = channel_factory(url='https://example.com/channel1')
|
|
download1 = Download(url='https://example.com/channel1', downloader=ChannelDownloader.name, frequency=1)
|
|
# A Channel which has two Downloads. One for it's URL, another which is an RSS feed in its directory.
|
|
channel2 = channel_factory(url='https://example.com/channel2')
|
|
download2a = Download(url='https://example.com/channel2/rss', downloader=RSSDownloader.name, frequency=1,
|
|
settings=dict(destination=str(channel2.directory)))
|
|
download2b = Download(url='https://example.com/channel2', downloader=ChannelDownloader.name, frequency=1,
|
|
settings=dict(destination=str(channel2.directory)))
|
|
# Download does not have a frequency, so it cannot be a channel download.
|
|
download2c = Download(url='https://example.com/channel2/video/1', downloader=VideoDownloader.name)
|
|
# A Channel which has a URL, but no Downloads.
|
|
channel3 = channel_factory()
|
|
test_session.add_all([download1, download2a, download2b, download2c])
|
|
test_session.commit()
|
|
|
|
assert not channel1.downloads
|
|
assert channel2.directory and not channel2.downloads
|
|
assert not channel3.downloads
|
|
|
|
lib.link_channel_and_downloads(test_session)
|
|
|
|
assert channel1.downloads
|
|
assert channel2.downloads
|
|
assert not channel3.downloads
|
|
|
|
assert test_session.query(Channel).count() == 3
|
|
assert test_session.query(Download).count() == 4
|
|
d1, d2a, d2b, d2c = test_session.query(Download).order_by(Download.url).all()
|
|
assert d1.url == channel1.url == 'https://example.com/channel1' and d1.frequency == 1
|
|
assert d1.collection_id == channel1.collection_id
|
|
assert d2a.url == 'https://example.com/channel2' and d2a.frequency == 1
|
|
assert d2a.collection_id == channel2.collection_id
|
|
assert d2b.url == 'https://example.com/channel2/rss' and d2b.frequency == 1
|
|
assert d2b.collection_id == channel2.collection_id
|
|
assert d2c.url == 'https://example.com/channel2/video/1' and d2c.frequency is None
|
|
assert not d2c.collection_id
|
|
|
|
|
|
def test_link_channel_and_downloads_destination_column(test_session, channel_factory, test_download_manager):
|
|
"""Test that link_channel_and_downloads matches by download.destination column, not just settings['destination'].
|
|
|
|
This simulates what happens when downloads are imported from config - the destination column is set directly,
|
|
not via settings['destination'].
|
|
"""
|
|
channel = channel_factory()
|
|
|
|
# Create download with destination COLUMN set (like import_config does)
|
|
# NOT using settings={'destination': ...}
|
|
download = Download(
|
|
url='https://example.com/channel/videos',
|
|
downloader=ChannelDownloader.name,
|
|
frequency=99,
|
|
destination=channel.directory, # Using column directly
|
|
)
|
|
test_session.add(download)
|
|
test_session.commit()
|
|
|
|
assert not download.collection_id, 'Download should not be linked yet'
|
|
|
|
lib.link_channel_and_downloads(test_session)
|
|
|
|
test_session.refresh(download)
|
|
assert download.collection_id == channel.collection_id, \
|
|
'Download should be linked to channel via destination column'
|
|
|
|
|
|
def test_link_channel_and_downloads_by_source_id(test_session, channel_factory, test_download_manager):
|
|
"""Test that link_channel_and_downloads matches by info_json channel_id to Channel.source_id.
|
|
|
|
When ChannelDownloader runs, it populates download.info_json with channel_id. This should match
|
|
against Channel.source_id even if the URLs don't match.
|
|
"""
|
|
channel = channel_factory(source_id='UC123456789')
|
|
|
|
# Create download with info_json containing channel_id (like ChannelDownloader sets)
|
|
# Note: different URL than channel URL, should still match by source_id
|
|
download = Download(
|
|
url='https://example.com/different-channel-url',
|
|
downloader=ChannelDownloader.name,
|
|
frequency=99,
|
|
info_json={'channel_id': 'UC123456789', 'id': 'UC123456789', 'uploader': 'Test'},
|
|
)
|
|
test_session.add(download)
|
|
test_session.commit()
|
|
|
|
assert not download.collection_id, 'Download should not be linked yet'
|
|
|
|
lib.link_channel_and_downloads(test_session)
|
|
|
|
test_session.refresh(download)
|
|
assert download.collection_id == channel.collection_id, \
|
|
'Download should be linked to channel via source_id matching'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_format_videos_destination(async_client, test_directory):
|
|
"""Videos destination is formatted according to the WROLPiConfig."""
|
|
wrolpi_config = get_wrolpi_config()
|
|
# Restore after mutating the global singleton so later tests (especially those that
|
|
# call channel_factory without async_client re-init) do not inherit an invalid template.
|
|
original_videos_destination = wrolpi_config.videos_destination
|
|
|
|
try:
|
|
# Channel directory without a tag.
|
|
wrolpi_config.videos_destination = 'videos/%(channel_tag)s/%(channel_name)s'
|
|
assert format_videos_destination('Simple Channel', None, None) \
|
|
== test_directory / 'videos/Simple Channel'
|
|
|
|
# One tag can be applied to a Channel.
|
|
assert format_videos_destination('Simple Channel', 'one', None) \
|
|
== test_directory / 'videos/one/Simple Channel'
|
|
|
|
# Channel Domain is also supported.
|
|
wrolpi_config.videos_destination = 'videos/%(channel_tag)s/%(channel_domain)s/%(channel_name)s'
|
|
assert format_videos_destination('Simple Channel', 'one', 'https://example.com') \
|
|
== test_directory / 'videos/one/example.com/Simple Channel'
|
|
|
|
# Channel name is not required.
|
|
wrolpi_config.videos_destination = '%(channel_tag)s/%(channel_domain)s'
|
|
assert format_videos_destination('Simple Channel', 'one', 'https://example.com') \
|
|
== test_directory / 'one/example.com'
|
|
|
|
# Invalid `videos_destination` raises an error.
|
|
wrolpi_config.videos_destination = '%(channel_tag)s/%(channel)s'
|
|
with pytest.raises(FileNotFoundError):
|
|
assert format_videos_destination()
|
|
finally:
|
|
wrolpi_config.videos_destination = original_videos_destination
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_videos_downloader_config_api(async_client, test_directory, test_videos_downloader_config):
|
|
# The WROLPi default resolutions.
|
|
config = lib.get_videos_downloader_config()
|
|
assert config.video_resolutions == ['1080p', '720p', '480p', 'maximum']
|
|
|
|
request, response = await async_client.get('/api/config?file_name=videos_downloader.yaml')
|
|
assert response.status == HTTPStatus.OK
|
|
config = response.json['config']
|
|
|
|
# Change the `video_resolutions`
|
|
config['video_resolutions'] = ['720p', 'maximum']
|
|
body = dict(config=config)
|
|
request, response = await async_client.post('/api/config?file_name=videos_downloader.yaml', json=body)
|
|
assert response.status == HTTPStatus.NO_CONTENT
|
|
# Config file was actually changed.
|
|
config = lib.get_videos_downloader_config()
|
|
assert config.video_resolutions == ['720p', 'maximum']
|
|
assert (test_directory / 'config/videos_downloader.yaml').is_file()
|
|
assert config.is_valid()
|
|
contents = config.read_config_file()
|
|
assert contents['video_resolutions'] == ['720p', 'maximum']
|
|
assert '1080p' not in config.get_file().read_text()
|
|
|
|
# Cannot set invalid `file_name_format`.
|
|
config = lib.get_videos_downloader_config().dict()
|
|
config['yt_dlp_options']['file_name_format'] = 'invalid format'
|
|
body = dict(config=config)
|
|
request, response = await async_client.post('/api/config?file_name=videos_downloader.yaml', json=body)
|
|
assert response.status == HTTPStatus.BAD_REQUEST
|
|
assert 'Invalid config' in response.json.get('error')
|
|
|
|
|
|
def test_video_location(test_session, video_factory, channel_factory, async_client):
|
|
# Video location without a channel, it opens the preview.
|
|
video1 = video_factory(title='vid1')
|
|
assert video1.location == f'/videos/{video1.file_group_id}'
|
|
|
|
# Video location with a channel.
|
|
channel = channel_factory(name='ChannelName')
|
|
video2 = video_factory(title='vid2', channel_id=channel.id)
|
|
assert video2.location == f'/videos/{video2.file_group_id}'
|
|
|
|
|
|
def test_format_video_filename_without_upload_date(test_session, video_factory, test_videos_downloader_config, caplog):
|
|
"""Videos without upload_date should not cause KeyError when format uses upload_year."""
|
|
import logging
|
|
from modules.videos.lib import format_video_filename
|
|
|
|
# Create video without upload_date
|
|
video = video_factory(title='Test Video', with_video_file=True, with_info_json={'id': 'abc123'})
|
|
assert video.file_group.published_datetime is None
|
|
|
|
# Use a format that includes upload_year subdirectory
|
|
template = '%(upload_year)s/%(uploader)s_%(upload_date)s_%(id)s_%(title)s.%(ext)s'
|
|
|
|
# Should not raise KeyError and should not log an error
|
|
with caplog.at_level(logging.ERROR):
|
|
result = format_video_filename(video, template)
|
|
|
|
# Should not have logged any errors (fallback was not used)
|
|
assert 'Invalid variable in video file_name_format' not in caplog.text
|
|
|
|
# Should return a properly formatted filename with the template structure
|
|
# (empty year and date components produce a path with subdirectory)
|
|
assert result.endswith('.mp4')
|
|
assert 'Test Video' in result
|
|
# The result should follow the template format, not the fallback format
|
|
# Template result: '{upload_year}/{uploader}_{upload_date}_{id}_{title}.mp4'
|
|
# With empty year/date: '/__Test Video_Test Video.mp4' (has a slash for subdirectory)
|
|
assert '/' in result # Confirms template was used (has subdirectory), not fallback
|
|
|
|
|
|
def test_format_video_filename_uses_channel_name_over_info_json(test_session, video_factory, channel_factory,
|
|
test_videos_downloader_config):
|
|
"""When a video belongs to a channel, format_video_filename should use the WROLPi channel name,
|
|
not the YouTube uploader name from info_json.
|
|
|
|
This ensures that when a user renames their channel, reorganized files use the new name.
|
|
"""
|
|
from modules.videos.lib import format_video_filename
|
|
|
|
# Create channel with WROLPi name
|
|
channel = channel_factory(name='My Custom Channel')
|
|
|
|
# Create video in channel with different YouTube uploader name in info_json
|
|
video = video_factory(
|
|
channel_id=channel.id,
|
|
title='My Video Title',
|
|
with_video_file=True,
|
|
with_info_json={
|
|
'id': 'abc123',
|
|
'uploader': 'Learning Self Reliance', # Different from channel name!
|
|
'upload_date': '20240115',
|
|
}
|
|
)
|
|
|
|
template = '%(uploader)s_%(upload_date)s_%(id)s_%(title)s.%(ext)s'
|
|
result = format_video_filename(video, template)
|
|
|
|
# Should use WROLPi channel name, NOT the YouTube uploader
|
|
assert 'My Custom Channel' in result
|
|
assert 'Learning Self Reliance' not in result
|
|
|
|
|
|
def test_extract_global_inheritable(test_directory, test_videos_downloader_config):
|
|
"""Verify _extract_global_inheritable returns all expected keys with correct mapping."""
|
|
config = lib.get_videos_downloader_config()
|
|
result = lib._extract_global_inheritable(config)
|
|
assert set(result.keys()) == lib.INHERITABLE_VIDEO_SETTINGS
|
|
# merge_output_format maps to video_format
|
|
assert result['video_format'] == config.merge_output_format
|
|
assert result['video_resolutions'] == config.video_resolutions
|
|
assert result['writesubtitles'] == config.writesubtitles
|
|
assert result['continue_dl'] == config.continue_dl
|
|
|
|
|
|
def test_get_effective_video_settings_no_overrides(test_directory, test_videos_downloader_config):
|
|
"""With no download settings, effective settings equal global defaults."""
|
|
config = lib.get_videos_downloader_config()
|
|
effective = lib.get_effective_video_settings()
|
|
assert effective['video_resolutions'] == config.video_resolutions
|
|
assert effective['video_format'] == config.merge_output_format
|
|
assert effective['writesubtitles'] == config.writesubtitles
|
|
|
|
|
|
def test_get_effective_video_settings_partial_override(test_directory, test_videos_downloader_config):
|
|
"""Download settings override only the keys they specify."""
|
|
config = lib.get_videos_downloader_config()
|
|
overrides = {'writesubtitles': False, 'sleep_requests': 2.0}
|
|
effective = lib.get_effective_video_settings(overrides)
|
|
assert effective['writesubtitles'] is False
|
|
assert effective['sleep_requests'] == 2.0
|
|
# Everything else stays at global default
|
|
assert effective['video_resolutions'] == config.video_resolutions
|
|
assert effective['writethumbnail'] == config.writethumbnail
|
|
|
|
|
|
def test_get_effective_video_settings_global_change_propagates(test_directory, test_videos_downloader_config):
|
|
"""Changing global config propagates to downloads without overrides."""
|
|
config = lib.get_videos_downloader_config()
|
|
# Change global
|
|
config.update({**config._config, 'sleep_requests': 5.0})
|
|
# Download with no override should pick up new global
|
|
effective = lib.get_effective_video_settings({})
|
|
assert effective['sleep_requests'] == 5.0
|
|
# Download with override should keep its value
|
|
effective = lib.get_effective_video_settings({'sleep_requests': 1.0})
|
|
assert effective['sleep_requests'] == 1.0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_download_defaults_endpoint(async_client, test_directory, test_videos_downloader_config):
|
|
"""GET /api/videos/download-defaults returns flattened global settings."""
|
|
request, response = await async_client.get('/api/videos/download-defaults')
|
|
assert response.status == HTTPStatus.OK
|
|
data = response.json
|
|
assert 'video_resolutions' in data
|
|
assert 'video_format' in data
|
|
assert 'writesubtitles' in data
|
|
assert 'continue_dl' in data
|
|
assert 'nooverwrites' in data
|
|
assert 'video_codecs' in data
|
|
assert 'audio_codecs' in data
|
|
assert 'transcode' in data
|
|
assert 'strict_codecs' in data
|
|
# Should NOT contain global-only settings
|
|
assert 'file_name_format' not in data
|
|
assert 'quiet' not in data
|
|
|
|
|
|
def test_codec_config_properties(test_directory, test_videos_downloader_config):
|
|
"""The codec settings default to off and round-trip through the config."""
|
|
config = lib.get_videos_downloader_config()
|
|
assert config.video_codecs == []
|
|
assert config.audio_codecs == []
|
|
assert config.transcode is False
|
|
assert config.strict_codecs is False
|
|
|
|
config.video_codecs = ['h264', 'vp9']
|
|
config.audio_codecs = ['aac']
|
|
config.transcode = True
|
|
config.strict_codecs = True
|
|
assert config.video_codecs == ['h264', 'vp9']
|
|
assert config.audio_codecs == ['aac']
|
|
assert config.transcode is True
|
|
assert config.strict_codecs is True
|
|
|
|
effective = lib.get_effective_video_settings()
|
|
assert effective['video_codecs'] == ['h264', 'vp9']
|
|
assert effective['transcode'] is True
|
|
|
|
# A download-level override wins.
|
|
effective = lib.get_effective_video_settings({'video_codecs': ['av1'], 'transcode': False})
|
|
assert effective['video_codecs'] == ['av1']
|
|
assert effective['transcode'] is False
|
|
assert effective['audio_codecs'] == ['aac']
|
|
|
|
|
|
def test_download_settings_codec_validation():
|
|
"""DownloadSettings accepts only known codec names."""
|
|
from wrolpi.schema import DownloadSettings
|
|
from wrolpi.errors import ValidationError
|
|
|
|
settings = DownloadSettings(video_codecs=['h264', 'vp9'], audio_codecs=['aac'],
|
|
transcode=True, strict_codecs=True)
|
|
assert settings.video_codecs == ['h264', 'vp9']
|
|
|
|
with pytest.raises(ValidationError):
|
|
DownloadSettings(video_codecs=['divx'])
|
|
|
|
with pytest.raises(ValidationError):
|
|
DownloadSettings(audio_codecs=['wma'])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_convert_poster_removes_tags_directory_link(async_client, test_session, test_directory, video_factory,
|
|
tag_factory, await_switches):
|
|
"""Converting a tagged Video's poster removes the old poster's Tags Directory link."""
|
|
tag = await tag_factory()
|
|
video = video_factory(with_video_file=True, with_poster_ext='webp', tag_names=[tag.name])
|
|
test_session.commit()
|
|
await await_switches()
|
|
webp_link = test_directory / 'tags' / tag.name / video.poster_path.name
|
|
assert webp_link.is_file()
|
|
|
|
new_poster, _ = lib.convert_or_generate_poster(video)
|
|
|
|
assert new_poster.suffix == '.jpg'
|
|
assert not webp_link.exists(), 'The converted poster left its link behind'
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_poster_keeps_tags_directory_link(async_client, test_session, test_directory, video_factory,
|
|
tag_factory, await_switches):
|
|
"""A valid JPEG poster that is not named .jpg is not converted, so its Tags Directory link is kept."""
|
|
tag = await tag_factory()
|
|
video = video_factory(with_video_file=True, with_poster_ext='jpeg', tag_names=[tag.name])
|
|
test_session.commit()
|
|
await await_switches()
|
|
poster_link = test_directory / 'tags' / tag.name / video.poster_path.name
|
|
assert poster_link.is_file()
|
|
|
|
lib.convert_or_generate_poster(video)
|
|
|
|
assert video.poster_path.is_file()
|
|
assert poster_link.is_file() and poster_link.samefile(video.poster_path)
|