"""The guards in front of a recursive delete.

`sweep_orphans` rmtree's directories under `_HLS_ROOT`, and that root is admin-configurable
(PLEX_HLS_DIR) — it can legitimately be a shared scratch path. `_SESSION_DIR_RE` is the ONLY thing
that keeps the sweep off a neighbour's data, so the names it accepts are pinned here.

Since the sweep stopped being awaited at startup it also runs CONCURRENTLY with playback, so the
second half of this file pins what keeps it off a LIVE session's segments."""

import os
from pathlib import Path

import pytest

from app.plex import stream
from app.plex.stream import _SESSION_DIR_RE


@pytest.mark.parametrize(
    "name",
    [
        "12345_0",  # legacy shape (pre multi-audio): {rating_key}_{start}
        "12345_842",
        "12345_0_None_+0.00",  # current shape, single audio track (audio_ord None)
        "12345_0_2_+0.00",  # current shape, explicit audio_ord
        "12345_842_multi_-1.25",  # multi-audio, negative offset
        "12345_842_multi_+12.50",
    ],
)
def test_our_session_directories_are_swept(name):
    assert _SESSION_DIR_RE.match(name)


@pytest.mark.parametrize(
    "name",
    [
        "backup_2024",  # the class an "optional tail" pattern wrongly matched
        "release_10",
        "important_data_2024",
        "node_modules",
        "logs",
        "plex-hls",
        "12345",  # no start offset — not a session dir
        "12345_0_multi",  # truncated: missing the offset
        "12345_0_multi_1.25",  # offset without its sign
        "snapshot_2024_backup_+1.00",  # session-ish tail, but the key isn't a rating_key
        "backup_2024_multi_+0.00",
        ".config",
    ],
)
def test_foreign_directories_are_left_alone(name):
    assert not _SESSION_DIR_RE.match(name)


class _FakeSession:
    def __init__(self, directory):
        self.dir = directory


@pytest.fixture
def hls_root(tmp_path, monkeypatch):
    """Point the sweep at a temp root with no sessions registered."""
    monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path)
    monkeypatch.setattr(stream, "_sessions", {})
    return tmp_path


def _session_dir(root, name, *, age_s=0.0):
    """A session directory, optionally backdated by `age_s` to look like a previous run's."""
    d = root / name
    d.mkdir()
    (d / "seg_0.ts").write_bytes(b"x")
    if age_s:
        t = d.stat().st_mtime - age_s
        os.utime(d, (t, t))
    return d


def _cutoff(root):
    """A "process start" read from the FILESYSTEM, not from a clock.

    The guard compares a directory's mtime against a timestamp, and the two do not have to come
    from the same clock: on a bind-mounted dev volume the file times are stamped by the host while
    `time.time()` runs in the container VM, and the drift between them is real (it made an earlier
    version of these tests flaky). Anchoring the cutoff to a file this test just created removes
    the clock from the assertion; that the DEFAULT cutoff is `_PROCESS_START` is pinned separately
    below."""
    probe = root / ".cutoff-probe"
    probe.write_bytes(b"")
    at = probe.stat().st_mtime
    probe.unlink()
    return at


class TestSweepOrphans:
    def test_a_previous_runs_leftovers_are_deleted(self, hls_root):
        d = _session_dir(hls_root, "12345_0_None_+0.00", age_s=3600)
        assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 1
        assert not d.exists()

    def test_foreign_directories_survive(self, hls_root):
        keep = hls_root / "backup_2024"
        keep.mkdir()
        (keep / "important.tar").write_bytes(b"x")
        assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0
        assert (keep / "important.tar").exists()

    def test_a_live_sessions_directory_is_never_deleted(self, hls_root):
        # Registered session, but with an OLD directory — only the live check can save it.
        d = _session_dir(hls_root, "12345_0_multi_+0.00", age_s=3600)
        stream._sessions["k"] = _FakeSession(d)
        assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0
        assert d.exists()

    def test_a_directory_this_run_created_is_never_deleted(self, hls_root):
        # THE RACE: the sweep is no longer awaited before the app serves, and `start_session`
        # REUSES the path of a same-key session — so between our listing and our rmtree, a stale
        # name can become a live directory. A fresh mtime is what tells them apart, and it must
        # hold even while `_sessions` is still empty (the session registers a beat later).
        cutoff = _cutoff(hls_root)
        fresh = _session_dir(hls_root, "12345_0_None_+0.00")  # created "after boot"
        old = _session_dir(hls_root, "999_0_None_+0.00", age_s=3600)
        assert stream.sweep_orphans(older_than=cutoff) == 1  # only the leftover
        assert fresh.exists()
        assert not old.exists()

    def test_the_default_cutoff_is_this_processs_start(self, hls_root, monkeypatch):
        # The parameter exists for the tests; production calls it with no argument, so pin what
        # that means — otherwise the guard above could be right and unused.
        d = _session_dir(hls_root, "12345_0_None_+0.00")
        monkeypatch.setattr(stream, "_MTIME_GRACE_S", 0.0)
        monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 1)
        assert stream.sweep_orphans() == 1  # "older than boot" → swept
        assert not d.exists()

    def test_a_directory_stamped_just_before_boot_survives(self, hls_root, monkeypatch):
        # MEASURED in the container, not theory: a file's mtime comes from the kernel's COARSE
        # clock and `time.time()` from the fine one, so a directory created 2ms AFTER
        # `_PROCESS_START` was stamped 2ms BEFORE it. Without the grace the sweep deletes a session
        # that started in the first instants after boot — which is exactly when the sweep runs.
        d = _session_dir(hls_root, "12345_0_None_+0.00")
        monkeypatch.setattr(stream, "_PROCESS_START", d.stat().st_mtime + 0.05)  # the skew
        assert stream.sweep_orphans() == 0
        assert d.exists()

    def test_a_missing_root_is_not_an_error(self, tmp_path, monkeypatch):
        monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path / "nope")
        monkeypatch.setattr(stream, "_sessions", {})
        assert stream.sweep_orphans() == 0

    def test_a_stray_file_with_a_session_name_is_left_alone(self, hls_root):
        # Only directories are ours; a same-named FILE belongs to someone else.
        f = hls_root / "12345_0"
        f.write_bytes(b"x")
        os.utime(f, (f.stat().st_mtime - 3600,) * 2)  # old enough that only is_dir() saves it
        assert stream.sweep_orphans(older_than=_cutoff(hls_root)) == 0
        assert f.exists()


class _FakeProc:
    """Stand-in for the ffmpeg Popen — start_session only stores it (and _kill/reap poll it later)."""

    def poll(self):
        return None


class _FakeItem:
    def __init__(self, rating_key="12345", file_path="/media/movie.mkv", codec_video="h264"):
        self.rating_key = rating_key
        self.file_path = file_path
        self.playable = "remux"
        self.codec_video = codec_video


class _CapturingProc:
    """Records the argv + kwargs each Popen was called with, then behaves like the ffmpeg proc."""

    calls: list[tuple] = []

    def __init__(self, *args, **kwargs):
        _CapturingProc.calls.append((args, kwargs))

    def poll(self):
        return None


def _venc(cmd):
    """The video codec ffmpeg was told to use (the value after the first `-c:v`)."""
    return cmd[cmd.index("-c:v") + 1]


class TestTranscodeCommand:
    """The transcode video re-encode (playable='transcode'): the SAME HLS command as remux with the
    video encoder swapped in for `-c:v copy`. The HW path can only be exercised on prod (VAAPI needs
    Linux + Intel /dev/dri), so its SHAPE is pinned here; the software path is the one UAT'd locally."""

    SRC = Path("/media/movie.mkv")
    OUT = Path("/out")

    def test_remux_still_stream_copies_the_video(self):
        cmd = stream._ffmpeg_cmd(self.SRC, 0.0, self.OUT, None)
        assert _venc(cmd) == "copy"
        assert "libx264" not in cmd and "h264_vaapi" not in cmd
        assert "-hwaccel" not in cmd  # no GPU decode on the copy path
        assert "-force_key_frames" not in cmd  # copy leans on the source's own keyframes

    def test_software_transcode_uses_libx264_and_mints_keyframes(self):
        cmd = stream._ffmpeg_cmd(self.SRC, 0.0, self.OUT, None, transcode=True, hw=False)
        assert _venc(cmd) == "libx264"
        assert "copy" not in cmd  # the video is no longer stream-copied
        # 8-bit High (yuv420p), not High10, so any browser decodes a 10-bit HEVC source.
        assert cmd[cmd.index("-pix_fmt") + 1] == "yuv420p"
        assert "-threads" in cmd  # a core reserved for request handling / playback
        assert "-hwaccel" not in cmd  # software path: CPU decode, no render node needed
        # A forced keyframe at every segment boundary so the HLS muxer cuts independent segments.
        assert (
            cmd[cmd.index("-force_key_frames") + 1] == f"expr:gte(t,n_forced*{stream._SEG_SECONDS})"
        )

    def test_hardware_transcode_uses_the_full_vaapi_pipeline(self):
        cmd = stream._ffmpeg_cmd(self.SRC, 0.0, self.OUT, None, transcode=True, hw=True)
        assert _venc(cmd) == "h264_vaapi"
        assert "libx264" not in cmd
        # GPU decode options must precede the input so the source is decoded on-device.
        assert cmd[cmd.index("-hwaccel") + 1] == "vaapi"
        assert "-hwaccel_output_format" in cmd
        assert cmd[cmd.index("-vaapi_device") + 1] == stream._VAAPI_DEVICE
        assert cmd.index("-hwaccel") < cmd.index("-i")
        # scale_vaapi=format=nv12 downconverts a 10-bit source to 8-bit nv12 for h264_vaapi.
        assert "scale_vaapi=format=nv12" in cmd
        assert (
            cmd[cmd.index("-force_key_frames") + 1] == f"expr:gte(t,n_forced*{stream._SEG_SECONDS})"
        )

    def test_multi_audio_transcode_reencodes_the_video_variant(self):
        cmd = stream._ffmpeg_cmd(
            self.SRC, 0.0, self.OUT, None, audio_langs=["eng", "hun"], transcode=True, hw=False
        )
        assert _venc(cmd) == "libx264"  # video re-encoded…
        assert "-var_stream_map" in cmd  # …but still multi-rendition (client-side audio switch)
        assert "copy" not in cmd

    def test_software_preset_and_crf_come_from_the_knobs(self):
        cmd = stream._ffmpeg_cmd(
            self.SRC, 0.0, self.OUT, None, transcode=True, hw=False, preset="slow", crf=18
        )
        assert cmd[cmd.index("-preset") + 1] == "slow"
        assert cmd[cmd.index("-crf") + 1] == "18"

    def test_software_max_height_caps_with_a_downscale_only_filter(self):
        cmd = stream._ffmpeg_cmd(
            self.SRC, 0.0, self.OUT, None, transcode=True, hw=False, max_height=720
        )
        # Downscale-only, aspect-preserving, even width; the comma in min() escaped so it's ONE filter.
        assert cmd[cmd.index("-vf") + 1] == "scale=-2:min(ih\\,720)"

    def test_software_max_height_zero_means_no_scale_filter(self):
        cmd = stream._ffmpeg_cmd(
            self.SRC, 0.0, self.OUT, None, transcode=True, hw=False, max_height=0
        )
        assert "-vf" not in cmd  # source resolution kept

    def test_hardware_ignores_max_height(self):
        # The GPU encode is cheap, so it runs at source resolution; max_height is a software-only knob.
        cmd = stream._ffmpeg_cmd(
            self.SRC, 0.0, self.OUT, None, transcode=True, hw=True, max_height=720
        )
        assert cmd[cmd.index("-vf") + 1] == "scale_vaapi=format=nv12"  # not a software scale


class TestWantHw:
    """VAAPI is used only where the render node was passed in AND (in auto) the source is HW-decodable,
    with an admin `plex_transcode_hw` policy (auto/on/off) on top."""

    @staticmethod
    def _policy(monkeypatch, mode):
        monkeypatch.setattr(stream.sysconfig, "get_str", lambda db, k: mode)

    def test_no_hw_when_not_transcoding(self, monkeypatch):
        self._policy(monkeypatch, "auto")
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=False) is False

    def test_no_hw_without_a_render_node(self, monkeypatch):
        self._policy(monkeypatch, "auto")
        monkeypatch.setattr(stream, "_render_node_present", lambda: False)
        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False

    def test_auto_uses_hw_for_a_decodable_source_with_the_render_node(self, monkeypatch):
        self._policy(monkeypatch, "auto")
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is True

    def test_auto_falls_back_to_software_for_a_non_decodable_source(self, monkeypatch):
        # AV1 / MPEG-4 ASP (XviD) aren't decodable on Gen 9.5 → the GPU can't feed its own encoder.
        self._policy(monkeypatch, "auto")
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True) is False
        assert stream._want_hw(None, _FakeItem(codec_video="mpeg4"), transcode=True) is False

    def test_off_forces_software_even_for_a_decodable_source(self, monkeypatch):
        self._policy(monkeypatch, "off")
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False

    def test_on_forces_hw_even_for_a_non_decodable_source(self, monkeypatch):
        # The admin override trusts the GPU to decode it; skips the decodability gate (render node still
        # required — HW is impossible without it).
        self._policy(monkeypatch, "on")
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True) is True
        monkeypatch.setattr(stream, "_render_node_present", lambda: False)
        assert stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True) is False


class TestColdStartWarmup:
    """The cold-start buffer warm-up: before reporting a session ready, start_session waits for the
    first _WARMUP_SEGMENTS video segments so hls.js doesn't drain a one-segment buffer and stall a
    few seconds in (the auto-advance-to-next-episode stall). We spy on wait_for instead of running
    ffmpeg."""

    @pytest.fixture
    def fake_env(self, tmp_path, monkeypatch):
        monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path)
        monkeypatch.setattr(stream, "_sessions", {})
        src = tmp_path / "movie.mkv"
        src.write_bytes(b"x")
        monkeypatch.setattr(stream.paths, "local_media_path", lambda db, fp: src)
        monkeypatch.setattr(stream.sysconfig, "get_int", lambda db, k: 2)
        monkeypatch.setattr(stream.subprocess, "Popen", lambda *a, **k: _FakeProc())
        return tmp_path

    def test_warmup_waits_for_the_first_few_segments(self, fake_env, monkeypatch):
        waited = []

        def spy(path, timeout=20.0):
            waited.append(path.name)
            return True

        monkeypatch.setattr(stream, "wait_for", spy)
        s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0)
        assert s is not None
        # seg_0 (unconditional now) + the warm-up run up to _WARMUP_SEGMENTS.
        assert waited == [f"seg_{i}.ts" for i in range(stream._WARMUP_SEGMENTS)]

    def test_warmup_stops_at_the_first_segment_that_isnt_ready(self, fake_env, monkeypatch):
        waited = []

        def spy(path, timeout=20.0):
            waited.append(path.name)
            return path.name != "seg_1.ts"  # seg_1 never appears → warm-up gives up there

        monkeypatch.setattr(stream, "wait_for", spy)
        s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0)
        assert s is not None
        assert waited == ["seg_0.ts", "seg_1.ts"]  # never asked for seg_2

    def test_no_warmup_wait_when_the_first_segment_never_arrives(self, fake_env, monkeypatch):
        # If seg_0 itself times out we don't warm up at all (nothing to build on) — and the session is
        # still returned so the client's own segment requests can drive it, as before.
        waited = []

        def spy(path, timeout=20.0):
            waited.append(path.name)
            return False

        monkeypatch.setattr(stream, "wait_for", spy)
        s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0)
        assert s is not None
        assert waited == ["seg_0.ts"]


class TestTranscodeStartSession:
    """The seam TestTranscodeCommand can't reach: start_session reading the three admin knobs
    (plex_transcode_preset/crf/max_height) and threading them + transcode/hw into the ffmpeg argv and
    the HlsSession. A swapped knob here is a SILENT mis-encode (still plays), so pin the whole wiring
    end-to-end with a real (captured) Popen argv — not just _ffmpeg_cmd called with literals."""

    @pytest.fixture
    def capturing_env(self, tmp_path, monkeypatch):
        monkeypatch.setattr(stream, "_HLS_ROOT", tmp_path)
        monkeypatch.setattr(stream, "_sessions", {})
        src = tmp_path / "movie.mkv"
        src.write_bytes(b"x")
        monkeypatch.setattr(stream.paths, "local_media_path", lambda db, fp: src)
        # Distinct sentinels per knob so a crf<->max_height swap (both ints, both via get_int) fails.
        monkeypatch.setattr(
            stream.sysconfig,
            "get_int",
            lambda db, k: {
                "plex_transcode_crf": 18,
                "plex_transcode_max_height": 480,
                "plex_max_transcodes": 4,
            }[k],
        )
        monkeypatch.setattr(
            stream.sysconfig,
            "get_str",
            lambda db, k: {"plex_transcode_preset": "slow"}.get(k, "auto"),
        )
        monkeypatch.setattr(stream, "wait_for", lambda path, timeout=20.0: True)
        _CapturingProc.calls = []
        monkeypatch.setattr(stream.subprocess, "Popen", _CapturingProc)
        return tmp_path

    def test_software_transcode_threads_the_knobs_into_the_command(
        self, capturing_env, monkeypatch
    ):
        monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: False)
        s = stream.start_session(
            db=None, item=_FakeItem(codec_video="mpeg4"), start_s=0.0, transcode=True
        )
        assert s is not None and s.transcode is True and s.hw is False
        (cmd, *_), kwargs = _CapturingProc.calls[0]
        assert _venc(cmd) == "libx264"
        assert cmd[cmd.index("-preset") + 1] == "slow"  # plex_transcode_preset
        assert cmd[cmd.index("-crf") + 1] == "18"  # plex_transcode_crf, NOT max_height
        assert cmd[cmd.index("-vf") + 1] == "scale=-2:min(ih\\,480)"  # plex_transcode_max_height
        # The CPU-bound child is niced via preexec_fn on POSIX (the API process itself must not be).
        if os.name == "posix":
            assert kwargs.get("preexec_fn") is stream._sw_child_preexec

    def test_hardware_transcode_uses_the_gpu_and_never_nices(self, capturing_env, monkeypatch):
        monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: True)
        s = stream.start_session(
            db=None, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
        )
        assert s is not None and s.transcode is True and s.hw is True
        (cmd, *_), kwargs = _CapturingProc.calls[0]
        assert _venc(cmd) == "h264_vaapi"
        assert "libx264" not in cmd
        assert (
            "-vf" not in cmd or cmd[cmd.index("-vf") + 1] == "scale_vaapi=format=nv12"
        )  # no SW cap
        assert kwargs.get("preexec_fn") is None  # GPU encode is cheap — never niced

    def test_remux_reads_no_transcode_knobs_and_stream_copies(self, capturing_env, monkeypatch):
        # A remux start must stream-copy the video and spawn no niced child (the transcode knobs and
        # the preexec nice are gated behind `transcode=True`).
        monkeypatch.setattr(stream, "_want_hw", lambda db, item, transcode: False)
        s = stream.start_session(db=None, item=_FakeItem(), start_s=0.0)  # transcode defaults False
        assert s is not None and s.transcode is False
        (cmd, *_), kwargs = _CapturingProc.calls[0]
        assert _venc(cmd) == "copy"
        assert kwargs.get("preexec_fn") is None