"""Subtitle sync: measure each sidecar against the copy's own audio (labs#13).
Measured 2026-10-04 on No Other Choice with ffsubsync, against speech
extracted once from the film: NOT ONE of 32 wyzie tracks matched the copy.
- 23 tracks: offset -8.9 s, scale 1.000 (another cut/intro)
- 7 tracks incl. en: offset -2.2 s, scale 1.001 (24 vs 23.976 fps -- drift
that grows to ~8 s over the film: the "keeps going out of sync" report)
- ar -20 s, de -18.5 s
Corrected tracks re-measured at 0.000-0.010 s / 1.000, cue count unchanged.
So, per sidecar:
in sync |offset| <= SUBSYNC_TOLERANCE and scale 1.000 -> untouched
fixable corrected copy re-measures in tolerance -> original to the store
(outside the library, never deleted), corrected takes its place
unfixable the correction does not re-measure clean (a different edit) ->
original moved to the store; the library no longer offers it
Results are recorded per file identity, so a re-run skips what it has done.
The two seams -- _subsync_reference (speech, once per copy) and
_subsync_measure (one track) -- are what the tests replace; no test runs
ffsubsync.
Run: .venv/bin/python3 -m pytest test_subtitle_sync.py -q
"""
import json
import os
import pytest
import addon
@pytest.fixture
def lib(tmp_path, monkeypatch):
monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path / "lib"))
monkeypatch.setattr(addon, "SUBS_ORIGINAL_DIR", str(tmp_path / "subs-original"))
monkeypatch.setattr(addon, "SUBTITLE_SYNC_FILE", str(tmp_path / "sync.json"))
monkeypatch.setattr(addon, "_subsync_available", lambda: True)
# The mechanics below use several languages; production syncs English
# only (SUBSYNC_LANGUAGES), which test_only_english_is_synced pins.
monkeypatch.setattr(addon, "SUBSYNC_LANGUAGES", None)
monkeypatch.setattr(addon, "_subsync_reference",
lambda dest, workdir: os.path.join(workdir, "ref.npz"))
notified = []
monkeypatch.setattr(addon, "_notify_jellyfin", lambda p, t: notified.append((p, t)))
dest = tmp_path / "lib" / "Movies" / "Film (2025) [tmdbid-1]" / "Film (2025).mp4"
dest.parent.mkdir(parents=True)
dest.write_bytes(b"video")
return str(dest), notified
def _srt(dest, lang, body="1\n00:00:01,000 --> 00:00:02,000\nhi\n"):
p = os.path.splitext(dest)[0] + f".{lang}.srt"
with open(p, "w") as f:
f.write(body)
return p
def _measure(table):
"""Fake _subsync_measure: (offset, scale) per input file NAME; writes the
'corrected' output so replacement can be observed."""
calls = []
def measure(ref, srt_in, srt_out):
# Faithful to ffsubsync 0.5.1, measured 2026-10-04: the output format
# comes from the output EXTENSION. Anything but .srt here (".synced",
# /dev/null) logs "unsupported output format", writes NOTHING, exits
# 0 -- after printing the offsets. The first fake wrote any name, which
# is how 32 good tracks got withdrawn in production.
name = os.path.basename(srt_in)
for prefix in ("synced-", "recheck-"):
if name.startswith(prefix):
name = name[len(prefix):] + ".synced"
calls.append(name)
offset, scale = table[name]
if srt_out and srt_out.endswith(".srt"):
with open(srt_out, "w") as f:
f.write(f"corrected {name}")
return {"offset": offset, "scale": scale, "score": 1.0}
return measure, calls
def test_an_in_sync_track_is_left_untouched(lib, monkeypatch):
dest, _ = lib
en = _srt(dest, "en")
measure, _ = _measure({"Film (2025).en.srt": (0.1, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
result = addon._sync_subtitles(dest)
assert result["in_sync"] == 1
assert open(en).read().startswith("1\n")
assert not os.path.exists(addon.SUBS_ORIGINAL_DIR) or not os.listdir(addon.SUBS_ORIGINAL_DIR)
def test_a_drifting_track_is_replaced_and_the_original_kept_outside_the_library(lib, monkeypatch):
dest, _ = lib
en = _srt(dest, "en")
measure, _ = _measure({"Film (2025).en.srt": (-2.25, 1.001),
"Film (2025).en.srt.synced": (0.01, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
result = addon._sync_subtitles(dest)
assert result["corrected"] == 1
assert open(en).read() == "corrected Film (2025).en.srt"
kept = os.path.join(addon.SUBS_ORIGINAL_DIR,
os.path.relpath(en, addon.LIBRARY_ROOT))
assert open(kept).read().startswith("1\n")
assert not kept.startswith(addon.LIBRARY_ROOT + os.sep)
def test_a_track_whose_correction_does_not_hold_is_removed_from_the_library(lib, monkeypatch):
dest, _ = lib
de = _srt(dest, "de")
measure, _ = _measure({"Film (2025).de.srt": (-18.5, 1.000),
"Film (2025).de.srt.synced": (4.0, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
result = addon._sync_subtitles(dest)
assert result["unfixable"] == 1
assert not os.path.exists(de)
kept = os.path.join(addon.SUBS_ORIGINAL_DIR, os.path.relpath(de, addon.LIBRARY_ROOT))
assert os.path.exists(kept)
def test_a_measurement_failure_leaves_the_track_alone(lib, monkeypatch):
"""A check that cannot run must not cost the user a subtitle."""
dest, _ = lib
en = _srt(dest, "en")
def broken(ref, srt_in, srt_out):
return None
monkeypatch.setattr(addon, "_subsync_measure", broken)
result = addon._sync_subtitles(dest)
assert result["unmeasured"] == 1
assert os.path.exists(en) and open(en).read().startswith("1\n")
def test_results_are_recorded_and_a_rerun_skips_them(lib, monkeypatch):
dest, _ = lib
_srt(dest, "en")
_srt(dest, "fr")
measure, calls = _measure({"Film (2025).en.srt": (0.0, 1.000),
"Film (2025).fr.srt": (-8.93, 1.000),
"Film (2025).fr.srt.synced": (0.0, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
with open(addon.SUBTITLE_SYNC_FILE) as f:
rec = json.load(f)
en = os.path.abspath(os.path.splitext(dest)[0] + ".en.srt")
fr = os.path.abspath(os.path.splitext(dest)[0] + ".fr.srt")
assert rec[en]["status"] == "in_sync"
assert rec[fr]["status"] == "corrected" and rec[fr]["offset"] == -8.93
calls.clear()
addon._sync_subtitles(dest)
assert calls == []
def test_a_new_track_beside_synced_ones_is_measured(lib, monkeypatch):
dest, _ = lib
_srt(dest, "en")
measure, calls = _measure({"Film (2025).en.srt": (0.0, 1.000),
"Film (2025).it.srt": (0.0, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
_srt(dest, "it")
calls.clear()
addon._sync_subtitles(dest)
assert calls == ["Film (2025).it.srt"]
def test_speech_is_extracted_once_per_copy(lib, monkeypatch):
dest, _ = lib
for lang in ("en", "fr", "it"):
_srt(dest, lang)
refs = []
monkeypatch.setattr(addon, "_subsync_reference",
lambda d, w: refs.append(d) or os.path.join(w, "ref.npz"))
measure, _ = _measure({f"Film (2025).{x}.srt": (0.0, 1.000) for x in ("en", "fr", "it")})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
assert refs == [dest]
def test_nothing_to_do_extracts_nothing(lib, monkeypatch):
dest, _ = lib
monkeypatch.setattr(addon, "_subsync_reference",
lambda d, w: pytest.fail("extracted speech for no tracks"))
assert addon._sync_subtitles(dest)["tracks"] == 0
def test_without_ffsubsync_nothing_changes(lib, monkeypatch):
dest, _ = lib
en = _srt(dest, "en")
monkeypatch.setattr(addon, "_subsync_available", lambda: False)
monkeypatch.setattr(addon, "_subsync_measure",
lambda *a: pytest.fail("measured without ffsubsync"))
assert addon._sync_subtitles(dest)["tracks"] == 0
assert os.path.exists(en)
def test_jellyfin_is_told_once_when_tracks_changed(lib, monkeypatch):
dest, notified = lib
_srt(dest, "fr")
measure, _ = _measure({"Film (2025).fr.srt": (-8.9, 1.000),
"Film (2025).fr.srt.synced": (0.0, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
assert notified == [(dest, "Modified")]
def test_jellyfin_is_not_told_when_nothing_changed(lib, monkeypatch):
dest, notified = lib
_srt(dest, "en")
measure, _ = _measure({"Film (2025).en.srt": (0.0, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
assert notified == []
def test_the_watch_panel_reports_the_sync(lib, monkeypatch):
dest, _ = lib
_srt(dest, "en")
_srt(dest, "fr")
measure, _ = _measure({"Film (2025).en.srt": (0.0, 1.000),
"Film (2025).fr.srt": (-8.9, 1.000),
"Film (2025).fr.srt.synced": (0.0, 1.000)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
assert addon._subtitle_summary(dest) == "2 tracks · 1 corrected · 1 in sync"
def test_parse_reads_ffsubsyncs_own_output():
out = ("[21:26:25] INFO score: 343738.036 ffsubsync.py:255\n"
" INFO offset seconds: -2.250 ffsubsync.py:256\n"
" INFO framerate scale factor: 1.001 ffsubsync.py:257\n")
assert addon._parse_subsync(out) == {"offset": -2.25, "scale": 1.001,
"score": 343738.036}
assert addon._parse_subsync("garbage") is None
def test_a_correction_that_was_not_written_leaves_the_track_alone(lib, monkeypatch):
"""The production failure: offsets measured, corrected file never
written. That is "could not repair", not "unfixable"."""
dest, _ = lib
fr = _srt(dest, "fr")
def measured_but_not_written(ref, srt_in, srt_out):
return {"offset": -8.9, "scale": 1.0, "score": 1.0}
monkeypatch.setattr(addon, "_subsync_measure", measured_but_not_written)
result = addon._sync_subtitles(dest)
assert result["unfixable"] == 0 and result["unmeasured"] == 1
assert os.path.exists(fr) and open(fr).read().startswith("1\n")
def test_a_recheck_that_cannot_run_leaves_the_track_alone(lib, monkeypatch):
dest, _ = lib
fr = _srt(dest, "fr")
calls = []
def first_ok_then_none(ref, srt_in, srt_out):
calls.append(srt_in)
if len(calls) == 1:
with open(srt_out, "w") as f:
f.write("corrected")
return {"offset": -8.9, "scale": 1.0, "score": 1.0}
return None
monkeypatch.setattr(addon, "_subsync_measure", first_ok_then_none)
result = addon._sync_subtitles(dest)
assert result["unfixable"] == 0 and result["unmeasured"] == 1
assert open(fr).read().startswith("1\n")
def test_every_output_path_keeps_the_srt_extension(lib, monkeypatch):
dest, _ = lib
_srt(dest, "fr")
outs = []
def record(ref, srt_in, srt_out):
outs.append(srt_out)
if srt_out:
open(srt_out, "w").write("x")
return {"offset": -8.9 if len(outs) == 1 else 0.0, "scale": 1.0, "score": 1.0}
monkeypatch.setattr(addon, "_subsync_measure", record)
addon._sync_subtitles(dest)
assert outs and all(o and o.endswith(".srt") for o in outs)
# --- English only (2026-10-04: "focus on english") ---------------------------
def test_only_english_is_synced_by_default(lib, monkeypatch):
dest, _ = lib
monkeypatch.setattr(addon, "SUBSYNC_LANGUAGES", ("en",))
_srt(dest, "en")
_srt(dest, "en.2")
fr = _srt(dest, "fr")
measure, calls = _measure({"Film (2025).en.srt": (0.0, 1.0),
"Film (2025).en.2.srt": (0.0, 1.0)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
result = addon._sync_subtitles(dest)
assert sorted(calls) == ["Film (2025).en.2.srt", "Film (2025).en.srt"]
assert result["tracks"] == 2
assert open(fr).read().startswith("1\n") # untouched
def test_english_only_extracts_nothing_when_there_is_no_english(lib, monkeypatch):
dest, _ = lib
monkeypatch.setattr(addon, "SUBSYNC_LANGUAGES", ("en",))
_srt(dest, "fr")
monkeypatch.setattr(addon, "_subsync_reference",
lambda d, w: pytest.fail("extracted speech with no English track"))
assert addon._sync_subtitles(dest)["tracks"] == 0
def test_the_summary_says_other_languages_are_not_checked(lib, monkeypatch):
dest, _ = lib
monkeypatch.setattr(addon, "SUBSYNC_LANGUAGES", ("en",))
_srt(dest, "en")
_srt(dest, "fr")
_srt(dest, "de")
measure, _ = _measure({"Film (2025).en.srt": (-2.25, 1.001),
"Film (2025).en.srt.synced": (0.0, 1.0)})
monkeypatch.setattr(addon, "_subsync_measure", measure)
addon._sync_subtitles(dest)
assert addon._subtitle_summary(dest) == \
"English: corrected (was -2.25 s, speed 1.001) · 2 other languages not checked"
def test_a_track_the_user_kept_is_reported_and_never_re_synced(lib, monkeypatch):
"""2026-10-04: the user restored two withdrawn Bleach tracks. They are
recorded as "kept" at the restored file's identity, so a later sync skips
them instead of withdrawing them again, and Watch says so."""
dest, _ = lib
monkeypatch.setattr(addon, "SUBSYNC_LANGUAGES", ("en",))
en = _srt(dest, "en")
with open(addon.SUBTITLE_SYNC_FILE, "w") as f:
json.dump({os.path.abspath(en): {"status": "kept", "offset": 22.18,
"scale": 1.0,
"identity": addon._file_identity(en)}}, f)
monkeypatch.setattr(addon, "_subsync_measure",
lambda *a: pytest.fail("re-measured a kept track"))
assert addon._sync_subtitles(dest)["tracks"] == 0
assert addon._subtitle_summary(dest) == "English: kept as fetched (measured 22.18 s off)"