"""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, run_ffprobe_json 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 TestRunFfprobeJson: """The single ffprobe-JSON invocation point (`probe.probe_json`, which the stored facts and the phone-safe rule both read). A failed probe or unparseable output must degrade to None, not raise — the malformed-input safety the old parse_probe_streams string parser used to own now lives here.""" def test_unparseable_output_returns_none(self, monkeypatch): import subprocess class _R: stdout = "not json" monkeypatch.setattr(subprocess, "run", lambda *a, **k: _R()) assert run_ffprobe_json(Path("x")) is None def test_ffprobe_failure_returns_none(self, monkeypatch): import subprocess def _boom(*a, **k): raise subprocess.CalledProcessError(1, "ffprobe") monkeypatch.setattr(subprocess, "run", _boom) assert run_ffprobe_json(Path("x")) is None def test_parses_streams_from_valid_json(self, monkeypatch): import subprocess class _R: stdout = '{"streams":[{"codec_name":"h264","codec_type":"video"}]}' monkeypatch.setattr(subprocess, "run", lambda *a, **k: _R()) data = run_ffprobe_json(Path("x"), entries="stream=codec_type,codec_name") assert data["streams"] == [{"codec_name": "h264", "codec_type": "video"}] 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"