peter
b2013a079a
worker.py (~1090 L) mixed the PARENT supervisor (claim + supervise + the process
loop) with the CHILD pipeline (the yt-dlp/ffmpeg download+edit that runs in the
spawned subprocess). Extract them:
worker.py 1090 -> 475 parent: claim/supervise/loop
downloads/pipeline.py (new 592) the child pipeline (_download/_process_edit +
hooks/progress/ffmpeg helpers + _Aborted)
downloads/jobdb.py (new 73) the short-lived-session helpers both sides
share (_set_job/_job_status/_touch_job/
_requeue_if_running/_reset_asset_pending +
the staging-dir paths)
The graph is one-directional (runner -> pipeline -> jobdb; worker -> jobdb; no
worker<->pipeline edge) — the parent never calls the child directly, it spawns
runner.py as a subprocess. Consumers rewired: runner import, and two tests follow
the moved symbols (_media_info/_produced_file -> pipeline; the ffmpeg-callback
patches target pipeline where the names are looked up).
Pure code motion, verbatim bodies. ruff clean, full lane 363 pass (unchanged).
140 lines
6.6 KiB
Python
140 lines
6.6 KiB
Python
"""Picking the file yt-dlp actually produced, for sources that aren't YouTube.
|
|
|
|
The regression these cover: `Path(info["filepath"] or "")` is `Path(".")`, which PASSES `.exists()`,
|
|
so a missing filepath (an arbitrary-URL extractor) skipped the staging fallback and crashed later
|
|
with "PosixPath('.') has an empty name" instead of downloading."""
|
|
import pytest
|
|
|
|
from app.downloads.edit import _concat_quote, parse_probe_streams
|
|
from app.downloads.pipeline import _media_info, _produced_file
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
|
|
def _staging(tmp_path: Path, *names: str) -> Path:
|
|
for n in names:
|
|
p = tmp_path / n
|
|
p.write_bytes(b"x" * (10 * (names.index(n) + 1)))
|
|
return tmp_path
|
|
|
|
|
|
class TestProducedFile:
|
|
def test_uses_the_reported_filepath(self, tmp_path):
|
|
staging = _staging(tmp_path, "v.mp4", "v.jpg")
|
|
info = {"requested_downloads": [{"filepath": str(tmp_path / "v.mp4")}]}
|
|
assert _produced_file(info, staging) == tmp_path / "v.mp4"
|
|
|
|
@pytest.mark.parametrize("info", [{}, {"requested_downloads": [{}]}, {"requested_downloads": [{"filepath": ""}]}])
|
|
def test_missing_filepath_falls_back_to_staging(self, tmp_path, info):
|
|
staging = _staging(tmp_path, "thumb.jpg", "clip.mp4")
|
|
assert _produced_file(info, staging) == tmp_path / "clip.mp4"
|
|
|
|
def test_stale_filepath_falls_back_to_staging(self, tmp_path):
|
|
staging = _staging(tmp_path, "clip.mp4")
|
|
info = {"requested_downloads": [{"filepath": str(tmp_path / "gone.mp4")}]}
|
|
assert _produced_file(info, staging) == tmp_path / "clip.mp4"
|
|
|
|
def test_picks_the_largest_media_file_ignoring_thumbs_and_partials(self, tmp_path):
|
|
staging = _staging(tmp_path, "small.mp4", "big.mp4", "cover.webp")
|
|
(tmp_path / "huge.mp4.part").write_bytes(b"x" * 999)
|
|
assert _produced_file({}, staging) == tmp_path / "big.mp4"
|
|
|
|
@pytest.mark.parametrize(
|
|
"leftover",
|
|
["vid.f137.mp4", "vid.mp4.part-Frag12", "vid.ytdl", "vid.temp.mp4", "vid.mp4.part"],
|
|
)
|
|
def test_ytdlp_working_files_never_win(self, tmp_path, leftover):
|
|
# Each of these can legitimately be the BIGGEST file in staging (a video-only per-format
|
|
# stream especially) — storing one would serve a silent or truncated "download".
|
|
(tmp_path / leftover).write_bytes(b"x" * 5000)
|
|
(tmp_path / "vid.mp4").write_bytes(b"x" * 10)
|
|
assert _produced_file({}, tmp_path) == tmp_path / "vid.mp4"
|
|
|
|
def test_only_leftovers_is_treated_as_no_output(self, tmp_path):
|
|
(tmp_path / "vid.f251.webm").write_bytes(b"x" * 5000)
|
|
with pytest.raises(RuntimeError):
|
|
_produced_file({}, tmp_path)
|
|
|
|
def test_no_output_at_all_raises(self, tmp_path):
|
|
staging = _staging(tmp_path, "only.jpg")
|
|
with pytest.raises(RuntimeError):
|
|
_produced_file({}, staging)
|
|
|
|
|
|
class TestMediaInfo:
|
|
def test_playlist_wrapper_unwraps_to_the_entry_that_downloaded(self):
|
|
info = {"id": "page", "title": "Article", "entries": [{"id": "vid", "requested_downloads": [{"filepath": "/x.mp4"}]}]}
|
|
assert _media_info(info)["id"] == "vid"
|
|
|
|
def test_plain_result_is_untouched(self):
|
|
info = {"id": "vid", "requested_downloads": [{"filepath": "/x.mp4"}]}
|
|
assert _media_info(info) is info
|
|
|
|
def test_picks_the_entry_that_actually_downloaded(self):
|
|
# A page whose first entry was skipped: taking entries[0] would label the produced file
|
|
# with the teaser's id/title.
|
|
info = {
|
|
"id": "page",
|
|
"entries": [
|
|
{"id": "teaser", "title": "Teaser"},
|
|
{"id": "main", "title": "Main", "requested_downloads": [{"filepath": "/x.mp4"}]},
|
|
],
|
|
}
|
|
assert _media_info(info)["id"] == "main"
|
|
|
|
def test_falls_back_to_the_first_entry_when_none_downloaded(self):
|
|
info = {"id": "page", "entries": [{"id": "a"}, {"id": "b"}]}
|
|
assert _media_info(info)["id"] == "a"
|
|
|
|
def test_wrapper_that_downloaded_itself_wins_over_its_entries(self):
|
|
info = {"id": "vid", "requested_downloads": [{"filepath": "/x.mp4"}], "entries": [{"id": "other"}]}
|
|
assert _media_info(info)["id"] == "vid"
|
|
|
|
def test_empty_entries_are_ignored(self):
|
|
info = {"id": "vid", "entries": [None]}
|
|
assert _media_info(info)["id"] == "vid"
|
|
|
|
|
|
class TestParseProbeStreams:
|
|
# ffprobe emits codec_name BEFORE codec_type, so the old line parser shifted every name onto
|
|
# the previous stream and an H.264+AAC mp4 came back as (aac, None) → a needless re-encode.
|
|
def test_reads_each_stream_by_its_own_type(self):
|
|
payload = '{"streams":[{"codec_name":"h264","codec_type":"video"},{"codec_name":"aac","codec_type":"audio"}]}'
|
|
assert parse_probe_streams(payload) == ("h264", "aac")
|
|
|
|
def test_audio_first_ordering_is_handled(self):
|
|
payload = '{"streams":[{"codec_name":"AAC","codec_type":"audio"},{"codec_name":"H264","codec_type":"video"}]}'
|
|
assert parse_probe_streams(payload) == ("h264", "aac")
|
|
|
|
def test_extra_streams_do_not_win_over_the_first_of_their_kind(self):
|
|
payload = (
|
|
'{"streams":[{"codec_name":"h264","codec_type":"video"},'
|
|
'{"codec_name":"vp9","codec_type":"video"},'
|
|
'{"codec_name":"mov_text","codec_type":"subtitle"},'
|
|
'{"codec_name":"opus","codec_type":"audio"}]}'
|
|
)
|
|
assert parse_probe_streams(payload) == ("h264", "opus")
|
|
|
|
def test_audio_only_and_video_only(self):
|
|
assert parse_probe_streams('{"streams":[{"codec_name":"opus","codec_type":"audio"}]}') == (None, "opus")
|
|
assert parse_probe_streams('{"streams":[{"codec_name":"av1","codec_type":"video"}]}') == ("av1", None)
|
|
|
|
def test_garbage_is_not_a_crash(self):
|
|
assert parse_probe_streams("") == (None, None)
|
|
assert parse_probe_streams("not json") == (None, None)
|
|
|
|
|
|
class TestConcatQuote:
|
|
# PurePosixPath so the escaping is asserted identically on a Windows dev box and in the
|
|
# Linux test container (a WindowsPath would eat the backslash as a separator).
|
|
def test_apostrophe_is_closed_escaped_and_reopened(self):
|
|
# A pre-0.51.0 file can still be named with an apostrophe; unescaped it breaks the list.
|
|
assert _concat_quote(PurePosixPath("/m/Don't_Look.mp4")) == "/m/Don'\\''t_Look.mp4"
|
|
|
|
def test_backslash_stays_literal(self):
|
|
# Inside single quotes ffmpeg treats `\` literally — doubling it would point at a
|
|
# different (non-existent) path.
|
|
assert _concat_quote(PurePosixPath("/m/a\\b.mp4")) == "/m/a\\b.mp4"
|
|
|
|
def test_ordinary_path_is_unchanged(self):
|
|
assert _concat_quote(PurePosixPath("/m/clip.mp4")) == "/m/clip.mp4"
|