mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
Using stubs to simplify tests.
This commit is contained in:
Roland
parent
63d768bdc3
commit
1951ca548d
4 changed files with 196 additions and 121 deletions
|
|
@ -272,6 +272,91 @@ def extract_info(url: str, ydl: YoutubeDL = YDL, process: bool = False) -> dict:
|
|||
return ydl.extract_info(url, download=False, process=process)
|
||||
|
||||
|
||||
def prepare_video_filename(url: str, out_dir: pathlib.Path, video_resolutions: str, video_format: str,
|
||||
audio_only: bool = False, audio_format: str = 'mp3') \
|
||||
-> Tuple[pathlib.Path, dict]:
|
||||
"""Predict the on-disk path of a video/audio file from its URL using yt-dlp.
|
||||
|
||||
Module-level (rather than a method on VideoDownloader) so tests can patch it once
|
||||
and avoid the descriptor-binding pitfalls of patching `Class.method`. Returns the
|
||||
absolute path the file will land at plus the yt-dlp entry dict for the URL.
|
||||
"""
|
||||
if not out_dir.is_dir():
|
||||
raise ValueError(f'Output directory does not exist! {out_dir=}')
|
||||
|
||||
# YoutubeDL expects specific options, add onto the default options.
|
||||
# Deep copy to avoid mutating the shared singleton config dict.
|
||||
config = get_videos_downloader_config()
|
||||
options = copy.deepcopy(config.yt_dlp_options)
|
||||
|
||||
if audio_only:
|
||||
options['format'] = 'bestaudio'
|
||||
options['postprocessors'] = [{
|
||||
'key': 'FFmpegExtractAudio',
|
||||
'preferredcodec': audio_format,
|
||||
}]
|
||||
# merge_output_format is not used for audio-only.
|
||||
options.pop('merge_output_format', None)
|
||||
else:
|
||||
options['merge_output_format'] = video_format
|
||||
options['format'] = video_resolutions
|
||||
|
||||
# Convert WROLPi-specific variables to yt-dlp syntax
|
||||
converted_format = convert_wrolpi_filename_format(config.file_name_format)
|
||||
# yt-dlp expects the absolute path.
|
||||
options['outtmpl'] = f'{out_dir}/{converted_format}'
|
||||
options['cachdir'] = YTDLP_CACHE_DIR
|
||||
|
||||
output_ext = audio_output_extension(audio_format) if audio_only else video_format
|
||||
|
||||
# Create a new YoutubeDL for the output directory.
|
||||
ydl = YoutubeDL(copy.deepcopy(options))
|
||||
ydl.params['logger'] = ydl_logger
|
||||
ydl.add_default_info_extractors()
|
||||
|
||||
# Get the path where the file will be saved.
|
||||
try:
|
||||
entry = extract_info(url, ydl=ydl, process=True)
|
||||
# If the entry is a playlist (e.g. Twitter/X post with multiple videos),
|
||||
# use the first entry for filename preparation.
|
||||
if entry.get('entries'):
|
||||
entries = list(entry['entries']) if not isinstance(entry['entries'], list) else entry['entries']
|
||||
if entries:
|
||||
entry = entries[0]
|
||||
final_filename = pathlib.Path(prepare_filename(entry, ydl=ydl)).absolute()
|
||||
except DownloadError as e:
|
||||
if ' Cannot write ' in str(e):
|
||||
# yt-dlp does not handle long file names well, get the name from the error (lol)
|
||||
last_line = str(e).splitlines()[-1]
|
||||
full_path = last_line.split(' file ')[-1].strip()
|
||||
if not full_path.startswith('/'):
|
||||
logger.error(f'Failed to extract filename from {last_line}')
|
||||
raise
|
||||
else:
|
||||
raise
|
||||
|
||||
# Split filename from parent.
|
||||
full_path = pathlib.Path(full_path)
|
||||
parent = full_path.parent
|
||||
filename, _ = split_path_stem_and_suffix(full_path.name)
|
||||
|
||||
# Trim long filename, add output suffix, add back into parent directory.
|
||||
filename = escape_file_name(filename)
|
||||
final_filename = parent / f'{filename}.{output_ext}'
|
||||
final_filename = trim_file_name(final_filename)
|
||||
logger.debug(f'File name was too long. Trimmed to: {final_filename.name}')
|
||||
|
||||
# Get entry info json.
|
||||
options['outtmpl'] = str(final_filename)
|
||||
ydl = YoutubeDL(copy.deepcopy(options))
|
||||
ydl.params['logger'] = ydl_logger
|
||||
ydl.add_default_info_extractors()
|
||||
entry = extract_info(url, ydl=ydl, process=True)
|
||||
|
||||
logger.debug(f'Downloading {url} to {repr(str(final_filename))}')
|
||||
return final_filename, entry
|
||||
|
||||
|
||||
def prepare_filename(entry: dict, ydl: YoutubeDL = YDL) -> str:
|
||||
"""Get filename from YoutubeDL. Separated for testing."""
|
||||
dir_name, file_name = os.path.split(ydl.prepare_filename(entry))
|
||||
|
|
@ -822,7 +907,7 @@ class VideoDownloader(Downloader, ABC):
|
|||
video_resolutions = effective['video_resolutions']
|
||||
video_resolutions_str = ','.join(j for i in video_resolutions for j in VIDEO_RESOLUTION_MAP[i])
|
||||
video_format = effective['video_format']
|
||||
video_path, entry = self.prepare_filename(url, out_dir, video_resolutions_str, video_format,
|
||||
video_path, entry = prepare_video_filename(url, out_dir, video_resolutions_str, video_format,
|
||||
audio_only=audio_only, audio_format=audio_format)
|
||||
|
||||
if settings.get('download_metadata_only'):
|
||||
|
|
@ -1083,86 +1168,6 @@ class VideoDownloader(Downloader, ABC):
|
|||
logger.warning('Could not find channel')
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def prepare_filename(url: str, out_dir: pathlib.Path, video_resolutions: str, video_format: str,
|
||||
audio_only: bool = False, audio_format: str = 'mp3') \
|
||||
-> Tuple[pathlib.Path, dict]:
|
||||
"""Get the full path of a video/audio file from its URL using yt-dlp."""
|
||||
if not out_dir.is_dir():
|
||||
raise ValueError(f'Output directory does not exist! {out_dir=}')
|
||||
|
||||
# YoutubeDL expects specific options, add onto the default options.
|
||||
# Deep copy to avoid mutating the shared singleton config dict.
|
||||
config = get_videos_downloader_config()
|
||||
options = copy.deepcopy(config.yt_dlp_options)
|
||||
|
||||
if audio_only:
|
||||
options['format'] = 'bestaudio'
|
||||
options['postprocessors'] = [{
|
||||
'key': 'FFmpegExtractAudio',
|
||||
'preferredcodec': audio_format,
|
||||
}]
|
||||
# merge_output_format is not used for audio-only.
|
||||
options.pop('merge_output_format', None)
|
||||
else:
|
||||
options['merge_output_format'] = video_format
|
||||
options['format'] = video_resolutions
|
||||
|
||||
# Convert WROLPi-specific variables to yt-dlp syntax
|
||||
converted_format = convert_wrolpi_filename_format(config.file_name_format)
|
||||
# yt-dlp expects the absolute path.
|
||||
options['outtmpl'] = f'{out_dir}/{converted_format}'
|
||||
options['cachdir'] = YTDLP_CACHE_DIR
|
||||
|
||||
output_ext = audio_output_extension(audio_format) if audio_only else video_format
|
||||
|
||||
# Create a new YoutubeDL for the output directory.
|
||||
ydl = YoutubeDL(copy.deepcopy(options))
|
||||
ydl.params['logger'] = ydl_logger
|
||||
ydl.add_default_info_extractors()
|
||||
|
||||
# Get the path where the file will be saved.
|
||||
try:
|
||||
entry = extract_info(url, ydl=ydl, process=True)
|
||||
# If the entry is a playlist (e.g. Twitter/X post with multiple videos),
|
||||
# use the first entry for filename preparation.
|
||||
if entry.get('entries'):
|
||||
entries = list(entry['entries']) if not isinstance(entry['entries'], list) else entry['entries']
|
||||
if entries:
|
||||
entry = entries[0]
|
||||
final_filename = pathlib.Path(prepare_filename(entry, ydl=ydl)).absolute()
|
||||
except DownloadError as e:
|
||||
if ' Cannot write ' in str(e):
|
||||
# yt-dlp does not handle long file names well, get the name from the error (lol)
|
||||
last_line = str(e).splitlines()[-1]
|
||||
full_path = last_line.split(' file ')[-1].strip()
|
||||
if not full_path.startswith('/'):
|
||||
logger.error(f'Failed to extract filename from {last_line}')
|
||||
raise
|
||||
else:
|
||||
raise
|
||||
|
||||
# Split filename from parent.
|
||||
full_path = pathlib.Path(full_path)
|
||||
parent = full_path.parent
|
||||
filename, _ = split_path_stem_and_suffix(full_path.name)
|
||||
|
||||
# Trim long filename, add output suffix, add back into parent directory.
|
||||
filename = escape_file_name(filename)
|
||||
final_filename = parent / f'{filename}.{output_ext}'
|
||||
final_filename = trim_file_name(final_filename)
|
||||
logger.debug(f'File name was too long. Trimmed to: {final_filename.name}')
|
||||
|
||||
# Get entry info json.
|
||||
options['outtmpl'] = str(final_filename)
|
||||
ydl = YoutubeDL(copy.deepcopy(options))
|
||||
ydl.params['logger'] = ydl_logger
|
||||
ydl.add_default_info_extractors()
|
||||
entry = extract_info(url, ydl=ydl, process=True)
|
||||
|
||||
logger.debug(f'Downloading {url} to {repr(str(final_filename))}')
|
||||
return final_filename, entry
|
||||
|
||||
@staticmethod
|
||||
async def download_info_json(download: Download, video_path: pathlib.Path) -> DownloadResult:
|
||||
"""User has requested to refresh Video metadata. Download only metadata files to temporary directory.
|
||||
|
|
|
|||
|
|
@ -98,7 +98,7 @@ async def test_download_no_channel(test_session, video_download_manager, test_di
|
|||
|
||||
url = 'https://www.youtube.com/watch?v=31jPEBiAC3c'
|
||||
mock_video_extract_info.return_value = info_json
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
mock_prepare_filename.return_value = (video_path, {'id': 'foo'})
|
||||
video_download_manager.create_download(test_session, url, video_downloader.name)
|
||||
await video_download_manager.wait_for_all_downloads()
|
||||
|
|
@ -118,7 +118,7 @@ async def test_download_video_tags(test_session, video_download_manager, video_f
|
|||
|
||||
url = 'https://www.youtube.com/watch?v=31jPEBiAC3c'
|
||||
mock_video_extract_info.return_value = copy(example_video_json)
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
mock_prepare_filename.return_value = (video_path, {'id': 'foo'})
|
||||
settings = dict(tag_names=[tag1.name, tag2.name])
|
||||
video_download_manager.create_download(test_session, url, video_downloader.name, settings=settings)
|
||||
|
|
@ -512,7 +512,7 @@ async def test_video_download(test_session, test_directory, mock_video_extract_i
|
|||
part_path.touch()
|
||||
|
||||
url = 'https://www.youtube.com/watch?v=31jPEBiAC3c'
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
mock_video_extract_info.return_value = example_video_json
|
||||
mock_prepare_filename.return_value = (video_path, {'id': 'foo'})
|
||||
|
||||
|
|
@ -549,7 +549,7 @@ async def test_download_result(test_session, test_directory, video_download_mana
|
|||
image_file.rename(video_file.with_suffix('.jpg'))
|
||||
|
||||
mock_video_extract_info.return_value = example_video_json
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
mock_prepare_filename.return_value = [video_file, {'id': 'foo'}]
|
||||
video_download_manager.create_download(test_session, 'https://example.com', video_downloader.name)
|
||||
await video_download_manager.wait_for_all_downloads()
|
||||
|
|
@ -670,13 +670,12 @@ async def test_download_playlist(test_session, test_directory, mock_video_extrac
|
|||
|
||||
mock_video_extract_info.return_value = example_playlist_json # Playlist info is fetched first.
|
||||
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download:
|
||||
mock_video_do_download.return_value = DownloadResult(success=True) # Don't download the videos.
|
||||
# Drive the phase-split API directly: prepare → execute → finalize.
|
||||
from modules.archive.conftest import make_test_ctx
|
||||
prepared = channel_downloader.prepare_download(test_session, download)
|
||||
executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download)
|
||||
result = channel_downloader.finalize_download(test_session, download, executed)
|
||||
# Drive the phase-split API directly: prepare → execute → finalize. No dispatch
|
||||
# loop, so child video downloads are never spawned here — no need to stub them.
|
||||
from modules.archive.conftest import make_test_ctx
|
||||
prepared = channel_downloader.prepare_download(test_session, download)
|
||||
executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download)
|
||||
result = channel_downloader.finalize_download(test_session, download, executed)
|
||||
assert result.success is True, 'Download was not successful'
|
||||
assert set(result.downloads) == {
|
||||
'https://www.youtube.com/watch?v=video_2_url', # Shorts is converted to regular URL.
|
||||
|
|
@ -904,11 +903,12 @@ async def test_channel_downloader_sets_collection_id(test_session, channel_facto
|
|||
}
|
||||
mock_video_extract_info.return_value = mock_info
|
||||
|
||||
# Mock Channel.refresh_files and file_worker.wait_for_job to avoid background tasks
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download, \
|
||||
mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \
|
||||
# Drive the phase-split API directly — no dispatch loop fires, so child video
|
||||
# downloads are never spawned and don't need stubbing. refresh_files and
|
||||
# file_worker.wait_for_job are still patched because ChannelDownloader's
|
||||
# prepare_channel_for_downloads calls them inline.
|
||||
with mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \
|
||||
mock.patch('modules.videos.downloader.file_worker.wait_for_job', new_callable=mock.AsyncMock):
|
||||
mock_video_do_download.return_value = DownloadResult(success=True)
|
||||
from modules.archive.conftest import make_test_ctx
|
||||
prepared = channel_downloader.prepare_download(test_session, download)
|
||||
executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download)
|
||||
|
|
@ -956,10 +956,9 @@ async def test_channel_passes_all_inheritable_settings_to_video(test_session, ch
|
|||
}
|
||||
mock_video_extract_info.return_value = mock_info
|
||||
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download, \
|
||||
mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \
|
||||
# Direct phase-split call — child video downloads are never dispatched here.
|
||||
with mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \
|
||||
mock.patch('modules.videos.downloader.file_worker.wait_for_job', new_callable=mock.AsyncMock):
|
||||
mock_video_do_download.return_value = DownloadResult(success=True)
|
||||
from modules.archive.conftest import make_test_ctx
|
||||
prepared = channel_downloader.prepare_download(test_session, download)
|
||||
executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download)
|
||||
|
|
@ -1003,10 +1002,9 @@ async def test_channel_does_not_pass_continue_dl_nooverwrites(test_session, chan
|
|||
}
|
||||
mock_video_extract_info.return_value = mock_info
|
||||
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.execute_download') as mock_video_do_download, \
|
||||
mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \
|
||||
# Direct phase-split call — child video downloads are never dispatched here.
|
||||
with mock.patch.object(Channel, 'refresh_files', return_value='test-job-id'), \
|
||||
mock.patch('modules.videos.downloader.file_worker.wait_for_job', new_callable=mock.AsyncMock):
|
||||
mock_video_do_download.return_value = DownloadResult(success=True)
|
||||
from modules.archive.conftest import make_test_ctx
|
||||
prepared = channel_downloader.prepare_download(test_session, download)
|
||||
executed = await channel_downloader.execute_download(prepared, make_test_ctx(), download=download)
|
||||
|
|
@ -1035,7 +1033,7 @@ async def test_video_download_uses_effective_settings(test_session, test_directo
|
|||
# Download with writesubtitles=False override (global default is True)
|
||||
settings = {'writesubtitles': False, 'sleep_requests': 5.0}
|
||||
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
mock_video_extract_info.return_value = example_video_json
|
||||
mock_prepare_filename.return_value = (video_path, {'id': 'foo'})
|
||||
|
||||
|
|
@ -1073,7 +1071,7 @@ async def test_live_video_retry_seconds(test_session, test_directory, mock_video
|
|||
url = 'https://www.youtube.com/watch?v=live_video'
|
||||
mock_video_extract_info.return_value = example_video_json
|
||||
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
# Return a path that won't exist (video wasn't downloaded because it's live).
|
||||
mock_prepare_filename.return_value = (simple_channel.directory / 'live video.mp4', {'id': 'foo'})
|
||||
|
||||
|
|
@ -1134,7 +1132,7 @@ async def test_video_download_playlist_appended_extension(test_session, test_dir
|
|||
url = 'https://x.com/example/status/123'
|
||||
mock_video_extract_info.return_value = example_video_json
|
||||
|
||||
with mock.patch('modules.videos.downloader.VideoDownloader.prepare_filename') as mock_prepare_filename:
|
||||
with mock.patch('modules.videos.downloader.prepare_video_filename') as mock_prepare_filename:
|
||||
# prepare_filename returns the path without the real extension (ending in .NA).
|
||||
bogus_path = video_path_na.parent / video_path_na.stem # Removes .mp4, leaving .NA as "extension"
|
||||
mock_prepare_filename.return_value = (bogus_path, {'id': 'foo'})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue