peter
473d9f02f2
Deep-review round 2 of the 2026-09 bug hunt, over the round-1 fix diff. - A window was LATE by the distance to the keyframe its input seek landed on, and then doubled for good once the full pass merged in: a cue already running at the window's start came out at a negative time, and the webvtt muxer's default shifted the whole output so it sat at 0. Measured on the real Westworld S01E06 at 25:00: every window cue +5.005 s, 682 cues merged instead of 608. `-avoid_negative_ts disabled` keeps the times exact (the parser reads ffmpeg's signed-field spelling of a negative time); now 608/608 on that file, and a real-ffmpeg test pins it. - A full pass that failed was re-armed every minute while the viewer watched what it did publish — a whole-container read per minute. It now backs off for an hour; windows keep running meanwhile. - A failed WINDOW set the track's failure memo, so any uncovered position read "failed" — a final answer that ends the player's poll, even with the full pass about to deliver, and one any account could plant with an absurd `at`. It now marks only the span it failed to read; "failed" is given once the full pass AND a window there have both failed. - Windows join by what they COVER, not a bucketed identity: a backward seek no longer waits behind a window that cannot cover it, and a playhead crossing a bucket edge no longer queues a second read. The queue is bounded per track and overall. - shutdown() sets a flag a worker checks before it spawns ffmpeg (one still in its Plex lookup used to start a read after the kill sweep); queued-job cancellation and the lifespan call are now tested. - A window starts where ffmpeg is told to seek (rounded to the ms it gets), a rounded-up millisecond carries into the minute (no more `00:00:60.000`), a key's first publish claims nothing before its VTT lands, and `at`/`offset`/`start`/`aoff` refuse non-finite values (`at` also an absurd one).
792 lines
35 KiB
Python
792 lines
35 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). Pure — no Postgres, no ffmpeg (the auth half is in
|
||
test_plex_subtitle_auth.py, which needs the DB).
|
||
"""
|
||
|
||
import asyncio
|
||
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 _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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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."""
|
||
pool = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
|
||
def _bitmap(rating_key, sub_ord, window_start):
|
||
raise subtitles.SubtitleUnavailable("bitmap")
|
||
|
||
monkeypatch.setattr(subtitles, "_extract_and_publish", _bitmap)
|
||
client.get("/api/plex/subtitle/98614/0/status")
|
||
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."""
|
||
(tmp_path / "sub_98614_0.vtt").write_text(_vtt(_cue("00:01.000", "00:02.000", "x")), "utf-8")
|
||
assert subtitles.read_meta("sub_98614_0") == ([[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 failed."""
|
||
_stub_item(monkeypatch, clean=False, cues=("39:58.000", "40:00.000"))
|
||
pool = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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"]) == ("failed", "extract_failed")
|
||
|
||
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
_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),
|
||
)
|
||
monkeypatch.setattr(subtitles, "_POOL", ThreadPoolExecutor(max_workers=1))
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
_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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
for rk in range(20):
|
||
subtitles.ensure(str(rk), 0, at=0.0)
|
||
assert len(pool.jobs) == subtitles._MAX_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 = _ManualPool()
|
||
monkeypatch.setattr(subtitles, "_POOL", pool)
|
||
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, "_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."""
|
||
monkeypatch.setattr(subtitles, "_POOL", ThreadPoolExecutor(max_workers=1))
|
||
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"
|