Mechanical, whole-tree `ruff format` pass — no behaviour change. Recorded in .git-blame-ignore-revs (next commit) so git blame skips it.
177 lines
7.7 KiB
Python
177 lines
7.7 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)
|
|
# An un-set caller gets the source-neutral "generic" default, never a false "youtube".
|
|
assert '<uniqueid type="generic"' in _nfo_xml(MediaMeta(**base))
|
|
assert '<uniqueid type="facebook"' in _nfo_xml(MediaMeta(**base, uniqueid_type="facebook"))
|
|
assert '<uniqueid type="youtube"' in _nfo_xml(MediaMeta(**base, uniqueid_type="youtube"))
|
|
|
|
|
|
def test_sidecars_keep_the_full_basename_when_the_title_has_a_dot(tmp_path: Path):
|
|
# A title ending in a period ("Sonka Házilag.") puts a dot in the media slug; the .nfo/.jpg
|
|
# sidecars must sit on the SAME (full) basename, not be truncated at that dot (which orphaned
|
|
# them from the media so Plex/Kodi never matched them).
|
|
from app.downloads.storage import write_sidecars
|
|
|
|
stem = "Chan_-_2026-08-12_-_Sonka_Hazilag._Villam_konyha_abc_a5"
|
|
rel = f"Chan/Season_2026/{stem}.mp4"
|
|
media = tmp_path / rel
|
|
media.parent.mkdir(parents=True)
|
|
media.write_text("x")
|
|
thumb = tmp_path / "t.jpg"
|
|
thumb.write_bytes(b"\xff\xd8\xff")
|
|
meta = MediaMeta(
|
|
video_id="abc", title="Sonka Házilag. | Villám konyha", uploader="Chan", upload_date=None
|
|
)
|
|
|
|
assert write_sidecars(str(tmp_path), rel, meta, thumb) is True
|
|
names = sorted(p.name for p in media.parent.iterdir())
|
|
assert f"{stem}.nfo" in names, names
|
|
assert f"{stem}.jpg" in names, names
|
|
assert "Chan_-_2026-08-12_-_Sonka_Hazilag.nfo" not in names # the old truncated orphan
|