peter
2f46947dcf
- pick the produced file safely: Path("") is Path("."), which passes .exists(),
so a missing yt-dlp filepath skipped the staging fallback and crashed in
with_name ("PosixPath('.') has an empty name") on non-YouTube sources
- unwrap a playlist/multi_video extraction to the entry that carries the media
- run_ffmpeg: drain stderr while it runs (a full 64KB pipe blocked ffmpeg
forever and hung the worker thread), 1s cancel polling, stall watchdog and a
bounded wait on exit
- escape apostrophes/backslashes in the ffmpeg concat list (pre-0.51.0 names)
- sweep orphaned HLS segment dirs at startup (nothing else ever reaped them)
77 lines
3.5 KiB
Python
77 lines
3.5 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
|
|
from app.worker 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"
|
|
|
|
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_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 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_is_doubled(self):
|
|
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"
|