"""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
import sys
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",
        # a torn-down session's directory, moved aside to be deleted outside the lock — left behind
        # only if the process died between the rename and the delete
        "12345_u7_842_multi_+12.50.dead-0123abcd",
    ],
)
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",
        "12345_u7_0_0_+0.00.dead-",  # a tombstone suffix with no id
        "12345_u7_0_0_+0.00.dead-notours!",
        "backup_u7_0_0_+0.00.dead-0123abcd",
    ],
)
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[(1, "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 teardown/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))
        self.killed = False

    def poll(self):
        return None

    def terminate(self):
        self.killed = True

    def wait(self, timeout=None):
        return 0

    def kill(self):
        self.killed = True


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

    def test_hdr_tonemap_uses_the_opencl_graph(self):
        # The HDR/10-bit HW path: VAAPI decode → OpenCL tonemap → VAAPI encode. Gen 9.5 can't tonemap in
        # the VAAPI VPP, so this is the only HW route for a 10-bit/HDR source.
        cmd = stream._ffmpeg_cmd(
            self.SRC, 0.0, self.OUT, None, transcode=True, hw=True, tonemap=True
        )
        assert _venc(cmd) == "h264_vaapi"
        assert "libx264" not in cmd
        # Named devices + an OpenCL device DERIVED from the VAAPI one, all before -i.
        assert cmd[cmd.index("-init_hw_device") + 1] == f"vaapi=va:{stream._VAAPI_DEVICE}"
        assert "opencl=ocl@va" in cmd
        assert cmd[cmd.index("-hwaccel_device") + 1] == "va"
        assert cmd.index("-init_hw_device") < cmd.index("-i")
        # The filter graph runs on the derived OpenCL device and tone-maps HDR→SDR to 8-bit nv12.
        assert cmd[cmd.index("-filter_hw_device") + 1] == "ocl"
        vf = cmd[cmd.index("-vf") + 1]
        assert "hwmap=derive_device=opencl" in vf
        assert "tonemap_opencl=" in vf and "format=nv12" in vf
        assert "hwmap=derive_device=vaapi:reverse=1" in vf
        assert "scale_vaapi" not in vf  # the plain 8-bit path's filter must NOT be used here
        assert (
            cmd[cmd.index("-force_key_frames") + 1] == f"expr:gte(t,n_forced*{stream._SEG_SECONDS})"
        )


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. A HW session on a 10-bit/HDR source
    tone-maps on the GPU via OpenCL (tonemap=True) when the OpenCL runtime is present, else drops to
    software — see the `_select_encode` cases below."""

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

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

    @staticmethod
    def _kind(monkeypatch, kind):
        # Pin the source classifier so the tests don't shell out to ffprobe. `kind` is one of
        # stream._VK_HWSAFE / _VK_HDR / _VK_SDR10.
        monkeypatch.setattr(stream, "_probe_video_kind", lambda src: kind)

    @staticmethod
    def _hwsafe(monkeypatch, safe):
        # Back-compat convenience for the hw-gate tests that only care about 8-bit-SDR vs not: True →
        # _VK_HWSAFE, False → _VK_SDR10 (a non-HDR non-hwsafe class, so the hw half still resolves).
        monkeypatch.setattr(
            stream, "_probe_video_kind", lambda src: stream._VK_HWSAFE if safe else stream._VK_SDR10
        )

    @staticmethod
    def _opencl(monkeypatch, present):
        # Pin the OpenCL-runtime probe so the routing doesn't depend on the test machine's /etc/OpenCL.
        monkeypatch.setattr(stream, "_opencl_present", lambda: present)

    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, src=self.SRC)
            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, src=self.SRC)
            is False
        )

    def test_auto_uses_hw_for_a_decodable_8bit_source_with_the_render_node(self, monkeypatch):
        self._policy(monkeypatch, "auto")
        self._hwsafe(monkeypatch, True)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert (
            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC)
            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")
        self._hwsafe(monkeypatch, True)  # decodability gate must fire BEFORE the probe
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert (
            stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True, src=self.SRC)
            is False
        )
        assert (
            stream._want_hw(None, _FakeItem(codec_video="mpeg4"), transcode=True, src=self.SRC)
            is False
        )

    def test_auto_hdr_source_falls_back_to_software_without_opencl(self, monkeypatch):
        # A decodable HEVC that is true-HDR (PQ/HLG) can't use scale_vaapi=format=nv12 (Gen 9.5), and with
        # no OpenCL runtime to tonemap on the GPU it drops to software libx264 (BUG 1's behaviour).
        self._policy(monkeypatch, "auto")
        self._kind(monkeypatch, stream._VK_HDR)
        self._opencl(monkeypatch, False)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (False, False)

    def test_auto_hdr_source_uses_hw_tonemap_with_opencl(self, monkeypatch):
        # With the OpenCL runtime present, a true-HDR HEVC tone-maps on the GPU (VAAPI decode → OpenCL
        # tonemap → VAAPI encode) instead of dropping to CPU — the HW HDR path.
        self._policy(monkeypatch, "auto")
        self._kind(monkeypatch, stream._VK_HDR)
        self._opencl(monkeypatch, True)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (True, True)

    def test_auto_10bit_SDR_source_uses_software_not_tonemap(self, monkeypatch):
        # F1 regression: a 10-bit *SDR* (bt709) source is NOT HDR, so it must NOT get the hable HDR tone
        # curve even when the OpenCL runtime is present — that would visibly alter an SDR picture. It takes
        # software libx264 (an exact 10→8-bit downconvert), never (True, True).
        self._policy(monkeypatch, "auto")
        self._kind(monkeypatch, stream._VK_SDR10)
        self._opencl(monkeypatch, True)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (False, False)

    def test_auto_8bit_source_uses_plain_hw_not_tonemap(self, monkeypatch):
        # An 8-bit SDR source takes the plain full-GPU scale path — never the tonemap graph — even where
        # OpenCL is available.
        self._policy(monkeypatch, "auto")
        self._kind(monkeypatch, stream._VK_HWSAFE)
        self._opencl(monkeypatch, True)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (True, 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, src=self.SRC)
            is False
        )

    def test_on_forces_hw_and_skips_both_source_gates(self, monkeypatch):
        # The admin override trusts the GPU; skips the decodability AND the bit-depth gate (the runtime
        # HW→SW fallback still backstops it). Render node still required — HW is impossible without it.
        self._policy(monkeypatch, "on")
        self._hwsafe(monkeypatch, False)  # would veto in auto; "on" ignores it
        self._opencl(
            monkeypatch, False
        )  # pin: tonemap-half is irrelevant here, keep it deterministic
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert (
            stream._want_hw(None, _FakeItem(codec_video="av1"), transcode=True, src=self.SRC)
            is True
        )
        monkeypatch.setattr(stream, "_render_node_present", lambda: False)
        assert (
            stream._want_hw(None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC)
            is False
        )

    def test_on_still_picks_the_tonemap_graph_for_an_hdr_source(self, monkeypatch):
        # "on" trusts the GPU but must still route a true-HDR source through the tonemap graph (plain
        # scale_vaapi would hit the VPP wall) when OpenCL is present.
        self._policy(monkeypatch, "on")
        self._kind(monkeypatch, stream._VK_HDR)
        self._opencl(monkeypatch, True)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (True, True)

    def test_on_10bit_SDR_uses_plain_hw_not_tonemap(self, monkeypatch):
        # F1 regression under "on": a 10-bit SDR source is NOT HDR, so even with OpenCL present "on" must
        # pick plain HW (True, False) — the HDR tone curve must never touch an SDR picture.
        self._policy(monkeypatch, "on")
        self._kind(monkeypatch, stream._VK_SDR10)
        self._opencl(monkeypatch, True)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (True, False)

    def test_on_without_opencl_is_plain_hw(self, monkeypatch):
        # "on" with no OpenCL runtime → plain HW, tonemap=False, regardless of source class (the runtime
        # HW→SW fallback backstops a source the plain path can't actually handle). Pins the tonemap half
        # of the "on" tuple, which test_on_forces_hw_and_skips_both_source_gates only checks via _want_hw.
        self._policy(monkeypatch, "on")
        self._kind(monkeypatch, stream._VK_HDR)  # even HDR: no runtime → can't tonemap on the GPU
        self._opencl(monkeypatch, False)
        monkeypatch.setattr(stream, "_render_node_present", lambda: True)
        assert stream._select_encode(
            None, _FakeItem(codec_video="hevc"), transcode=True, src=self.SRC
        ) == (True, 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.sysconfig, "get_bool", lambda db, k: False)
        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, proc=None):
            waited.append(path.name)
            return True

        monkeypatch.setattr(stream, "wait_for", spy)
        s = stream.start_session(db=None, user_id=7, 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, proc=None):
            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, user_id=7, 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, proc=None):
            waited.append(path.name)
            return False

        monkeypatch.setattr(stream, "wait_for", spy)
        s = stream.start_session(db=None, user_id=7, 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,
                "plex_max_sessions": 8,
            }[k],
        )
        monkeypatch.setattr(
            stream.sysconfig,
            "get_str",
            lambda db, k: {"plex_transcode_preset": "slow"}.get(k, "auto"),
        )
        # Loudness normalisation (item 14) is off unless a test turns it on via `stream._loudnorm_on`.
        monkeypatch.setattr(
            stream.sysconfig,
            "get_bool",
            lambda db, k: {"plex_audio_loudnorm": getattr(stream, "_loudnorm_on", False)}[k],
        )
        monkeypatch.setattr(stream, "wait_for", lambda path, timeout=20.0, proc=None: 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, "_select_encode", lambda db, item, transcode, src: (False, False)
        )
        s = stream.start_session(
            db=None, user_id=7, 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, "_select_encode", lambda db, item, transcode, src: (True, False)
        )
        s = stream.start_session(
            db=None, user_id=7, 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 and s.tonemap is False
        assert len(_CapturingProc.calls) == 1  # first segment arrived — no HW→SW fallback respawn
        (cmd, *_), kwargs = _CapturingProc.calls[0]
        assert _venc(cmd) == "h264_vaapi"
        assert "libx264" not in cmd
        # Plain-HW path unconditionally emits the nv12 downconvert (a disjunction on "-vf not in cmd"
        # would silently pass if the filter were dropped — assert it is actually present).
        assert 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_hdr_tonemap_transcode_threads_the_opencl_graph_and_flags_the_session(
        self, capturing_env, monkeypatch
    ):
        # A HW tonemap session (10-bit/HDR + OpenCL) must build the OpenCL graph in the argv and carry
        # tonemap=True on the session, still on the GPU (hw=True) and never niced.
        monkeypatch.setattr(stream, "_select_encode", lambda db, item, transcode, src: (True, True))
        s = stream.start_session(
            db=None, user_id=7, 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 and s.tonemap is True
        (cmd, *_), kwargs = _CapturingProc.calls[0]
        assert _venc(cmd) == "h264_vaapi"
        assert "opencl=ocl@va" in cmd and cmd[cmd.index("-filter_hw_device") + 1] == "ocl"
        assert "tonemap_opencl=" in cmd[cmd.index("-vf") + 1]
        assert kwargs.get("preexec_fn") is None

    def test_a_hw_session_that_produces_nothing_falls_back_to_software(
        self, capturing_env, monkeypatch
    ):
        # BUG 1 backstop: _want_hw picked HW but the VAAPI session emits no first segment (a VAProfile
        # refusal the gates didn't catch). start_session must respawn on libx264 and report hw=False so
        # the item still plays and the route/player honestly show CPU.
        monkeypatch.setattr(
            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
        )
        seen = {"n": 0}

        def flaky_wait(path, timeout=20.0, proc=None):
            seen["n"] += 1
            return seen["n"] > 1  # the first seg-0 wait (HW) fails; every wait after succeeds (SW)

        monkeypatch.setattr(stream, "wait_for", flaky_wait)
        s = stream.start_session(
            db=None, user_id=7, 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 False
        )  # honest CPU after the fallback
        assert len(_CapturingProc.calls) == 2  # the HW attempt, then the SW respawn
        (hw_cmd, *_), _ = _CapturingProc.calls[0]
        (sw_cmd, *_), sw_kwargs = _CapturingProc.calls[1]
        assert _venc(hw_cmd) == "h264_vaapi"
        assert _venc(sw_cmd) == "libx264"
        if os.name == "posix":
            assert sw_kwargs.get("preexec_fn") is stream._sw_child_preexec  # the SW child is niced

    def test_a_hw_TONEMAP_session_that_produces_nothing_falls_back_to_software(
        self, capturing_env, monkeypatch
    ):
        # The fallback gate is `not ready and s.hw`, shared by plain-HW and tonemap sessions — but only
        # the plain-HW variant was covered. Pin the tonemap variant too: a (True, True) session that
        # emits nothing (an OpenCL/tonemap init refusal) must respawn on libx264 and end hw=False,
        # tonemap=False, so an HDR item still plays on CPU.
        monkeypatch.setattr(stream, "_select_encode", lambda db, item, transcode, src: (True, True))
        seen = {"n": 0}

        def flaky_wait(path, timeout=20.0, proc=None):
            seen["n"] += 1
            return (
                seen["n"] > 1
            )  # the first seg-0 wait (HW tonemap) fails; every wait after succeeds

        monkeypatch.setattr(stream, "wait_for", flaky_wait)
        s = stream.start_session(
            db=None, user_id=7, 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 False and s.tonemap is False
        assert len(_CapturingProc.calls) == 2  # the HW tonemap attempt, then the SW respawn
        (hw_cmd, *_), _ = _CapturingProc.calls[0]
        (sw_cmd, *_), _ = _CapturingProc.calls[1]
        assert (
            "tonemap_opencl=" in hw_cmd[hw_cmd.index("-vf") + 1]
        )  # first attempt built the OpenCL graph
        assert _venc(sw_cmd) == "libx264"  # respawn is plain software, no tonemap

    def test_a_superseded_hw_session_does_not_respawn_over_the_newer_one(
        self, capturing_env, monkeypatch
    ):
        # F2/race: start_session waits on seg_0 OUTSIDE _lock. If a concurrent same-key start (a rapid
        # seek-restart) or _enforce_cap replaces _sessions[key] while we wait, our proc dying must NOT
        # trigger the SW fallback — that would tear down the NEWER session and breach the transcode cap.
        # The identity guard must detect it's been superseded. It answers with its OWN (dead) session,
        # never the newer one: the answer goes to the request that lost, whose player drops it as stale
        # and stops the session it names — naming the newer one stopped the playback that won.
        monkeypatch.setattr(
            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
        )

        class _Newer:  # stands in for the session a concurrent _spawn registered
            hw = True
            tonemap = False
            transcode = True
            last_access = 0.0

        newer = _Newer()

        captured = {}

        def wait_then_supersede(path, timeout=20.0, proc=None):
            # Simulate the concurrent replacement happening during our out-of-lock wait, then report the
            # (now-killed) HW proc produced nothing.
            captured["s"] = stream._sessions[(7, "12345")]
            stream._sessions[(7, "12345")] = newer
            return False

        monkeypatch.setattr(stream, "wait_for", wait_then_supersede)
        s = stream.start_session(
            db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
        )
        # No SW respawn — only the initial HW _spawn ran; the newer session was left intact.
        assert len(_CapturingProc.calls) == 1
        assert s is captured["s"]  # its own session, not the newer one
        assert stream._sessions[(7, "12345")] is newer

    def test_a_cap_culled_hw_session_does_not_respawn_and_is_not_reported_missing(
        self, capturing_env, monkeypatch
    ):
        # F2/race, the cull sub-case: while we wait on seg_0 out of lock, cross-key cap pressure lets
        # _enforce_cap evict THIS session's key entirely (no replacement). The fallback must still not
        # respawn (that would re-breach the cap the cull just enforced), and must NOT return None — a None
        # is turned by the route into a dishonest "file not found" 404 for a file that exists. It yields
        # our own (now-dead) session instead, so the client's segment requests degrade to the retryable
        # "not available yet" path.
        monkeypatch.setattr(
            stream, "_select_encode", lambda db, item, transcode, src: (True, False)
        )
        captured = {}

        def wait_then_cull(path, timeout=20.0, proc=None):
            captured["s"] = stream._sessions[
                (7, "12345")
            ]  # the HW session the initial _spawn just registered
            del stream._sessions[
                (7, "12345")
            ]  # simulate _enforce_cap dropping the LRU under cap pressure
            return False

        monkeypatch.setattr(stream, "wait_for", wait_then_cull)
        s = stream.start_session(
            db=None, user_id=7, item=_FakeItem(codec_video="hevc"), start_s=0.0, transcode=True
        )
        assert (
            len(_CapturingProc.calls) == 1
        )  # no SW respawn — the cull is respected, cap not re-breached
        assert (
            s is captured["s"]
        )  # our own session yielded, NOT None (which the route would 404 as missing)
        assert "12345" not in stream._sessions  # the culled key stays gone

    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, "_select_encode", lambda db, item, transcode, src: (False, False)
        )
        s = stream.start_session(
            db=None, user_id=7, 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


class TestProbeVideoKind:
    """The ffprobe-based source classifier that routes the transcode pipeline: 8-bit SDR → plain VAAPI;
    true-HDR (PQ/HLG) → the OpenCL tonemap graph; 10-bit *SDR* → software (an exact downconvert, NEVER
    the HDR tone curve — the F1 distinction). Exercises the REAL function body (not a monkeypatch)."""

    @staticmethod
    def _probe(monkeypatch, stdout):
        class _R:
            def __init__(self):
                self.stdout = stdout

        monkeypatch.setattr(stream.subprocess, "run", lambda *a, **k: _R())

    def test_8bit_sdr_is_hwsafe(self, monkeypatch):
        self._probe(monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=8\ncolor_transfer=bt709\n")
        assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE

    def test_10bit_sdr_pix_fmt_is_sdr10(self, monkeypatch):
        # yuv420p10le with a non-HDR transfer (bt2020-10 is NOT PQ/HLG) = 10-bit SDR → software, not
        # tonemap. Signalled by the pix_fmt suffix even when bits_per_raw_sample is absent.
        self._probe(monkeypatch, "pix_fmt=yuv420p10le\ncolor_transfer=bt2020-10\n")
        assert stream._probe_video_kind(Path("/x")) == stream._VK_SDR10

    def test_bits_per_raw_sample_over_8_sdr_is_sdr10(self, monkeypatch):
        self._probe(monkeypatch, "pix_fmt=yuv420p\nbits_per_raw_sample=10\ncolor_transfer=bt709\n")
        assert stream._probe_video_kind(Path("/x")) == stream._VK_SDR10

    def test_pq_transfer_is_hdr(self, monkeypatch):
        # PQ (smpte2084) is true HDR — needs a real tonemap, even if only 8-bit-tagged.
        self._probe(
            monkeypatch, "pix_fmt=yuv420p10le\nbits_per_raw_sample=10\ncolor_transfer=smpte2084\n"
        )
        assert stream._probe_video_kind(Path("/x")) == stream._VK_HDR

    def test_hlg_transfer_is_hdr(self, monkeypatch):
        # HLG (arib-std-b67) is the other HDR transfer — previously untested; it must classify as HDR.
        self._probe(
            monkeypatch,
            "pix_fmt=yuv420p10le\nbits_per_raw_sample=10\ncolor_transfer=arib-std-b67\n",
        )
        assert stream._probe_video_kind(Path("/x")) == stream._VK_HDR

    def test_empty_probe_output_is_hwsafe(self, monkeypatch):
        # A probe that returns nothing (no streams / stripped metadata) → assume hwsafe; the runtime
        # HW→SW fallback backstops a wrong guess.
        self._probe(monkeypatch, "")
        assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE

    def test_probe_failure_assumes_hwsafe(self, monkeypatch):
        # ffprobe missing/erroring → assume hwsafe; the runtime HW→SW fallback is the real backstop, so a
        # probe miss costs at most one wasted HW attempt, never a stranded item.
        def boom(*a, **k):
            raise OSError("no ffprobe")

        monkeypatch.setattr(stream.subprocess, "run", boom)
        assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE


class TestProbeExitStatus:
    """A probe whose ffprobe EXITS NON-ZERO is a failed probe, not an answer. Each probe used to
    read stdout without looking at the exit status, so a refusing ffprobe silently picked the fallback
    — HW-safe, single audio, the requested offset — and skipped the warning its except branch promises.
    Drives the REAL subprocess.run with the probe's own kwargs against a stand-in command that prints a
    plausible answer and then fails, so the test fails whenever the exit status is ignored."""

    @staticmethod
    def _failing_ffprobe(monkeypatch, stdout: str):
        real_run = stream.subprocess.run
        script = f"import sys; sys.stdout.write({stdout!r}); sys.exit(1)"
        monkeypatch.setattr(
            stream.subprocess,
            "run",
            lambda _cmd, **kw: real_run([sys.executable, "-c", script], **kw),
        )

    @staticmethod
    def _events(caplog) -> list[str]:
        return [getattr(r, "event", "") for r in caplog.records]

    def test_video_kind(self, monkeypatch, caplog):
        self._failing_ffprobe(monkeypatch, "pix_fmt=yuv420p10le\ncolor_transfer=smpte2084\n")
        with caplog.at_level("INFO"):
            assert stream._probe_video_kind(Path("/x")) == stream._VK_HWSAFE
        assert "plex.stream.probe_kind_failed" in self._events(caplog)

    def test_audio_langs(self, monkeypatch, caplog):
        self._failing_ffprobe(monkeypatch, "hun\neng\n")
        with caplog.at_level("INFO"):
            assert stream._probe_audio_langs(Path("/x")) == []
        assert "plex.stream.probe_audio_failed" in self._events(caplog)

    def test_first_pts(self, monkeypatch, caplog):
        self._failing_ffprobe(monkeypatch, "12.5\n")
        with caplog.at_level("INFO"):
            assert stream._probe_first_pts(Path("/x")) is None
        assert "plex.stream.first_pts_probe_failed" in self._events(caplog)


class TestOpenclPresent:
    """The OpenCL-runtime probe (_opencl_present) — the precondition for the HW HDR tonemap graph. Drives
    the REAL glob against a temp ICD dir instead of monkeypatching the function, so a regression to the
    ICD path / glob / OSError guard would be caught (this routes every HDR source on prod)."""

    def test_true_when_an_icd_file_is_registered(self, tmp_path, monkeypatch):
        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", tmp_path)
        (tmp_path / "intel.icd").write_text("libintelocl.so\n")
        assert stream._opencl_present() is True

    def test_false_when_the_dir_is_empty(self, tmp_path, monkeypatch):
        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", tmp_path)
        assert stream._opencl_present() is False

    def test_false_and_no_raise_when_the_dir_is_missing(self, tmp_path, monkeypatch):
        # A missing dir: Path.glob yields nothing (no raise) → False.
        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", tmp_path / "nope")
        assert stream._opencl_present() is False

    def test_false_when_glob_raises_oserror(self, monkeypatch):
        # The `except OSError` guard proper: an unreadable ICD dir (permission denied) makes glob RAISE,
        # not return empty. That must be swallowed to False, never propagated up through _select_encode
        # into start_session (which would crash HDR routing).
        class _BoomDir:
            def glob(self, pattern):
                raise OSError("permission denied")

        monkeypatch.setattr(stream, "_OPENCL_ICD_DIR", _BoomDir())
        assert stream._opencl_present() is False


class TestWaitForProcFastFail:
    """wait_for(proc=…) must stop as soon as the ffmpeg proc exits (a VAAPI init refusal dies in ~1s
    producing nothing) instead of blocking the whole timeout — that fast-fail is what keeps the HW→SW
    fallback from turning into a 25s stall. Exercises the REAL wait_for (every other test monkeypatches
    it wholesale, so this branch would otherwise never run)."""

    class _DeadProc:
        def poll(self):
            return 1  # already exited non-zero

    def test_returns_true_when_the_segment_exists_despite_a_dead_proc(self, tmp_path):
        seg = tmp_path / "seg_0.ts"
        seg.write_bytes(b"x")
        # The proc exited but DID write a first segment → ready. (The final existence re-check.)
        assert stream.wait_for(seg, timeout=5.0, proc=self._DeadProc()) is True

    def test_returns_false_fast_when_a_dead_proc_wrote_nothing(self, tmp_path):
        # Missing segment + dead proc → give up immediately, not after `timeout`. Assert it's fast by
        # using a long timeout the test would visibly hang on if the fast-fail branch regressed.
        seg = tmp_path / "seg_0.ts"
        assert stream.wait_for(seg, timeout=30.0, proc=self._DeadProc()) is False

    def test_empty_segment_from_a_dead_proc_is_not_ready(self, tmp_path):
        # A zero-byte segment is not a real segment (size > 0 required).
        seg = tmp_path / "seg_0.ts"
        seg.write_bytes(b"")
        assert stream.wait_for(seg, timeout=30.0, proc=self._DeadProc()) is False


class TestAudioLoudnorm:
    """`plex_audio_loudnorm` (bug hunt 2026-09, item 14): the HLS path always downmixes 5.1 to stereo
    AAC and never normalised it, so a cinema mix arrives with whispered dialogue and deafening action.
    The knob is OFF by default — it makes a film ~11 dB louder and narrows its dynamic range, which is
    a taste call. Measured on a real 5.1 AC3 film: I -27.6 -> -16.6 LUFS, LRA 13.3 -> 9.8 LU.

    Reuses TestTranscodeStartSession's capturing fixture: what matters is the argv ffmpeg is actually
    spawned with, not `_ffmpeg_cmd` called with literals."""

    capturing_env = TestTranscodeStartSession.capturing_env

    @staticmethod
    def _spawn(monkeypatch, *, on: bool, multi: bool = False):
        monkeypatch.setattr(stream, "_loudnorm_on", on, raising=False)
        monkeypatch.setattr(
            stream, "_select_encode", lambda db, item, transcode, src: (False, False)
        )
        if multi:
            monkeypatch.setattr(stream, "_probe_audio_langs", lambda src: ["hun", "eng"])
        s = stream.start_session(db=None, user_id=7, item=_FakeItem(), start_s=0.0, multi=multi)
        assert s is not None
        (cmd, *_), _ = _CapturingProc.calls[0]
        return s, cmd

    def test_the_knob_off_leaves_the_audio_exactly_as_it_was(self, capturing_env, monkeypatch):
        _, cmd = self._spawn(monkeypatch, on=False)
        assert "-af" not in cmd
        assert cmd[cmd.index("-c:a") + 1] == "aac" and cmd[cmd.index("-ac") + 1] == "2"

    def test_the_knob_on_normalises_AFTER_the_downmix(self, capturing_env, monkeypatch):
        # Order is the whole point: measuring 5.1 and folding it down afterwards would normalise
        # something other than what the viewer hears.
        _, cmd = self._spawn(monkeypatch, on=True)
        af = cmd[cmd.index("-af") + 1]
        stages = af.split(",")
        assert stages[0] == "aformat=channel_layouts=stereo"
        assert stages[1].startswith("loudnorm=")
        # loudnorm runs its own 192 kHz resampler internally; the AAC encoder needs 48 kHz back.
        assert stages[-1] == "aresample=48000"

    def test_every_audio_rendition_of_a_multi_audio_file_is_normalised(
        self, capturing_env, monkeypatch
    ):
        # A multi-audio session emits one AAC output per language. `-af` without a stream index
        # covers all of them — a per-stream filter would have normalised only the first.
        _, cmd = self._spawn(monkeypatch, on=True, multi=True)
        assert cmd.count("-map") == 3  # video + two audio tracks
        assert cmd.count("-af") == 1
        assert "loudnorm=" in cmd[cmd.index("-af") + 1]

    def test_the_session_directory_records_the_setting(self, capturing_env, monkeypatch):
        # The knob changes the AUDIO THE SEGMENTS CONTAIN, so it belongs in the directory name with
        # the other parameters that do. Otherwise a session left from before the flip is reused.
        off, _ = self._spawn(monkeypatch, on=False)
        _CapturingProc.calls = []
        on, _ = self._spawn(monkeypatch, on=True)
        assert off.dir.name != on.dir.name
        assert on.dir.name.endswith("-ln_+0.00")

    @pytest.mark.parametrize("on", [False, True])
    def test_the_orphan_sweep_still_recognises_its_own_directory(
        self, capturing_env, monkeypatch, on
    ):
        # The tag is appended with a DASH, not an underscore, precisely so `_SESSION_DIR_RE` (the only
        # thing keeping the sweep off a shared scratch path) still matches. A session directory the
        # sweep no longer recognises is one that never gets cleaned up.
        s, _ = self._spawn(monkeypatch, on=on)
        assert _SESSION_DIR_RE.match(s.dir.name)


class TestSessionsPerViewer:
    """HLS sessions were keyed by the ITEM, so a second account opening the same film killed the
    first one's session; each then reclaimed and evicted the other. And only transcodes counted
    toward a cap, so remuxes — each a whole segmented copy of the film on disk — were unbounded."""

    capturing_env = TestTranscodeStartSession.capturing_env

    @staticmethod
    def _start(monkeypatch, user_id, rating_key="12345"):
        monkeypatch.setattr(
            stream, "_select_encode", lambda db, item, transcode, src: (False, False)
        )
        return stream.start_session(
            db=None, user_id=user_id, item=_FakeItem(rating_key=rating_key), start_s=0.0
        )

    def test_two_viewers_of_one_film_each_keep_their_own_session(self, capturing_env, monkeypatch):
        monkeypatch.setattr(stream, "_sessions", {})
        a = self._start(monkeypatch, user_id=1)
        b = self._start(monkeypatch, user_id=2)
        assert stream._sessions == {(1, "12345"): a, (2, "12345"): b}
        assert not a.proc.killed
        assert a.dir != b.dir and _SESSION_DIR_RE.match(a.dir.name)

    def test_one_account_holds_at_most_two_sessions(self, capturing_env, monkeypatch):
        monkeypatch.setattr(stream, "_sessions", {})
        first = self._start(monkeypatch, user_id=1, rating_key="1")
        first.last_access -= 60  # the least recently used of the three
        self._start(monkeypatch, user_id=1, rating_key="2")
        self._start(monkeypatch, user_id=1, rating_key="3")
        assert sorted(stream._sessions) == [(1, "2"), (1, "3")]
        assert first.proc.killed

    def test_remux_sessions_count_toward_the_overall_ceiling(self, capturing_env, monkeypatch):
        monkeypatch.setattr(stream, "_sessions", {})
        monkeypatch.setattr(
            stream.sysconfig,
            "get_int",
            lambda db, k: {"plex_max_transcodes": 4, "plex_max_sessions": 3}[k],
        )
        oldest = self._start(monkeypatch, user_id=1)
        oldest.last_access -= 60
        for uid in (2, 3, 4):
            self._start(monkeypatch, user_id=uid)
        assert len(stream._sessions) == 3 and (1, "12345") not in stream._sessions
        assert oldest.proc.killed and not oldest.transcode