"""An HLS artifact request names the session it belongs to, and gets nothing else. Sessions are keyed by rating_key ALONE, while a session's directory is per (key, offset, tag). So a seek-restart — or a second viewer of the same item — replaces the registered session, and the old viewer's next request for `seg_12.ts`, resolved by rating_key, was answered 200 with the NEW session's twelfth segment: bytes produced from a different `-ss` offset under `-copyts`, i.e. a timeline the client is not on. Wrong content with a success status is the one failure a player cannot notice, and the reclaim-on-404 added in round 4 could never fire on it (deep-review round 5). No `db` fixture: the route reads nothing from the database (its own `Depends(get_db)` went with the removed auto-start, and `current_user` is overridden wholesale), so these must not sit in the DB lane — there they SKIP on a machine with no Postgres while pytest prints green (round 6). """ import time import pytest from fastapi.testclient import TestClient import app.state as state from app.auth import current_user from app.main import app from app.models import User from app.plex import stream as plex_stream class _Proc: def __init__(self, running: bool = False): self.running = running def poll(self): return None if self.running else 0 def terminate(self): pass def wait(self, timeout=None): return 0 class _Session: def __init__(self, session_id: str, directory, running: bool = False): self.id = session_id self.dir = directory self.proc = _Proc(running) self.last_access = time.time() @pytest.fixture def sessions(monkeypatch): live: dict = {} monkeypatch.setattr(plex_stream, "_sessions", live) return live @pytest.fixture def client(monkeypatch): monkeypatch.setattr(state, "_configured_cache", True) user = User(email="hls@example.com") app.dependency_overrides[current_user] = lambda: user try: yield TestClient(app) finally: app.dependency_overrides.pop(current_user, None) def test_a_segment_is_served_from_the_session_the_url_names(client, tmp_path, sessions): d = tmp_path / "m1_0_0_+0.00" d.mkdir() (d / "seg_11.ts").write_bytes(b"") # its predecessor, so seg_12 is the next one (d / "seg_12.ts").write_bytes(b"mine") sessions["m1"] = _Session("sid-a", d) r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") assert r.status_code == 200, r.text assert r.content == b"mine" def test_a_segment_of_a_replaced_session_is_a_404_not_the_new_sessions_bytes( client, tmp_path, sessions ): """Viewer A is at 0:00; viewer B seeks to 25:00 and their session takes the registry slot. A's next segment must fail honestly — the player reclaims at the playhead — not arrive as B's.""" theirs = tmp_path / "m1_1500_0_+0.00" theirs.mkdir() (theirs / "seg_11.ts").write_bytes(b"") (theirs / "seg_12.ts").write_bytes(b"theirs") sessions["m1"] = _Session("sid-b", theirs) r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") assert r.status_code == 404, r.text assert b"theirs" not in r.content def test_a_playlist_with_no_live_session_is_a_404_not_a_silent_restart_at_zero( client, monkeypatch, sessions ): """The entry playlist used to auto-start a session when none was registered — always at 0.0, silently throwing away the viewer's position. A 404 sends the player through its own reclaim, which restarts at the playhead.""" started: list[float] = [] monkeypatch.setattr( plex_stream, "start_session", lambda *a, **k: started.append(a[2] if len(a) > 2 else 0.0), ) r = client.get("/api/plex/stream/m1/hls/sid-gone/index.m3u8") assert r.status_code == 404, r.text assert started == [] def test_a_segment_that_will_never_arrive_fails_at_once(client, tmp_path, sessions): """`stream_hls_file` is a SYNC route: it runs on the ~40-slot threadpool and holds a pooled DB connection while it waits. It used to poll out the whole 20 s for a name that could never appear — an ordinary seek past the produced region was enough to park a worker and a connection, and forty hopeless names parked the API (deep-review round 5). ffmpeg writes segments in order, so a segment whose predecessor is absent is not the next one, and that is knowable at once — even while ffmpeg is still running, which is the case `proc` alone cannot shorten.""" d = tmp_path / "m1_0_0_+0.00" d.mkdir() (d / "seg_0.ts").write_bytes(b"x") # the session is producing; seg_99999 is simply not next sessions["m1"] = _Session("sid-a", d, running=True) started = time.monotonic() r = client.get("/api/plex/stream/m1/hls/sid-a/seg_99999.ts") elapsed = time.monotonic() - started assert r.status_code == 404, r.text # The wait is 20 s; anything in that neighbourhood means the producer's exit was not consulted. assert elapsed < 2.0, f"waited {elapsed:.1f}s for a segment that can never arrive" @pytest.mark.parametrize( "name", [ "seg_0_9000.ts", # the VIDEO rendition of a multi-audio session: variant 0, segment 9000 "seg_a0_9000.ts", # the first audio rendition: `name:a0` from -var_stream_map ], ) def test_a_multi_audio_segment_beyond_the_region_fails_at_once_too( client, tmp_path, sessions, name ): """The round-5 guard read `seg__.ts`, but ffmpeg substitutes `%v` FIRST — so the real names are `seg__.ts` and every multi-audio request fell through to the full 20 s wait: the video rendition because its variant "0" parsed as segment 0, the audio renditions (`a0`, `a1`) because they did not parse at all. Dual-audio is the common shape in this library, so the guard was inert exactly where it was needed (round 6).""" d = tmp_path / "m1_0_multi_+0.00" d.mkdir() (d / "seg_0_0.ts").write_bytes(b"x") (d / "seg_a0_0.ts").write_bytes(b"x") sessions["m1"] = _Session("sid-a", d, running=True) started = time.monotonic() r = client.get(f"/api/plex/stream/m1/hls/sid-a/{name}") elapsed = time.monotonic() - started assert r.status_code == 404, r.text assert elapsed < 2.0, f"waited {elapsed:.1f}s for {name}" def test_a_multi_audio_segment_whose_predecessor_exists_is_served(client, tmp_path, sessions): """The other half of the same rule: reading the names correctly must not start refusing real ones. `seg_a0_1.ts` follows `seg_a0_0.ts` — the previous SEGMENT of that rendition, not the previous variant.""" d = tmp_path / "m1_0_multi_+0.00" d.mkdir() (d / "seg_a0_0.ts").write_bytes(b"x") (d / "seg_a0_1.ts").write_bytes(b"audio") sessions["m1"] = _Session("sid-a", d, running=True) r = client.get("/api/plex/stream/m1/hls/sid-a/seg_a0_1.ts") assert r.status_code == 200, r.text assert r.content == b"audio" def test_a_name_the_session_cannot_produce_fails_at_once(client, tmp_path, sessions): """The guard was a blocklist: anything that was not a single-audio segment name was waved through to the timeout, so `GET …/anything.m3u8` still parked a worker for the full 30 s — and the route comment claimed a flood of hopeless names was what it prevented. A session produces its playlists and its segments and nothing else, so any other name is answerable now.""" d = tmp_path / "m1_0_0_+0.00" d.mkdir() sessions["m1"] = _Session("sid-a", d, running=True) started = time.monotonic() r = client.get("/api/plex/stream/m1/hls/sid-a/nope.m3u8") elapsed = time.monotonic() - started assert r.status_code == 404, r.text assert elapsed < 2.0, f"waited {elapsed:.1f}s for a playlist this session never writes" @pytest.mark.parametrize("digits", [100, 5000]) def test_an_absurd_segment_index_is_a_404_not_a_500(client, tmp_path, sessions, digits): """`int()` on an unbounded digit run raises ValueError past 4300 digits, and only OSError was caught — so a long enough name reached the client as a 500 with a traceback. Same bug class the subtitle routes were bounded for; the answer is the same 404 any other unproducible name gets.""" d = tmp_path / "m1_0_0_+0.00" d.mkdir() sessions["m1"] = _Session("sid-a", d, running=True) r = client.get(f"/api/plex/stream/m1/hls/sid-a/seg_{'9' * digits}.ts") assert r.status_code == 404, r.text def test_a_session_torn_down_during_the_read_is_a_404_not_a_500( client, monkeypatch, tmp_path, sessions ): """`_kill` rmtrees the whole directory and nothing holds `_lock` across the wait and the read, so the file can vanish in between — on every seek-restart, since `loadSession` awaits the new session before destroying the old hls instance. The player's recovery is keyed on the 404 STATUS, so a 500 here skipped the reclaim and ended playback with "couldn't play".""" d = tmp_path / "m1_0_0_+0.00" d.mkdir() (d / "seg_11.ts").write_bytes(b"") sessions["m1"] = _Session("sid-a", d) # The file existed when the wait checked, and is gone by the time the route reads it. monkeypatch.setattr(plex_stream, "wait_for", lambda *a, **k: True) r = client.get("/api/plex/stream/m1/hls/sid-a/seg_12.ts") assert r.status_code == 404, r.text def test_a_dead_producer_is_not_waited_out(client, tmp_path, sessions): """The route passes `proc` to `wait_for`. Without it a session whose ffmpeg has already exited — a finished remux, a VAAPI init refusal — polls the whole 20 s for a segment that will never be written, holding a threadpool worker and its pooled DB connection. The predecessor IS on disk here, so `artifact_may_arrive` says yes and only `proc` can end the wait.""" d = tmp_path / "m1_0_0_+0.00" d.mkdir() (d / "seg_5.ts").write_bytes(b"x") sessions["m1"] = _Session("sid-a", d, running=False) started = time.monotonic() r = client.get("/api/plex/stream/m1/hls/sid-a/seg_6.ts") elapsed = time.monotonic() - started assert r.status_code == 404, r.text assert elapsed < 2.0, f"waited {elapsed:.1f}s on a producer that had already exited" @pytest.mark.parametrize("is_multi", [False, True]) def test_the_name_ffmpeg_is_asked_for_is_a_name_the_guard_understands(tmp_path, is_multi): """One vocabulary, enforced. `segment_name` is what the warm-up and the PTS probe wait for, and `artifact_may_arrive` is what the route parses — they were three separate statements of the layout and two of them were wrong. Pin them to each other instead.""" name = plex_stream.segment_name(3, is_multi) prev = plex_stream.segment_name(2, is_multi) assert not plex_stream.artifact_may_arrive(tmp_path, name) (tmp_path / prev).write_bytes(b"x") assert plex_stream.artifact_may_arrive(tmp_path, name)