Consolidates source-specific policy (youtube / facebook / reddit / generic) that was scattered across worker.py, backfill.py, service.py, formats.py and storage.py with no shared source identity. app/downloads/adapters.py holds the base (= generic yt-dlp behaviour) + per-source overrides; resolve_adapter(source_kind, source_ref, info) picks one by source_kind, the yt-dlp extractor_key, or the URL host, falling back to generic. yt-dlp stays the engine — the adapter is policy only. Moved into adapters (behaviour-equivalent): - uploader_url derivation (YouTube channel_url / Facebook uploader_id / reddit None) — removes the worker/backfill duplication (build_uploader_url + _should_refetch gone). - the player-client fallback sets + retriable markers + YouTube extractor_args/POT (build_ydl_opts now takes source-agnostic opts plus an `extra` dict from the adapter). - source_url reconstruction (service.source_url delegates to the adapter). Two intentional (approved) behaviour changes: - the player-client fallback is now YouTube-only; non-YouTube sources make a single extraction attempt instead of a pointless double attempt with youtube-only client args. - the .nfo <uniqueid type=...> follows the source (adapter.nfo_uniqueid_type) instead of the hardcoded "youtube" that mislabelled every url-source download. Deliberately left generic (documented non-moves): title cleaning (normalize_title is harmless across sources), the classify marker table, thumbnail_to_jpg, the format selectors, and og.py's YouTube-hqdefault og:image preference (a self-contained thumbnail heuristic on the share path). Verified end-to-end on localdev across all four adapters: downloads succeed and uploader_url + the .nfo uniqueid type are correct per source.
150 lines
6.5 KiB
Python
150 lines
6.5 KiB
Python
"""On-disk layout helpers: name sanitation, the Plex/flat path shapes, and — most importantly —
|
|
the safe_abs_path traversal guard that every file-serving endpoint depends on."""
|
|
from datetime import date
|
|
|
|
from app.downloads.storage import (
|
|
MediaMeta,
|
|
download_filename,
|
|
rel_path,
|
|
safe_abs_path,
|
|
sanitize,
|
|
)
|
|
from pathlib import Path
|
|
|
|
|
|
class TestSanitize:
|
|
def test_spaces_become_underscores_and_result_has_no_separators(self):
|
|
out = sanitize("My Great Video")
|
|
assert out == "My_Great_Video"
|
|
assert "/" not in out and "\\" not in out and " " not in out
|
|
|
|
def test_illegal_and_emoji_chars_are_removed(self):
|
|
out = sanitize('a/b:c*?"<>| 🔥')
|
|
for bad in '/\\:*?"<>|🔥':
|
|
assert bad not in out
|
|
|
|
def test_unsafe_punctuation_is_stripped_to_a_single_separator(self):
|
|
# Shell-unpleasant punctuation (! etc.) is removed; runs of separators collapse to one "_".
|
|
assert sanitize("wow!!!___---end") == "wow_end"
|
|
assert "!" not in sanitize("Rovinjból! A vihar")
|
|
|
|
def test_length_is_capped(self):
|
|
assert len(sanitize("x" * 500, limit=10)) <= 10
|
|
|
|
def test_cap_stays_within_the_byte_limit(self):
|
|
# Names are folded to ASCII, so a char cap is also the byte cap — accented input can't
|
|
# overflow the 255-BYTE FS component limit.
|
|
out = sanitize("á" * 500, limit=10)
|
|
assert out.isascii() and len(out.encode("utf-8")) <= 10
|
|
|
|
def test_empty_or_all_junk_falls_back_to_untitled(self):
|
|
assert sanitize("") == "untitled"
|
|
assert sanitize("🔥🔥") == "untitled"
|
|
|
|
|
|
class TestRelPath:
|
|
meta = MediaMeta(
|
|
video_id="abc123",
|
|
title="Cool Clip",
|
|
uploader="Some Channel",
|
|
upload_date=date(2024, 5, 2),
|
|
)
|
|
|
|
def test_flat_layout_is_a_single_file(self):
|
|
assert rel_path(self.meta, "mp4", layout="flat") == "Cool_Clip_abc123.mp4"
|
|
|
|
def test_plex_layout_nests_channel_and_season(self):
|
|
out = rel_path(self.meta, "mp4", layout="plex")
|
|
assert out == "Some_Channel/Season_2024/Some_Channel_-_2024-05-02_-_Cool_Clip_abc123.mp4"
|
|
|
|
def test_missing_date_uses_year_zero_and_placeholder_date(self):
|
|
meta = MediaMeta(video_id="x", title="t", uploader="c", upload_date=None)
|
|
out = rel_path(meta, "mp4")
|
|
assert "Season_0/" in out and "0000-00-00" in out
|
|
|
|
def test_asset_id_disambiguates_two_formats_of_the_same_video(self):
|
|
# Same video, two assets (e.g. 1080p vs 720p) must NOT collide on one path.
|
|
a = rel_path(self.meta, "mp4", asset_id=5)
|
|
b = rel_path(self.meta, "mp4", asset_id=6)
|
|
assert a != b
|
|
assert a.endswith("_abc123_a5.mp4") and b.endswith("_abc123_a6.mp4")
|
|
# Omitting it keeps the plain name (unchanged for callers/tests that don't pass one).
|
|
assert rel_path(self.meta, "mp4", layout="flat") == "Cool_Clip_abc123.mp4"
|
|
assert rel_path(self.meta, "mp4", layout="flat", asset_id=7) == "Cool_Clip_abc123_a7.mp4"
|
|
|
|
def test_every_component_stays_within_the_255_byte_fs_limit(self):
|
|
# A long accented channel AND title, plus the asset-id suffix, must not overflow any path
|
|
# component (the byte cap in sanitize is what guarantees this).
|
|
meta = MediaMeta(
|
|
video_id="abcdefghijk",
|
|
title="Á" * 300,
|
|
uploader="Ö" * 300,
|
|
upload_date=date(2024, 5, 2),
|
|
)
|
|
out = rel_path(meta, "webm", asset_id=999999)
|
|
for component in out.split("/"):
|
|
assert len(component.encode("utf-8")) <= 255, component
|
|
|
|
|
|
class TestDownloadFilename:
|
|
def test_appends_the_container_extension(self):
|
|
# Served names follow the same ASCII-slug policy as on-disk names: spaces → underscores.
|
|
assert download_filename("My Video", "mp4", Path("/x/f.webm")) == "My_Video.mp4"
|
|
|
|
def test_does_not_double_an_extension_already_present(self):
|
|
assert download_filename("clip.mp4", "mp4", Path("/x/f.mp4")) == "clip.mp4"
|
|
|
|
def test_falls_back_to_the_file_suffix_when_no_container(self):
|
|
assert download_filename("song", None, Path("/x/f.m4a")) == "song.m4a"
|
|
|
|
|
|
class TestSafeAbsPath:
|
|
"""The path-traversal guard: only paths that resolve INSIDE download_root AND exist are returned."""
|
|
|
|
def test_returns_the_path_for_a_file_inside_the_root(self, tmp_path: Path):
|
|
(tmp_path / "sub").mkdir()
|
|
f = tmp_path / "sub" / "video.mp4"
|
|
f.write_text("data")
|
|
got = safe_abs_path(str(tmp_path), "sub/video.mp4")
|
|
assert got is not None and got == f.resolve()
|
|
|
|
def test_rejects_dot_dot_traversal_out_of_the_root(self, tmp_path: Path):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
secret = tmp_path / "secret.txt"
|
|
secret.write_text("nope")
|
|
# ../secret.txt escapes root even though the file exists — must be refused.
|
|
assert safe_abs_path(str(root), "../secret.txt") is None
|
|
|
|
def test_rejects_an_absolute_path_escape(self, tmp_path: Path):
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
outside = tmp_path / "outside.txt"
|
|
outside.write_text("nope")
|
|
assert safe_abs_path(str(root), str(outside)) is None
|
|
|
|
def test_rejects_a_sibling_dir_whose_name_prefixes_the_root(self, tmp_path: Path):
|
|
# The one input that distinguishes the correct `root not in path.parents` guard from the
|
|
# classic buggy `str(path).startswith(str(root))`: /root vs a real sibling /rootx. A file in
|
|
# rootx is OUTSIDE root, but its path DOES start with the root string — a prefix check would
|
|
# wrongly serve it. This pins the guard against that regression.
|
|
root = tmp_path / "root"
|
|
root.mkdir()
|
|
sibling = tmp_path / "rootx"
|
|
sibling.mkdir()
|
|
(sibling / "steal.mp4").write_text("nope")
|
|
assert safe_abs_path(str(root), "../rootx/steal.mp4") is None
|
|
|
|
def test_returns_none_for_a_nonexistent_file_inside_the_root(self, tmp_path: Path):
|
|
assert safe_abs_path(str(tmp_path), "does/not/exist.mp4") is None
|
|
|
|
|
|
def test_nfo_uniqueid_type_reflects_the_source():
|
|
# The .nfo <uniqueid type=...> now follows the source (via SourceAdapter.nfo_uniqueid_type),
|
|
# instead of the old hardcoded "youtube" that mislabelled every url-source download.
|
|
from app.downloads.storage import _nfo_xml
|
|
|
|
base = dict(video_id="v1", title="t", uploader="c", upload_date=None)
|
|
assert '<uniqueid type="youtube"' in _nfo_xml(MediaMeta(**base)) # default preserved
|
|
assert '<uniqueid type="facebook"' in _nfo_xml(MediaMeta(**base, uniqueid_type="facebook"))
|