peter
ca9ca85b18
The 7 baselined Optional accesses in plex/stream.py were one fact the type did not carry: the initial spawn (replace_only=None) always supersedes, so only the fallback respawn can get None. An assert at the call site says so; the pyright baseline drops 195 -> 188. test_abuse_limits: the public-URL and YouTube enqueue tests stopped at the profile lookup (400 no_profile) and passed on that unrelated refusal. They now send an inline spec and assert the job is queued. The same shape — a request whose response is discarded — sat at 53 more places across the suite (all returned what they should, measured with a TestClient probe); each now asserts its status, and test_response_checked keeps it that way (seen red).
1177 lines
55 KiB
Python
1177 lines
55 KiB
Python
"""Subtitle extraction must not happen inside the request, and must read as little as possible.
|
||
|
||
Bug hunt 2026-09, item 8 — "switching the subtitle track shows nothing; after a few stop/starts it
|
||
suddenly works". Measured on prod against a 17.06 GB MKV with two embedded SubRip tracks:
|
||
|
||
full pass, one track 154 s | both tracks in ONE run 111 s | a 630 s window at 1:00:00 8 s
|
||
|
||
The cost is not the WebVTT conversion, it is that ffmpeg streams every byte of the container between
|
||
its seek point and its stop point. The old endpoint did that pass INSIDE the <track>'s request, once
|
||
per track, with no timeout — so the browser sat on an empty track for minutes (and an API worker
|
||
thread sat with it), and picking the second language cost the whole read again.
|
||
|
||
These tests pin the three things that fixes it: the request never waits for ffmpeg, the extraction
|
||
publishes a window around the playhead before the whole track, and one ffmpeg run covers every
|
||
embedded text track — plus what the deep-review of that fix found (cues doubled at the window/full
|
||
seam, a truncated pass cached as complete, a poll that respawned ffmpeg every second, a sidecar stuck
|
||
behind an unrelated pass). No Postgres (the auth half is in test_plex_subtitle_auth.py, which needs
|
||
the DB). Almost every test stubs ffmpeg out, but not all: a few run a real child process, and
|
||
`test_a_window_lines_up_with_the_full_pass_on_a_real_file` needs the real ffmpeg — it is the only
|
||
guard of `-avoid_negative_ts disabled`. It skips without one, except in the gate's DB lane, where
|
||
conftest refuses to run without ffmpeg rather than let it skip.
|
||
"""
|
||
|
||
import asyncio
|
||
import errno
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
import app.state as app_state
|
||
import app.main as app_main
|
||
from app.auth import current_user
|
||
from app.main import app
|
||
from app.plex import subtitles
|
||
|
||
needs_ffmpeg = pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg not available")
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolated_cache(tmp_path, monkeypatch):
|
||
"""A throwaway cache root + empty registries: this module's state is process-global."""
|
||
monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path)
|
||
subtitles.reset_state()
|
||
yield
|
||
subtitles.reset_state()
|
||
|
||
|
||
@pytest.fixture
|
||
def client(monkeypatch, override_dep) -> TestClient:
|
||
# Neither route reads the user; the real `current_user` (and what it rejects) is pinned in
|
||
# test_plex_subtitle_auth.py against the DB.
|
||
override_dep(current_user, lambda: SimpleNamespace(id=1))
|
||
monkeypatch.setattr(app_state, "_configured_cache", True) # pass the instance-setup gate, no DB
|
||
return TestClient(app)
|
||
|
||
|
||
class _ManualPool:
|
||
"""Deterministic stand-in for the extraction thread pool: jobs run when the test says so."""
|
||
|
||
def __init__(self) -> None:
|
||
self.jobs: list[tuple] = []
|
||
|
||
def submit(self, fn, *args) -> None:
|
||
self.jobs.append((fn, args))
|
||
|
||
def run_next(self) -> None:
|
||
fn, args = self.jobs.pop(0)
|
||
fn(*args)
|
||
|
||
def run_all(self) -> int:
|
||
jobs, self.jobs = self.jobs, []
|
||
for fn, args in jobs:
|
||
fn(*args)
|
||
return len(jobs)
|
||
|
||
def shutdown(self, wait=True, cancel_futures=False) -> None:
|
||
pass # jobs still run only when the test says so — which is what the shutdown tests probe
|
||
|
||
|
||
def _manual_pool(monkeypatch) -> _ManualPool:
|
||
"""One manual pool standing in for BOTH extraction pools, so a test sees every job in the order
|
||
it was submitted (window or full pass — `window_start` tells them apart)."""
|
||
pool = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_WIN_POOL", pool)
|
||
monkeypatch.setattr(subtitles, "_FULL_POOL", pool)
|
||
return pool
|
||
|
||
|
||
def _real_pools(monkeypatch) -> None:
|
||
"""Fresh single-worker pools (the module's own may have been shut down by an earlier test)."""
|
||
monkeypatch.setattr(subtitles, "_WIN_POOL", ThreadPoolExecutor(max_workers=1))
|
||
monkeypatch.setattr(subtitles, "_FULL_POOL", ThreadPoolExecutor(max_workers=1))
|
||
|
||
|
||
def _cue(start: str, end: str, text: str) -> str:
|
||
return f"{start} --> {end}\n{text}"
|
||
|
||
|
||
def _vtt(*cues: str) -> str:
|
||
return "WEBVTT\n\n" + "\n\n".join(cues) + "\n"
|
||
|
||
|
||
# --- the request must never wait for the extraction ---------------------------------------------
|
||
|
||
|
||
def test_a_cold_track_answers_at_once_instead_of_waiting_for_ffmpeg(client, monkeypatch):
|
||
"""THE regression. A cold track used to block the request for the length of a full file read;
|
||
now it answers 202 and the extraction runs behind it. Timed, because "does it block?" is the
|
||
whole question — the fake extraction holds a worker for 5 s, which the request must not feel."""
|
||
started = threading.Event()
|
||
release = threading.Event()
|
||
|
||
def _slow(rating_key, sub_ord, window_start):
|
||
started.set()
|
||
release.wait(timeout=5)
|
||
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", _slow)
|
||
try:
|
||
t0 = time.monotonic()
|
||
st = client.get("/api/plex/subtitle/98614/0/status")
|
||
elapsed = time.monotonic() - t0
|
||
assert st.status_code == 200 and st.json()["state"] == "preparing", st.text
|
||
assert elapsed < 2.0, f"the request waited {elapsed:.1f}s on the extraction"
|
||
assert started.wait(timeout=5), "the extraction never started in the background"
|
||
# …and the <track> fetch answers just as fast while that extraction is still running.
|
||
t0 = time.monotonic()
|
||
resp = client.get("/api/plex/subtitle/98614/0")
|
||
assert resp.status_code == 202, resp.text
|
||
assert time.monotonic() - t0 < 2.0
|
||
finally:
|
||
release.set()
|
||
|
||
|
||
def test_a_burst_of_polls_starts_one_window_and_one_full_pass(monkeypatch):
|
||
"""The player polls once a second while it waits. Each poll must JOIN the running jobs, not
|
||
start another ffmpeg over the same 17 GB."""
|
||
pool = _manual_pool(monkeypatch)
|
||
for _ in range(5):
|
||
subtitles.ensure("98614", 0, at=600.0)
|
||
assert len(pool.jobs) == 2 # one window (at 600 s) + the one full pass
|
||
# submit(_run, ident, rating_key, sub_ord, window_start, at) — window_start tells them apart
|
||
assert [a[3] for _, a in pool.jobs] == [570.0, None] # the window first, then the full pass
|
||
|
||
|
||
# --- window first, whole track after ------------------------------------------------------------
|
||
|
||
|
||
def test_status_walks_preparing_to_partial_to_ready(client, monkeypatch):
|
||
"""A window around the playhead makes the track showable long before it is complete, and the
|
||
generation changes when it grows so the browser refetches instead of keeping the short VTT."""
|
||
pool = _manual_pool(monkeypatch)
|
||
key = subtitles.cache_key("98614", 0)
|
||
|
||
def _publish(rating_key, sub_ord, window_start):
|
||
if window_start is not None:
|
||
subtitles.publish(
|
||
key, _vtt(_cue("59:54.187", "59:58.650", "window")), (3570.0, 3900.0), False
|
||
)
|
||
else:
|
||
subtitles.publish(key, _vtt(_cue("01:12.000", "01:15.000", "whole")), (0.0, 1e9), True)
|
||
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", _publish)
|
||
|
||
first = client.get("/api/plex/subtitle/98614/0/status?at=3600").json()
|
||
assert first["state"] == "preparing" and first["gen"] == 0
|
||
|
||
pool.run_next() # the window job, submitted first — the full pass stays queued behind it
|
||
windowed = client.get("/api/plex/subtitle/98614/0/status?at=3600").json()
|
||
assert windowed["state"] == "partial", "a window around the playhead is showable"
|
||
assert windowed["gen"] > 0
|
||
# …and the <track> request now gets real WebVTT rather than another 202.
|
||
served = client.get("/api/plex/subtitle/98614/0?at=3600")
|
||
assert served.status_code == 200 and "window" in served.text
|
||
|
||
pool.run_next() # the full pass, scheduled by that same first poll
|
||
done = client.get("/api/plex/subtitle/98614/0/status?at=3600").json()
|
||
assert done["state"] == "ready"
|
||
assert done["gen"] != windowed["gen"], "a widened track must not reuse the window's generation"
|
||
assert "whole" in client.get("/api/plex/subtitle/98614/0").text
|
||
|
||
|
||
def test_a_position_outside_the_extracted_window_is_not_called_showable(client, monkeypatch):
|
||
"""Honest states: cues exist, but none for where the viewer just seeked to. Saying "ready" there
|
||
would put an empty track on screen with no explanation — the original complaint."""
|
||
_manual_pool(monkeypatch)
|
||
subtitles.publish(
|
||
subtitles.cache_key("98614", 0),
|
||
_vtt(_cue("00:10.000", "00:12.000", "x")),
|
||
(0.0, 330.0),
|
||
False,
|
||
)
|
||
assert client.get("/api/plex/subtitle/98614/0/status?at=100").json()["state"] == "partial"
|
||
assert client.get("/api/plex/subtitle/98614/0/status?at=5000").json()["state"] == "preparing"
|
||
|
||
|
||
def test_an_unshowable_track_says_why_and_maps_to_a_status(client, monkeypatch):
|
||
pool = _manual_pool(monkeypatch)
|
||
|
||
def _bitmap(rating_key, sub_ord, window_start):
|
||
raise subtitles.SubtitleUnavailable("bitmap")
|
||
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", _bitmap)
|
||
assert client.get("/api/plex/subtitle/98614/0/status").status_code == 200
|
||
pool.run_all()
|
||
st = client.get("/api/plex/subtitle/98614/0/status").json()
|
||
assert st["state"] == "failed" and st["code"] == "bitmap"
|
||
resp = client.get("/api/plex/subtitle/98614/0")
|
||
assert resp.status_code == 415 # a structured detail, so the UI can translate it
|
||
assert resp.json()["detail"] == {"code": "subtitle", "reason": "bitmap"}
|
||
|
||
|
||
# --- the ffmpeg pass itself ---------------------------------------------------------------------
|
||
|
||
|
||
class _FakeFfmpeg:
|
||
"""Stands in for the ffmpeg child: writes the outputs its argv asked for, exits with `returncode`."""
|
||
|
||
def __init__(self, cmd: list[str], returncode: int) -> None:
|
||
self.cmd, self.returncode = cmd, returncode
|
||
|
||
def communicate(self, timeout=None):
|
||
for i, arg in enumerate(self.cmd):
|
||
if arg == "-y":
|
||
Path(self.cmd[i + 1]).write_text(
|
||
_vtt(_cue("00:01.000", "00:02.000", "hi")), "utf-8"
|
||
)
|
||
return b"", b""
|
||
|
||
def kill(self) -> None:
|
||
pass
|
||
|
||
|
||
def _capture_ffmpeg(monkeypatch, returncode: int = 0) -> list[list[str]]:
|
||
"""Record the argv of each ffmpeg run and satisfy it by writing the outputs it asked for."""
|
||
runs: list[list[str]] = []
|
||
|
||
def _fake_popen(cmd, **kwargs):
|
||
runs.append(cmd)
|
||
return _FakeFfmpeg(cmd, returncode)
|
||
|
||
monkeypatch.setattr(subtitles.subprocess, "Popen", _fake_popen)
|
||
return runs
|
||
|
||
|
||
def test_one_ffmpeg_run_extracts_every_embedded_text_track(monkeypatch):
|
||
"""The read is the cost and it is SHARED. Extracting one track at a time is what made picking
|
||
Hungarian cost a full pass and then switching to English cost another one."""
|
||
runs = _capture_ffmpeg(monkeypatch)
|
||
out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0, 1], None, None)
|
||
assert len(runs) == 1, "each extra ffmpeg run is another full read of the file"
|
||
assert set(out) == {0, 1} and clean
|
||
assert runs[0].count("-map") == 2
|
||
|
||
|
||
def test_a_non_zero_exit_is_reported_even_when_it_left_output(monkeypatch):
|
||
"""ffmpeg can die partway and still leave a VTT behind. That text is worth keeping — but the
|
||
caller has to be TOLD, or it publishes a truncated track as the whole thing."""
|
||
_capture_ffmpeg(monkeypatch, returncode=1)
|
||
out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
|
||
assert out and not clean
|
||
|
||
|
||
def test_a_windowed_pass_bounds_the_read_and_never_uses_copyts(monkeypatch):
|
||
"""`-copyts` is the trap: with it ffmpeg IGNORES -t and reads the whole container anyway
|
||
(measured — a [0, 600 s] window still took the full 145 s). So the window is extracted on a
|
||
zero-based clock and shifted back to absolute time by the caller."""
|
||
runs = _capture_ffmpeg(monkeypatch)
|
||
subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], 3570.0, 330.0)
|
||
cmd = runs[0]
|
||
assert "-copyts" not in cmd
|
||
assert cmd[cmd.index("-ss") + 1].startswith(
|
||
"3570"
|
||
) # input seek — before -i, so it is a real seek
|
||
assert cmd.index("-ss") < cmd.index("-i")
|
||
assert cmd[cmd.index("-t") + 1].startswith("330")
|
||
|
||
|
||
def test_the_full_pass_is_not_bounded(monkeypatch):
|
||
runs = _capture_ffmpeg(monkeypatch)
|
||
subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
|
||
assert "-t" not in runs[0] and "-ss" not in runs[0]
|
||
|
||
|
||
# --- the pure pieces the above rests on ---------------------------------------------------------
|
||
|
||
|
||
def test_merge_vtt_unions_windows_in_time_order():
|
||
late = _vtt(_cue("59:54.187", "59:58.650", "late"))
|
||
early = _vtt(_cue("01:12.000", "01:15.000", "early"), _cue("59:54.187", "59:58.650", "late"))
|
||
merged = subtitles.merge_vtt(late, early)
|
||
assert merged.startswith("WEBVTT\n\n")
|
||
assert merged.count("-->") == 2, "the shared region must not be duplicated"
|
||
assert merged.index("early") < merged.index("late")
|
||
|
||
|
||
def test_covered_ranges_coalesce_and_answer_membership():
|
||
ranges = subtitles.merge_ranges([[300.0, 600.0], [0.0, 330.0], [900.0, 1200.0]])
|
||
assert ranges == [[0.0, 600.0], [900.0, 1200.0]]
|
||
assert subtitles.covers(ranges, 450.0)
|
||
assert not subtitles.covers(ranges, 700.0)
|
||
|
||
|
||
def test_a_cached_vtt_from_an_older_build_counts_as_complete(tmp_path):
|
||
"""Before this change the cache only ever held whole tracks, and none of them has a sidecar.
|
||
Reading one as "covers nothing" would re-extract every already-cached track after the upgrade."""
|
||
key = subtitles.cache_key("98614", 0)
|
||
(tmp_path / f"{key}.vtt").write_text(_vtt(_cue("00:01.000", "00:02.000", "x")), "utf-8")
|
||
assert subtitles.read_meta(key) == ([[0.0, 1_000_000_000.0]], True)
|
||
|
||
|
||
# --- the deep-review round: the seams between the window and the full pass ----------------------
|
||
|
||
|
||
def test_a_window_and_the_full_pass_do_not_double_a_cue():
|
||
"""THE doubled-subtitle bug. The two passes spell the same timestamp differently — a window comes
|
||
back through shift_vtt (always HH:MM:SS.mmm), a full pass is ffmpeg's own output, whose webvtt
|
||
muxer drops the hour below one hour — so de-duplicating on the text kept both, and every cue of a
|
||
window in the first hour was on screen twice."""
|
||
window = subtitles.shift_vtt(_vtt(_cue("00:30.123", "00:32.000", "hello")), -570.0)
|
||
full = _vtt(_cue("10:00.123", "10:02.000", "hello")) # ffmpeg's spelling of the same cue
|
||
merged = subtitles.merge_vtt(window, full)
|
||
assert merged.count("-->") == 1, merged
|
||
assert "00:10:00.123 --> 00:10:02.000" in merged # one canonical spelling in the served file
|
||
|
||
|
||
def test_two_different_cues_at_the_same_time_are_both_kept():
|
||
"""De-duplication is by timing AND text: a sign and a line of dialogue can share a timestamp."""
|
||
both = _vtt(_cue("00:10.000", "00:12.000", "sign"), _cue("00:10.000", "00:12.000", "dialogue"))
|
||
merged = subtitles.merge_vtt(both, both)
|
||
assert merged.count("-->") == 2 and "sign" in merged and "dialogue" in merged
|
||
|
||
|
||
class _NullSession:
|
||
def __enter__(self):
|
||
return None
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
|
||
# A file's subtitle streams as Plex lists them: a downloaded sidecar first (no `index` — it is not in
|
||
# the file), then the embedded ones, one of them a bitmap track that can never become text.
|
||
_STREAMS = [
|
||
{"key": "/library/streams/1", "codec": "srt"}, # sub_ord 0 — sidecar
|
||
{"index": 3, "codec": "subrip"}, # sub_ord 1 — embedded ordinal 0 (Hungarian)
|
||
{"index": 4, "codec": "pgs"}, # sub_ord 2 — embedded ordinal 1, bitmap
|
||
{"index": 5, "codec": "subrip"}, # sub_ord 3 — embedded ordinal 2 (English)
|
||
]
|
||
|
||
|
||
def _stub_item(monkeypatch, *, clean: bool = True, cues=("00:05.000", "00:07.000")):
|
||
"""Everything around `_extract_and_publish` stubbed, so the function itself is what runs."""
|
||
monkeypatch.setattr(subtitles, "SessionLocal", _NullSession)
|
||
monkeypatch.setattr(
|
||
subtitles, "_text_streams", lambda db, rk: (SimpleNamespace(file_path="/f.mkv"), _STREAMS)
|
||
)
|
||
monkeypatch.setattr(
|
||
subtitles.plex_paths, "local_media_path", lambda db, fp: "/plex-media/f.mkv"
|
||
)
|
||
monkeypatch.setattr(
|
||
subtitles, "_sidecar_vtt", lambda db, s: _vtt(_cue("00:01.000", "00:03.000", "side"))
|
||
)
|
||
calls: list[tuple] = []
|
||
|
||
def _extract(src, emb_ords, start, length):
|
||
calls.append((list(emb_ords), start, length))
|
||
return {o: _vtt(_cue(*cues, f"track {o}")) for o in emb_ords}, clean
|
||
|
||
monkeypatch.setattr(subtitles, "_ffmpeg_extract", _extract)
|
||
return calls
|
||
|
||
|
||
def test_one_window_publishes_every_embedded_text_track_at_absolute_time(monkeypatch):
|
||
"""What `_extract_and_publish` itself is for — every other test stubs it out. One read over the
|
||
file's EMBEDDED text streams (the sidecar is not in the file and must not shift the `-map`
|
||
ordinals; the bitmap track is skipped), each published under its own sub_ord, and a window's
|
||
zero-based cues shifted back to where they sit in the film."""
|
||
calls = _stub_item(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, 3570.0)
|
||
assert calls == [([0, 2], 3570.0, 330.0)], "one ffmpeg run over embedded ordinals 0 and 2"
|
||
english = subtitles.cache_read(subtitles.cache_key("98614", 3)) or ""
|
||
assert "00:59:35.000 --> 00:59:37.000" in english, english # 3570 + 5 s
|
||
assert "track 2" in english
|
||
assert "track 0" in (subtitles.cache_read(subtitles.cache_key("98614", 1)) or "")
|
||
for never in (0, 2): # the sidecar and the bitmap track come from nowhere in this pass
|
||
assert subtitles.cache_read(subtitles.cache_key("98614", never)) is None
|
||
|
||
|
||
def test_a_full_pass_that_died_is_not_cached_as_the_whole_track(monkeypatch):
|
||
"""A truncated full pass used to be published as complete: `ensure` answered "ready" forever, the
|
||
player stopped polling, and — the cache has no eviction — the subtitles simply ended wherever
|
||
ffmpeg had died, in every later session too. Now it covers what it provably reached; past that
|
||
a window is tried, and only when that fails too is the position answered as unavailable."""
|
||
_stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000"))
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, None)
|
||
key = subtitles.cache_key("98614", 3)
|
||
assert subtitles.read_meta(key) == ([[0.0, 2400.0]], False)
|
||
assert subtitles.ensure("98614", 3, at=100.0)["state"] == "partial"
|
||
assert subtitles.ensure("98614", 3, at=3000.0)["state"] == "preparing"
|
||
assert [a[3] for _, a in pool.jobs] == [2970.0], "a window past the break — not the full pass"
|
||
_window_raises(monkeypatch)
|
||
pool.run_all()
|
||
after = subtitles.ensure("98614", 3, at=3000.0)
|
||
assert (after["state"], after["code"]) == ("unavailable", "extract_failed")
|
||
|
||
|
||
def test_a_retried_full_pass_resumes_where_every_track_already_reaches(monkeypatch):
|
||
"""A pass that ran out of time keeps what it reached, but the retry an hour later started again
|
||
at 0 — so a file too big to read inside `EXTRACT_TIMEOUT_S` never completed. It now starts at
|
||
the end of the span every track has from the start (less `WINDOW_PRE_S`, so a cue across the
|
||
join is read whole), and finishing from there completes the track."""
|
||
calls = _stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000"))
|
||
_manual_pool(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, None)
|
||
assert calls[-1] == ([0, 2], None, None), "the first pass reads from the head"
|
||
calls = _stub_item(monkeypatch, clean=True, cues=("00:05.000", "00:07.000"))
|
||
subtitles._extract_and_publish("98614", 3, None)
|
||
assert calls == [([0, 2], 2370.0, None)]
|
||
for sub_ord in (1, 3):
|
||
assert subtitles.read_meta(subtitles.cache_key("98614", sub_ord))[1] is True
|
||
assert "00:39:35.000 --> 00:39:37.000" in (
|
||
subtitles.cache_read(subtitles.cache_key("98614", 3)) or ""
|
||
), "the resumed read's cues sit at absolute time (2370 + 5 s)"
|
||
|
||
|
||
def test_a_full_pass_starts_at_the_head_while_any_track_lacks_it(monkeypatch):
|
||
"""One read feeds every embedded track, so the least-covered one decides: a window elsewhere in
|
||
the film, or one track with the head and another without, still means reading from 0."""
|
||
calls = _stub_item(monkeypatch)
|
||
_manual_pool(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, 3570.0) # a window mid-film, both tracks
|
||
subtitles.publish(subtitles.cache_key("98614", 1), _vtt(), (0.0, 600.0), False)
|
||
subtitles._extract_and_publish("98614", 3, None)
|
||
assert calls[-1] == ([0, 2], None, None)
|
||
|
||
|
||
def test_the_status_says_whether_the_track_is_still_growing(monkeypatch):
|
||
"""`partial` alone kept the player polling every 2 s for the rest of the film once the full
|
||
pass had failed and backed off — a state that never reaches `ready` on its own."""
|
||
_stub_item(monkeypatch, clean=True, cues=("00:05.000", "00:07.000"))
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, 0.0)
|
||
assert subtitles.ensure("98614", 3, at=10.0)["growing"] is True
|
||
pool.jobs.clear()
|
||
_stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000"))
|
||
subtitles._extract_and_publish("98614", 3, None)
|
||
assert subtitles.ensure("98614", 3, at=100.0) == {
|
||
"state": "partial",
|
||
"code": None,
|
||
"gen": subtitles.generation(subtitles.cache_key("98614", 3)),
|
||
"growing": False,
|
||
}
|
||
|
||
|
||
def test_polls_that_follow_the_playhead_join_one_window(monkeypatch):
|
||
"""The poll sends the LIVE position, so it moves a second per poll. A window keyed on that
|
||
position was a new job every second — another read of the file per poll until the first landed."""
|
||
pool = _manual_pool(monkeypatch)
|
||
for at in (600.0, 601.0, 602.0, 603.0, 604.0):
|
||
subtitles.ensure("98614", 1, at=at)
|
||
assert len(pool.jobs) == 2, "one window + the one full pass"
|
||
|
||
|
||
def test_a_window_for_a_different_stretch_does_not_wait_behind_the_first(monkeypatch):
|
||
"""Measured on localdev: after a resume the first poll can still report 0 (the position is only
|
||
known once playback ticks), so the first window is cut around the wrong spot. The poll that
|
||
follows, at the real position, must get its own window at once — not queue behind that one."""
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles.ensure("98614", 1, at=0.0)
|
||
subtitles.ensure("98614", 1, at=1198.0)
|
||
assert sorted(a[3] for _, a in pool.jobs if a[3] is not None) == [0.0, 1168.0]
|
||
|
||
|
||
def test_a_queued_window_that_was_covered_meanwhile_reads_nothing(monkeypatch):
|
||
"""Every embedded track rides each pass over its file, so by the time a queued window for one
|
||
language runs, another language's window may already have published it."""
|
||
pool = _manual_pool(monkeypatch)
|
||
calls = _stub_item(monkeypatch)
|
||
subtitles.ensure("98614", 3, at=600.0)
|
||
subtitles.publish(
|
||
subtitles.cache_key("98614", 3),
|
||
_vtt(_cue("10:00.000", "10:02.000", "x")),
|
||
(570.0, 900.0),
|
||
False,
|
||
)
|
||
pool.run_next() # the window job
|
||
assert calls == []
|
||
|
||
|
||
def test_a_sidecar_does_not_wait_behind_an_embedded_full_pass(monkeypatch):
|
||
"""The full pass is one job per ITEM (the embedded fan-out shares it). A sidecar picked while an
|
||
embedded track's pass was running used to find that identity taken and wait out the whole read —
|
||
minutes — for a track that is one small fetch."""
|
||
pool = _manual_pool(monkeypatch)
|
||
_stub_item(monkeypatch)
|
||
subtitles._INFLIGHT[("full", "98614")] = None # an embedded track's full pass is running
|
||
subtitles.ensure("98614", 0, at=600.0)
|
||
pool.run_all()
|
||
assert subtitles.ensure("98614", 0, at=600.0)["state"] == "ready"
|
||
|
||
|
||
def test_publishes_into_one_track_do_not_lose_each_other(monkeypatch):
|
||
"""`publish` is a read-merge-write. Two jobs publishing into the same track at once — the window
|
||
and the full pass both fan out to every embedded track — used to let the later writer drop the
|
||
earlier one's cues. Forced interleaving: the first publisher pauses right after its read."""
|
||
key = subtitles.cache_key("98614", 0)
|
||
real_read = subtitles.cache_read
|
||
paused, resume = threading.Event(), threading.Event()
|
||
reads = {"n": 0}
|
||
|
||
def _read(k):
|
||
reads["n"] += 1
|
||
out = real_read(k)
|
||
if reads["n"] == 1:
|
||
paused.set()
|
||
resume.wait(timeout=5)
|
||
return out
|
||
|
||
monkeypatch.setattr(subtitles, "cache_read", _read)
|
||
first = threading.Thread(
|
||
target=subtitles.publish,
|
||
args=(key, _vtt(_cue("00:10.000", "00:12.000", "alpha")), (0.0, 330.0), False),
|
||
)
|
||
first.start()
|
||
assert paused.wait(timeout=5)
|
||
second = threading.Thread(
|
||
target=subtitles.publish,
|
||
args=(key, _vtt(_cue("10:00.000", "10:02.000", "bravo")), (570.0, 900.0), False),
|
||
)
|
||
second.start()
|
||
time.sleep(0.3) # without the lock, the second publisher finishes in here
|
||
resume.set()
|
||
first.join(timeout=5)
|
||
second.join(timeout=5)
|
||
merged = real_read(key) or ""
|
||
assert "alpha" in merged and "bravo" in merged, merged
|
||
assert subtitles.read_meta(key)[0] == [[0.0, 330.0], [570.0, 900.0]]
|
||
|
||
|
||
def test_shutdown_kills_a_running_extraction(monkeypatch):
|
||
"""The pool's workers are non-daemon threads that Python joins at exit, so one still inside a
|
||
full pass held a restarting process open for the rest of the read. Stand-in child: a real
|
||
process that would sleep for a minute."""
|
||
real_popen = subprocess.Popen
|
||
monkeypatch.setattr(
|
||
subtitles.subprocess,
|
||
"Popen",
|
||
lambda cmd, **kw: real_popen([sys.executable, "-c", "import time; time.sleep(60)"], **kw),
|
||
)
|
||
_real_pools(monkeypatch)
|
||
finished = threading.Event()
|
||
|
||
def _extract():
|
||
try:
|
||
subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
|
||
except subtitles.SubtitleUnavailable:
|
||
pass # killed before it wrote anything — exactly what a shutdown does
|
||
finally:
|
||
finished.set()
|
||
|
||
threading.Thread(target=_extract, daemon=True).start()
|
||
deadline = time.monotonic() + 10
|
||
while not subtitles._PROCS and time.monotonic() < deadline:
|
||
time.sleep(0.02)
|
||
assert subtitles._PROCS, "the extraction never registered its child"
|
||
subtitles.shutdown()
|
||
assert finished.wait(timeout=10), "the extraction was still running after shutdown"
|
||
|
||
|
||
def test_the_track_fetch_serves_what_is_cached_and_schedules_nothing(client, monkeypatch):
|
||
"""The <track> URL carries no position, and it used to run the scheduler anyway — judging
|
||
readiness at position 0. A viewer resumed an hour in got an unrequested extraction over the head
|
||
of the file, and with a failure memo alive the fetch refused a window that WAS on disk while the
|
||
status poll (which knows the position) called the track showable: silently blank subtitles."""
|
||
pool = _manual_pool(monkeypatch)
|
||
assert client.get("/api/plex/subtitle/98614/0").status_code == 202
|
||
assert pool.jobs == [], "the fetch scheduled an extraction"
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles.publish(key, _vtt(_cue("59:54.187", "59:58.650", "window")), (3570.0, 3900.0), False)
|
||
subtitles._fail(key, "extract_failed") # the full pass behind that window failed
|
||
resp = client.get("/api/plex/subtitle/98614/0")
|
||
assert resp.status_code == 200 and "window" in resp.text
|
||
assert pool.jobs == []
|
||
|
||
|
||
@pytest.mark.parametrize("at", ["inf", "-1", "nan"])
|
||
def test_an_impossible_position_is_rejected(client, monkeypatch, at):
|
||
"""`?at=inf` reached the scheduler's arithmetic and came back as a 500 with a traceback."""
|
||
pool = _manual_pool(monkeypatch)
|
||
assert client.get(f"/api/plex/subtitle/98614/0/status?at={at}").status_code == 422
|
||
assert pool.jobs == []
|
||
|
||
|
||
@pytest.mark.parametrize("q", ["offset=-inf", "offset=nan", "offset=inf"])
|
||
def test_an_impossible_offset_is_rejected(client, q):
|
||
"""The sibling of `at`: `-inf` crashed the shift with a 500, `nan` stacked every cue at 0."""
|
||
assert client.get(f"/api/plex/subtitle/98614/0?{q}").status_code == 422
|
||
|
||
|
||
def test_an_absurd_position_is_rejected_before_it_reaches_ffmpeg(client, monkeypatch):
|
||
"""ffmpeg refuses an `-ss` past its int64 microseconds, and that failure was recorded against the
|
||
TRACK — every other viewer at an uncovered position was told "failed" for a minute."""
|
||
pool = _manual_pool(monkeypatch)
|
||
assert client.get("/api/plex/subtitle/98614/0/status?at=1e13").status_code == 422
|
||
assert pool.jobs == []
|
||
|
||
|
||
# --- the second deep-review round ---------------------------------------------------------------
|
||
|
||
|
||
def _window_raises(monkeypatch):
|
||
"""From here on every window read fails outright; a full pass still works."""
|
||
real = subtitles._ffmpeg_extract
|
||
|
||
def _extract(src, emb_ords, start, length):
|
||
if length is not None:
|
||
raise subtitles.SubtitleUnavailable("extract_failed")
|
||
return real(src, emb_ords, start, length)
|
||
|
||
monkeypatch.setattr(subtitles, "_ffmpeg_extract", _extract)
|
||
|
||
|
||
def _later(monkeypatch, seconds: float) -> None:
|
||
"""Move the module's clock forward (the failure memos are monotonic-time TTLs)."""
|
||
now = time.monotonic() + seconds
|
||
monkeypatch.setattr(subtitles.time, "monotonic", lambda: now)
|
||
|
||
|
||
def test_a_full_pass_that_failed_is_not_read_again_while_its_output_is_watched(monkeypatch):
|
||
"""Published as partial, a failed full pass left the viewer inside what it DID publish — so the
|
||
status said `partial`, the poll kept going, and each time the one-minute failure memo lapsed it
|
||
re-armed a read of the whole container. A failure that reproduces did so every time: a full
|
||
read of the file every few minutes for as long as they watched."""
|
||
_stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000"))
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, None)
|
||
_later(monkeypatch, subtitles.FAIL_TTL_S + 1)
|
||
for _ in range(3):
|
||
assert subtitles.ensure("98614", 3, at=100.0)["state"] == "partial"
|
||
assert pool.jobs == [], "a failed full pass was queued again"
|
||
|
||
|
||
def test_a_failed_window_does_not_end_the_poll_while_the_full_pass_may_cover_it(monkeypatch):
|
||
""" "failed" is final — the player stops asking. A window that failed used to set the TRACK's
|
||
failure memo, so a viewer anywhere outside the covered part got "failed" and never heard about
|
||
the full pass that landed a minute later."""
|
||
pool = _manual_pool(monkeypatch)
|
||
_stub_item(monkeypatch)
|
||
_window_raises(monkeypatch)
|
||
subtitles.ensure("98614", 3, at=3000.0)
|
||
pool.run_next() # the window fails; the full pass is still queued behind it
|
||
assert subtitles.ensure("98614", 3, at=3000.0)["state"] == "preparing"
|
||
assert subtitles.ensure("98614", 3, at=500.0)["state"] == "preparing"
|
||
assert [a[3] for _, a in pool.jobs] == [None, 470.0], "the failed stretch was not tried again"
|
||
|
||
|
||
def test_a_backward_seek_gets_its_own_window(monkeypatch):
|
||
"""A window bucketed by position joined a seek back to 6:40 onto a running window at 9:10 that
|
||
could not cover it — the right one waited for the wrong one."""
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles.ensure("98614", 1, at=580.0)
|
||
subtitles.ensure("98614", 1, at=400.0)
|
||
assert sorted(a[3] for _, a in pool.jobs if a[3] is not None) == [370.0, 550.0]
|
||
|
||
|
||
def test_a_playhead_crossing_a_bucket_edge_joins_the_window_that_covers_it(monkeypatch):
|
||
"""…and the same bucketing queued a second read when a playhead crossed an edge while the first
|
||
window — which already covered it — was still running."""
|
||
pool = _manual_pool(monkeypatch)
|
||
for at in (628.0, 629.0, 630.0, 631.0, 632.0):
|
||
subtitles.ensure("98614", 1, at=at)
|
||
assert len(pool.jobs) == 2, "one window + the one full pass"
|
||
|
||
|
||
def test_a_scrubbing_viewer_cannot_queue_a_window_per_stop(monkeypatch):
|
||
"""The pool's queue is unbounded and every job is a read of the library."""
|
||
pool = _manual_pool(monkeypatch)
|
||
for at in (0.0, 1000.0, 2000.0, 3000.0, 4000.0):
|
||
subtitles.ensure("98614", 1, at=at)
|
||
assert len([1 for _, a in pool.jobs if a[3] is not None]) == subtitles._MAX_TRACK_WINDOWS
|
||
|
||
|
||
def test_nothing_more_is_queued_while_the_extractor_is_saturated(monkeypatch):
|
||
pool = _manual_pool(monkeypatch)
|
||
for rk in range(20): # twenty viewers, so no one account's share is what stops the passes
|
||
subtitles.ensure(str(rk), 0, at=0.0, viewer=rk)
|
||
windows = [a for _, a in pool.jobs if a[3] is not None]
|
||
assert len(windows) == subtitles._MAX_WINDOWS_INFLIGHT
|
||
assert len(pool.jobs) - len(windows) == subtitles._MAX_FULL_INFLIGHT
|
||
|
||
|
||
def test_a_window_starts_where_ffmpeg_is_told_to_seek(monkeypatch):
|
||
"""ffmpeg gets `-ss` to the millisecond, and its cues come back on THAT zero. Shifted back by an
|
||
unrounded start they landed up to half a millisecond off the full pass's — spelled differently,
|
||
and so kept twice by the merge."""
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles.ensure("98614", 1, at=1198.4375)
|
||
(window_start,) = [a[3] for _, a in pool.jobs if a[3] is not None]
|
||
assert window_start == 1168.438
|
||
|
||
|
||
def test_a_millisecond_rounded_up_carries_into_the_minute():
|
||
"""`00:00:60.000` is not a WebVTT timestamp — the browser drops the cue."""
|
||
assert subtitles._vtt_s_to_ts(59.9996) == "00:01:00.000"
|
||
assert subtitles._vtt_s_to_ts(3599.9996) == "01:00:00.000"
|
||
shifted = subtitles.shift_vtt(_vtt(_cue("00:01:00.000", "00:01:02.000", "x")), 0.0004)
|
||
assert "00:01:00.000 --> 00:01:02.000" in shifted, shifted
|
||
|
||
|
||
def test_ffmpegs_spelling_of_a_negative_time_reads_exactly():
|
||
"""With `-avoid_negative_ts disabled` a cue already running at the window's start comes out
|
||
before zero, and ffmpeg's muxer puts the sign on every field."""
|
||
assert subtitles._vtt_ts_to_s("00:-5.-05") == pytest.approx(-5.005)
|
||
assert subtitles._vtt_ts_to_s("-1:-5.-300") == pytest.approx(-65.3)
|
||
|
||
|
||
def test_dying_between_the_two_renames_cannot_pass_a_window_off_as_the_whole_track(monkeypatch):
|
||
"""A VTT without a sidecar is how an older build cached a COMPLETE track. A publish is two
|
||
renames, the VTT first — so dying between them on a key's first publish left a window that
|
||
read as complete, "ready" for good."""
|
||
key = subtitles.cache_key("98614", 0)
|
||
real_write = subtitles.cache_write
|
||
|
||
def _die_after(k, vtt):
|
||
real_write(k, vtt)
|
||
raise KeyboardInterrupt # the process goes away right after the VTT landed
|
||
|
||
monkeypatch.setattr(subtitles, "cache_write", _die_after)
|
||
with pytest.raises(KeyboardInterrupt):
|
||
subtitles.publish(key, _vtt(_cue("10:00.000", "10:02.000", "x")), (570.0, 900.0), False)
|
||
assert subtitles.read_meta(key) == ([], False)
|
||
|
||
|
||
def test_shutdown_drops_queued_extractions(monkeypatch):
|
||
"""Only the kill half used to be pinned. A job still QUEUED at shutdown would otherwise start
|
||
after the kill sweep — a fresh read of the file holding the process open."""
|
||
pool = ThreadPoolExecutor(max_workers=1)
|
||
monkeypatch.setattr(subtitles, "_WIN_POOL", pool)
|
||
monkeypatch.setattr(subtitles, "_FULL_POOL", pool)
|
||
busy, release, ran = threading.Event(), threading.Event(), threading.Event()
|
||
pool.submit(lambda: (busy.set(), release.wait(timeout=5)))
|
||
pool.submit(ran.set)
|
||
assert busy.wait(timeout=5)
|
||
subtitles.shutdown()
|
||
release.set()
|
||
pool.shutdown(wait=True)
|
||
assert not ran.is_set(), "a queued extraction ran after shutdown"
|
||
|
||
|
||
def test_a_worker_that_reaches_ffmpeg_after_shutdown_never_starts_it(monkeypatch):
|
||
"""A worker still in its Plex lookup when the kill sweep ran used to start its ffmpeg anyway."""
|
||
_real_pools(monkeypatch)
|
||
started: list = []
|
||
monkeypatch.setattr(subtitles.subprocess, "Popen", lambda cmd, **kw: started.append(cmd))
|
||
subtitles.shutdown()
|
||
with pytest.raises(subtitles.SubtitleUnavailable):
|
||
subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
|
||
assert started == []
|
||
|
||
|
||
def test_the_app_stops_the_extractor_when_it_shuts_down(monkeypatch):
|
||
"""No test ran the lifespan, so deleting this one call switched all of the above off with every
|
||
lane green."""
|
||
|
||
class _Configured:
|
||
def __enter__(self):
|
||
return None
|
||
|
||
def __exit__(self, *exc):
|
||
return False
|
||
|
||
async def _no_sweep():
|
||
return None
|
||
|
||
calls: list[str] = []
|
||
monkeypatch.setattr(app_main, "SessionLocal", _Configured)
|
||
monkeypatch.setattr(app_main.state, "is_configured", lambda db: True)
|
||
monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep)
|
||
monkeypatch.setattr(app_main, "start_scheduler", lambda: None)
|
||
monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
|
||
monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: calls.append("stopped"))
|
||
|
||
async def _cycle():
|
||
async with app_main.lifespan(app):
|
||
pass
|
||
|
||
asyncio.run(_cycle())
|
||
assert calls == ["stopped"]
|
||
|
||
|
||
# --- a window against a REAL ffmpeg -------------------------------------------------------------
|
||
|
||
|
||
def _make_mkv(tmp_path: Path) -> Path:
|
||
"""A minute of video with a keyframe every 10 s and a subtitle every 2 s — so a seek to 21.3 s
|
||
lands on the keyframe at 20 s, with the 20.0–21.5 s cue already running at the window's start."""
|
||
srt = tmp_path / "s.srt"
|
||
srt.write_text(
|
||
"".join(
|
||
f"{i + 1}\n00:00:{2 * i:02d},000 --> 00:00:{2 * i + 1:02d},500\nline {i}\n\n"
|
||
for i in range(30)
|
||
),
|
||
"utf-8",
|
||
)
|
||
out = tmp_path / "f.mkv"
|
||
subprocess.run(
|
||
[
|
||
"ffmpeg", "-nostdin", "-loglevel", "error",
|
||
"-f", "lavfi", "-i", "color=c=black:s=32x32:r=10:d=60",
|
||
"-i", str(srt),
|
||
"-map", "0:v", "-map", "1:s", "-c:v", "mpeg4", "-g", "100", "-c:s", "srt",
|
||
"-y", str(out),
|
||
],
|
||
check=True,
|
||
) # fmt: skip
|
||
return out
|
||
|
||
|
||
@needs_ffmpeg
|
||
def test_a_window_lines_up_with_the_full_pass_on_a_real_file(tmp_path, monkeypatch):
|
||
"""THE late-and-doubled window. The input seek lands on the keyframe before the window's start,
|
||
a cue already running there comes out at a negative time, and by default ffmpeg's muxer shifted
|
||
the whole window so that cue sat at 0 — every cue late by the distance to the keyframe (1–5 s on
|
||
a real library file), and kept twice for good once the full pass merged in beside it."""
|
||
src = _make_mkv(tmp_path)
|
||
monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path / "cache")
|
||
full, clean = subtitles._ffmpeg_extract(str(src), [0], None, None)
|
||
assert clean
|
||
win, clean = subtitles._ffmpeg_extract(str(src), [0], 21.3, 20.0)
|
||
assert clean
|
||
key = subtitles.cache_key("1", 0)
|
||
subtitles.publish(key, subtitles.shift_vtt(win[0], -21.3), (21.3, 41.3), False)
|
||
served = subtitles.cache_read(key) or ""
|
||
assert "00:00:22.000 --> 00:00:23.500\nline 11" in served, served # on time, not 1.3 s late
|
||
subtitles.publish(key, full[0], (0.0, subtitles._FULL_END_S), True)
|
||
merged = subtitles.cache_read(key) or ""
|
||
assert merged.count("-->") == 30, "a window cue survived beside its full-pass twin"
|
||
|
||
|
||
@needs_ffmpeg
|
||
def test_a_full_pass_resumed_mid_file_lines_up_on_a_real_file(tmp_path, monkeypatch):
|
||
"""A retried full pass seeks like a window does but reads to the end (no `-t`): the cue already
|
||
running at its start must come out on time, not shifted to 0 with everything after it."""
|
||
src = _make_mkv(tmp_path)
|
||
monkeypatch.setattr(subtitles, "CACHE_DIR", tmp_path / "cache")
|
||
head, clean = subtitles._ffmpeg_extract(str(src), [0], None, 30.0)
|
||
assert clean
|
||
rest, clean = subtitles._ffmpeg_extract(str(src), [0], 21.3, None)
|
||
assert clean
|
||
key = subtitles.cache_key("1", 0)
|
||
subtitles.publish(key, head[0], (0.0, 30.0), False)
|
||
subtitles.publish(key, subtitles.shift_vtt(rest[0], -21.3), (21.3, subtitles._FULL_END_S), True)
|
||
merged = subtitles.cache_read(key) or ""
|
||
assert "00:00:22.000 --> 00:00:23.500\nline 11" in merged, merged
|
||
assert "00:00:58.000 --> 00:00:59.500\nline 29" in merged, merged
|
||
assert merged.count("-->") == 30, "a cue across the join was kept twice or lost"
|
||
|
||
|
||
# --- the third deep-review round ----------------------------------------------------------------
|
||
|
||
|
||
def test_a_transient_failure_is_said_but_is_not_a_final_answer(monkeypatch):
|
||
"""`failed` ends the player's poll and unmounts the track. A Plex hiccup memo (a minute) was given
|
||
as `failed`, so one blip ended the subtitles for the rest of the session — at cached positions
|
||
too, and after the full pass had completed behind it. It is `unavailable` now: said, asked again."""
|
||
pool = _manual_pool(monkeypatch)
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles.publish(key, _vtt(_cue("01:40.000", "01:42.000", "x")), (70.0, 400.0), False)
|
||
subtitles._fail(key, "plex_error")
|
||
here = subtitles.ensure("98614", 0, at=3000.0)
|
||
assert (here["state"], here["code"]) == ("unavailable", "plex_error")
|
||
assert here["gen"] > 0, "the client needs to know there are cues elsewhere to keep showing"
|
||
assert subtitles.ensure("98614", 0, at=100.0)["state"] == "partial"
|
||
assert pool.jobs == [], "nothing is re-read while the memo is alive"
|
||
_later(monkeypatch, subtitles.FAIL_TTL_S + 1)
|
||
assert subtitles.ensure("98614", 0, at=3000.0)["state"] == "preparing"
|
||
assert [a[3] for _, a in pool.jobs] == [2970.0, None], "asked again once the memo lapsed"
|
||
|
||
|
||
def test_the_next_window_is_read_ahead_of_its_edge(monkeypatch):
|
||
"""With the full pass backed off (or slower than a window lasts) a window was only asked for once
|
||
the playhead had LEFT coverage — the track unmounted at every edge for a poll plus a read."""
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles._FULL_FAILED["98614"] = time.monotonic()
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
|
||
assert subtitles.ensure("98614", 0, at=1200.0)["state"] == "partial"
|
||
assert pool.jobs == [], "far from the edge: nothing to read yet"
|
||
assert subtitles.ensure("98614", 0, at=1250.0)["state"] == "partial"
|
||
assert [a[3] for _, a in pool.jobs] == [1271.0], "the next window, before the edge is reached"
|
||
|
||
|
||
def test_a_window_that_died_partway_marks_the_stretch_it_did_not_reach(monkeypatch):
|
||
"""Without the span memo, a window that died before the playhead left it uncovered AND not
|
||
failed — so every poll queued the same read of the library again."""
|
||
_stub_item(monkeypatch, clean=False, cues=("00:10.000", "00:12.000"))
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles._extract_and_publish("98614", 3, 3570.0)
|
||
key = subtitles.cache_key("98614", 3)
|
||
assert subtitles.read_meta(key) == ([[3570.0, 3582.0]], False)
|
||
assert subtitles.ensure("98614", 3, at=3700.0)["state"] == "preparing"
|
||
assert [a[3] for _, a in pool.jobs] == [None], "the dead stretch was queued again"
|
||
|
||
|
||
def test_a_crashed_pass_is_recorded_like_a_failed_one(monkeypatch):
|
||
"""A timeout or a crash lands in `_run`'s catch-all. Recorded against the TRACK, a crashed window
|
||
told every viewer at any uncovered position "failed", and a crashed full pass re-armed a
|
||
whole-file read every minute instead of backing off."""
|
||
|
||
def _crash(rating_key, sub_ord, window_start):
|
||
raise RuntimeError("boom")
|
||
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", _crash)
|
||
key = subtitles.cache_key("98614", 3)
|
||
subtitles._run(("win", "98614", 3, 570.0), "98614", 3, 570.0, 600.0)
|
||
assert subtitles._window_failed(key, 600.0) and not subtitles._window_failed(key, 3000.0)
|
||
subtitles._run(("full", "98614"), "98614", 3, None, 600.0)
|
||
assert subtitles._full_failed("98614")
|
||
assert subtitles._recent_failure(key) is None, "a crash is not a verdict on the whole track"
|
||
|
||
|
||
def test_the_failure_memos_lapse(monkeypatch):
|
||
"""Both memos exist to stop a re-read, not to forbid one: permanent, a transient failure would
|
||
have held until the process restarted."""
|
||
pool = _manual_pool(monkeypatch)
|
||
key = subtitles.cache_key("98614", 3)
|
||
subtitles._pass_failed("98614", [key], None)
|
||
subtitles._pass_failed("98614", [key], 570.0)
|
||
assert subtitles.ensure("98614", 3, at=600.0)["state"] == "unavailable"
|
||
_later(monkeypatch, subtitles.FAIL_TTL_S + 1)
|
||
subtitles.ensure("98614", 3, at=600.0)
|
||
assert [a[3] for _, a in pool.jobs] == [570.0], "the window memo lapsed; the full one has not"
|
||
pool.jobs.clear()
|
||
subtitles._INFLIGHT.clear()
|
||
_later(monkeypatch, subtitles.FULL_RETRY_S + 1)
|
||
subtitles.ensure("98614", 3, at=600.0)
|
||
assert None in [a[3] for _, a in pool.jobs], "the full pass was never tried again"
|
||
|
||
|
||
def test_lapsed_memos_are_swept_not_kept_forever(monkeypatch):
|
||
"""They were pruned only when the SAME key was read again, and the status route takes any key:
|
||
polling fresh ones grew the API process without bound."""
|
||
for i in range(50):
|
||
subtitles._fail(f"sub_x{i}_0", "no_item")
|
||
subtitles._pass_failed(f"x{i}", [f"sub_x{i}_0"], None)
|
||
subtitles._pass_failed(f"x{i}", [f"sub_x{i}_0"], 0.0)
|
||
_later(monkeypatch, subtitles.FULL_RETRY_S + 1)
|
||
subtitles._fail("sub_y_0", "no_item")
|
||
assert list(subtitles._FAILURES) == ["sub_y_0"]
|
||
assert subtitles._FULL_FAILED == {} and subtitles._WIN_FAILED == {}
|
||
|
||
|
||
def test_a_window_is_not_refused_while_full_passes_hold_their_slots(monkeypatch):
|
||
"""One pool and one cap for both kinds of job: full passes (minutes each) held every worker and
|
||
every slot, and a seek window — seconds of work — waited behind them or was not queued at all."""
|
||
win, full = _ManualPool(), _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_WIN_POOL", win)
|
||
monkeypatch.setattr(subtitles, "_FULL_POOL", full)
|
||
for rk in range(subtitles._MAX_FULL_INFLIGHT):
|
||
subtitles._INFLIGHT[("full", str(rk))] = None
|
||
subtitles.ensure("98614", 1, at=600.0)
|
||
assert [a[3] for _, a in win.jobs] == [570.0]
|
||
assert full.jobs == [], "past its bound no full pass is queued"
|
||
|
||
|
||
def test_one_account_cannot_hold_every_full_pass_slot(monkeypatch):
|
||
"""The full-pass slots are global and minutes long each: one account opening subtitles on film
|
||
after film held them all, and nobody else's track ever completed. Past its share an account
|
||
still gets windows — what makes a track watchable now — and the next account still gets a pass."""
|
||
win, full = _ManualPool(), _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_WIN_POOL", win)
|
||
monkeypatch.setattr(subtitles, "_FULL_POOL", full)
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", lambda *a: None)
|
||
for rk in range(subtitles._MAX_FULL_PER_USER + 1):
|
||
subtitles.ensure(str(1000 + rk), 1, at=600.0, viewer=7)
|
||
assert len(full.jobs) == subtitles._MAX_FULL_PER_USER
|
||
assert len(win.jobs) == subtitles._MAX_FULL_PER_USER + 1, (
|
||
"the account over its share still gets windows"
|
||
)
|
||
subtitles.ensure("2000", 1, at=600.0, viewer=8)
|
||
assert len(full.jobs) == subtitles._MAX_FULL_PER_USER + 1, "another account is not shut out"
|
||
full.run_all() # the account's passes end: its share is free again
|
||
subtitles.ensure("3000", 1, at=600.0, viewer=7)
|
||
assert any(a[1] == "3000" and a[3] is None for _, a in full.jobs)
|
||
|
||
|
||
def test_a_poll_after_shutdown_queues_nothing(monkeypatch):
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles.shutdown()
|
||
subtitles.ensure("98614", 1, at=600.0)
|
||
assert pool.jobs == []
|
||
|
||
|
||
def test_a_job_dequeued_after_shutdown_reads_nothing(monkeypatch):
|
||
pool = _manual_pool(monkeypatch)
|
||
ran: list = []
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", lambda *a: ran.append(a))
|
||
subtitles.ensure("98614", 1, at=600.0)
|
||
subtitles.shutdown()
|
||
pool.run_all()
|
||
assert ran == []
|
||
|
||
|
||
def test_a_child_started_while_the_shutdown_sweep_runs_is_killed(monkeypatch):
|
||
"""The sweep can run between the pre-Popen check and the child's registration."""
|
||
killed: list = []
|
||
|
||
class _Child:
|
||
returncode = -9
|
||
|
||
def kill(self):
|
||
killed.append(True)
|
||
|
||
def communicate(self, timeout=None):
|
||
return b"", b""
|
||
|
||
def _popen(cmd, **kw):
|
||
subtitles._STOPPING = True # the shutdown sweep ran while this child was being started
|
||
return _Child()
|
||
|
||
monkeypatch.setattr(subtitles.subprocess, "Popen", _popen)
|
||
with pytest.raises(subtitles.SubtitleUnavailable):
|
||
subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
|
||
assert killed == [True]
|
||
|
||
|
||
def test_a_pass_that_times_out_keeps_what_it_reached(monkeypatch):
|
||
"""It was killed and its outputs deleted unread — so a file too big to read inside the timeout
|
||
never got past the head, and with the full-pass backoff the same doomed read came back hourly."""
|
||
real_popen = subprocess.Popen
|
||
cue = _vtt(_cue("00:05.000", "00:07.000", "reached"))
|
||
|
||
def _popen(cmd, **kw):
|
||
out = cmd[cmd.index("-y") + 1]
|
||
script = f"import time\nopen({out!r}, 'w').write({cue!r})\ntime.sleep(60)"
|
||
return real_popen([sys.executable, "-c", script], **kw)
|
||
|
||
monkeypatch.setattr(subtitles.subprocess, "Popen", _popen)
|
||
monkeypatch.setattr(subtitles, "EXTRACT_TIMEOUT_S", 1.0)
|
||
out, clean = subtitles._ffmpeg_extract("/plex-media/f.mkv", [0], None, None)
|
||
assert not clean, "a pass that ran out of time must not pass for a whole one"
|
||
assert "reached" in out[0]
|
||
|
||
|
||
def test_an_orphan_sidecar_is_not_merged_into(monkeypatch):
|
||
"""A sidecar with no VTT behind it claims cues that are not there. Merged into, it made one
|
||
window read as the whole track — `ready`, and the full pass skipped as already done."""
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles._write_meta(key, [[0.0, subtitles._FULL_END_S]], True)
|
||
subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
|
||
assert subtitles.read_meta(key) == ([[970.0, 1300.0]], False)
|
||
assert subtitles.ensure("98614", 0, at=5000.0)["state"] != "ready"
|
||
|
||
|
||
def test_an_unreadable_track_is_not_overwritten_by_one_window(monkeypatch):
|
||
"""`cache_read` answers None for a file it could not READ as well as for one that is not there.
|
||
Taken for absent, a whole track was replaced by the one window being published."""
|
||
key = subtitles.cache_key("98614", 0)
|
||
whole = _vtt(_cue("00:01.000", "00:02.000", "whole"))
|
||
subtitles.publish(key, whole, (0.0, subtitles._FULL_END_S), True)
|
||
monkeypatch.setattr(subtitles, "cache_read", lambda k: None) # an I/O error on the read
|
||
subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
|
||
assert "whole" in subtitles._vtt_path(key).read_text("utf-8")
|
||
|
||
|
||
def test_a_region_that_could_not_be_written_is_not_claimed(monkeypatch):
|
||
monkeypatch.setattr(subtitles, "cache_write", lambda key, vtt: False)
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (0.0, 1e9), True)
|
||
assert subtitles.read_meta(key) == ([], False)
|
||
|
||
|
||
def test_a_nul_in_the_rating_key_is_not_a_500(client, monkeypatch):
|
||
"""`%00` reached `Path.stat`, which raises ValueError rather than OSError — a 500 on every poll,
|
||
after the scheduler had already queued two jobs for a key that names nothing."""
|
||
_manual_pool(monkeypatch)
|
||
assert client.get("/api/plex/subtitle/%00/0/status").status_code == 200
|
||
|
||
|
||
def test_a_nul_rating_key_names_no_row():
|
||
"""Postgres refuses NUL in text, so looking it up was a DataError (a 500) on every Plex route."""
|
||
from app.models import PlexItem
|
||
from app.plex.lookups import by_rating_key
|
||
|
||
assert by_rating_key(None, PlexItem, "12\x0034") is None # type: ignore[arg-type]
|
||
|
||
|
||
# --- the fourth deep-review round ---------------------------------------------------------------
|
||
|
||
|
||
def test_a_pass_whose_output_could_not_be_kept_is_backed_off_like_a_failed_one(monkeypatch):
|
||
"""`publish` refusing (a full or read-only cache volume) returned quietly, so `_run` recorded
|
||
nothing and every 1 s poll queued the window AND the whole-container pass again — the file read
|
||
end to end over and over, the player on "preparing" for good."""
|
||
_stub_item(monkeypatch)
|
||
pool = _manual_pool(monkeypatch)
|
||
monkeypatch.setattr(subtitles, "cache_write", lambda key, vtt: False)
|
||
assert subtitles.ensure("98614", 3, at=600.0)["state"] == "preparing"
|
||
assert [a[3] for _, a in pool.jobs] == [570.0, None]
|
||
pool.run_all()
|
||
after = subtitles.ensure("98614", 3, at=600.0)
|
||
assert (after["state"], after["code"]) == ("unavailable", "extract_failed")
|
||
for _ in range(3):
|
||
subtitles.ensure("98614", 3, at=600.0)
|
||
assert pool.jobs == [], "the same reads were queued again"
|
||
|
||
|
||
def test_a_sidecar_that_could_not_be_kept_is_recorded_as_failed(monkeypatch):
|
||
_stub_item(monkeypatch)
|
||
pool = _manual_pool(monkeypatch)
|
||
monkeypatch.setattr(subtitles, "cache_write", lambda key, vtt: False)
|
||
subtitles.ensure("98614", 0, at=600.0)
|
||
pool.run_all()
|
||
subtitles.ensure("98614", 0, at=600.0)
|
||
assert pool.jobs == [], "a refused sidecar publish was queued again"
|
||
|
||
|
||
def test_an_unreadable_cached_track_is_said_once_and_not_rescheduled(monkeypatch):
|
||
"""A VTT on disk that cannot be read: `publish` will not merge into it and a job whose sidecar
|
||
covers the position skips as done, so scheduling could only ever leave the player on
|
||
"preparing"."""
|
||
pool = _manual_pool(monkeypatch)
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
|
||
monkeypatch.setattr(subtitles, "cache_read", lambda k: None) # an I/O error on the read
|
||
st = subtitles.ensure("98614", 0, at=1000.0)
|
||
assert (st["state"], st["code"], st["gen"]) == ("unavailable", "cache_unreadable", 0)
|
||
assert pool.jobs == []
|
||
|
||
|
||
def test_an_absurd_rating_key_or_ordinal_is_refused_not_a_500(client, monkeypatch):
|
||
"""They become a cache FILE name: a 250-digit key outgrew the OS name limit, and pathlib's
|
||
`exists()` re-raises that (ENAMETOOLONG) instead of answering False — a 500 on every poll."""
|
||
pool = _manual_pool(monkeypatch)
|
||
long_key = "9" * 250
|
||
for url in (
|
||
f"/api/plex/subtitle/{long_key}/0/status",
|
||
f"/api/plex/subtitle/98614/{long_key}/status",
|
||
f"/api/plex/subtitle/{long_key}/0",
|
||
):
|
||
assert client.get(url).status_code == 422, url
|
||
assert pool.jobs == []
|
||
|
||
# …and the cache helpers themselves no longer throw on such a name. Asserted against a forced
|
||
# ENAMETOOLONG rather than the real filesystem: on Linux pathlib re-raises it (the defect), but
|
||
# Windows reports ERROR_INVALID_NAME, which pathlib's `exists()` already swallows — so on a dev
|
||
# machine this assertion passed with `_exists` reverted and only the container lane was honest
|
||
# about it (deep-review round 5).
|
||
def _too_long(self):
|
||
raise OSError(errno.ENAMETOOLONG, "File name too long")
|
||
|
||
monkeypatch.setattr(Path, "exists", _too_long)
|
||
assert subtitles._exists(Path("whatever")) is False
|
||
assert subtitles.read_meta(subtitles.cache_key(long_key, 0)) == ([], False)
|
||
|
||
|
||
def test_a_read_ahead_window_that_failed_is_not_queued_again(monkeypatch):
|
||
"""Near a covered edge the probe moves just past it, and `_schedule` has no failed-span check of
|
||
its own: without the guard, every poll re-queued the same failing window read."""
|
||
pool = _manual_pool(monkeypatch)
|
||
subtitles._FULL_FAILED["98614"] = time.monotonic()
|
||
key = subtitles.cache_key("98614", 0)
|
||
subtitles.publish(key, _vtt(_cue("16:20.000", "16:22.000", "x")), (970.0, 1300.0), False)
|
||
subtitles._window_fail(key, (1271.0, 1601.0))
|
||
for at in (1250.0, 1252.0, 1254.0):
|
||
assert subtitles.ensure("98614", 0, at=at)["state"] == "partial"
|
||
assert pool.jobs == [], "the failed read-ahead window was queued again"
|