siftlode/backend/tests/test_storage.py

"""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,
    files_root_problem,
    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_no_extension_is_served_without_a_trailing_dot(self):
        assert download_filename("Makefile", None, Path("/x/upload-7")) == "Makefile"

    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 files_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


def _mounts(*points):
    """A /proc/self/mountinfo table with these mount points (the fields around them are filler)."""
    return "\n".join(f"{i} 1 0:{i} / {p} rw - ext4 /dev/x rw" for i, p in enumerate(points, 22))


def test_a_files_root_on_the_containers_own_disk_is_refused(tmp_path):
    # The image's empty /files, left unmounted by an old compose file: every file would 404 and
    # every download would go with the container.
    root = str(tmp_path)
    plain = _mounts("/", "/proc", "/etc/hosts")
    assert "not a mounted volume" in files_root_problem(root, plain, in_container=True)
    assert "does not exist" in files_root_problem(str(tmp_path / "gone"))
    # Its own mount, or inside one: the volume — whatever device it reports. On a storage driver that
    # keeps the container's root on the host's disk (vfs) a volume has the same device as /.
    assert files_root_problem(root, _mounts("/", root), in_container=True) is None
    assert files_root_problem(root, _mounts("/", str(tmp_path.parent)), in_container=True) is None
    # A mount point with a space in it, as the kernel escapes it.
    spaced = tmp_path / "my files"
    spaced.mkdir()
    escaped = str(spaced).replace(" ", "\\040")
    assert files_root_problem(str(spaced), _mounts("/", escaped), in_container=True) is None
    # Outside a container a plain directory is a fine root.
    assert files_root_problem(root, plain, in_container=False) is None


def test_the_old_name_of_the_files_root_still_finds_the_files(monkeypatch):
    from app.config import Settings

    monkeypatch.delenv("FILES_ROOT", raising=False)
    monkeypatch.setenv("DOWNLOAD_ROOT", "/downloads")
    assert Settings().files_root == "/downloads"
    monkeypatch.setenv("FILES_ROOT", "/files")
    assert Settings().files_root == "/files"