backend/tests/test_health_timing_alignment.py
Rama master · réplica de lectura del repositorio original.
"""Health sweep: sidecars already on disk whose timing does not fit the video.
Prod 2026-11 (Mushoku Tensei S01): English sidecars 6 s late (E20) or for
another cut (E17, E21) sat next to the videos for months.
"""
from __future__ import annotations
from unittest.mock import patch
import pytest
import subtitle_timing
from services.subtitle_health.checkers import timing_alignment
from services.subtitle_health.models import IssueType, Severity, TargetKind
from services.subtitle_health.scan import ScanContext, Target
from subtitle_dialogue import dialogue_starts
from subtitle_timing import Reference
from tests.test_subtitle_timing import _speech, _srt, _translation_of
def _ctx(tmp_path, *targets):
video = tmp_path / "Show - S01E20.mp4"
video.write_bytes(b"")
return ScanContext(episode_id=20, video_path=str(video), targets=list(targets))
def _sidecar(tmp_path, suffix: str, content: bytes, *, forced=False) -> Target:
path = tmp_path / f"Show - S01E20{suffix}"
path.write_bytes(content)
return Target(
kind=TargetKind.SIDECAR,
path=str(path),
stream_index=None,
lang=suffix.split(".")[1],
codec="subrip",
raw=content,
forced=forced,
)
def _detect(ctx, reference, overruns=False):
with (
patch.object(subtitle_timing, "find_reference", return_value=reference),
patch.object(timing_alignment, "_overruns", return_value=overruns),
):
return timing_alignment.detect(ctx)
SPEECH = _speech()
class TestDetect:
def test_a_late_track_against_an_embedded_reference_is_shifted(self, tmp_path):
late = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(SPEECH, offset=6.0)))
[issue] = _detect(_ctx(tmp_path, late), Reference(SPEECH, True, "embedded"))
assert issue.type == IssueType.TIMING_OFFSET
assert issue.severity == Severity.CONFIRMED
assert issue.suggested_fix == timing_alignment.SHIFT_ACTION
def test_another_cut_against_an_embedded_reference_is_quarantined(self, tmp_path):
wrong = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(_speech(seed=9))))
[issue] = _detect(_ctx(tmp_path, wrong), Reference(SPEECH, True, "embedded"))
assert issue.type == IssueType.TIMING_MISMATCH
assert issue.suggested_fix == timing_alignment.QUARANTINE_ACTION
def test_against_another_sidecar_a_mismatch_is_only_suspicious(self, tmp_path):
# Which of the two is wrong cannot be told from their timing alone.
wrong = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(_speech(seed=9))))
[issue] = _detect(_ctx(tmp_path, wrong), Reference(SPEECH, False, "sidecar de"))
assert issue.severity == Severity.SUSPICIOUS
assert issue.suggested_fix is None
def test_a_mismatch_that_runs_past_the_video_is_the_wrong_one(self, tmp_path):
# E21: the English track ended 81 s after the episode.
wrong = _sidecar(tmp_path, ".en.ass", _srt(_translation_of(_speech(seed=9))))
[issue] = _detect(
_ctx(tmp_path, wrong), Reference(SPEECH, False, "sidecar de"), overruns=True
)
assert issue.severity == Severity.CONFIRMED
assert issue.suggested_fix == timing_alignment.QUARANTINE_ACTION
def test_a_shift_against_another_sidecar_is_not_applied(self, tmp_path):
late = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(SPEECH, offset=6.0)))
[issue] = _detect(_ctx(tmp_path, late), Reference(SPEECH, False, "sidecar de"))
assert issue.suggested_fix is None
@pytest.mark.parametrize("suffix", [".en.srt"])
def test_fitting_and_forced_tracks_raise_nothing(self, tmp_path, suffix):
fits = _sidecar(tmp_path, suffix, _srt(_translation_of(SPEECH)))
forced = _sidecar(
tmp_path, ".en.forced.srt", _srt(_translation_of(_speech(seed=9))), forced=True
)
assert _detect(_ctx(tmp_path, fits, forced), Reference(SPEECH, True, "embedded")) == []
def test_no_reference_raises_nothing(self, tmp_path):
wrong = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(_speech(seed=9))))
assert _detect(_ctx(tmp_path, wrong), None) == []
def test_shift_fixer_remeasures_and_keeps_a_backup(tmp_path):
from services.subtitle_health.fixers import shift_timing
late = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(SPEECH, offset=6.0)))
with (
patch.object(subtitle_timing, "find_reference", return_value=Reference(SPEECH, True, "e")),
patch(
"services.subtitle_health.fixers.shift_timing.backup_sidecar", return_value="/t/bak"
) as backup,
patch("services.subtitle_health.store.record_fix", return_value=7),
):
result = shift_timing.apply_to_sidecar(late.path, "/media/v.mp4", finding_id=1)
assert result["changed"] and result["fix_id"] == 7
assert result["shifted_s"] == pytest.approx(-6.0, abs=0.2)
backup.assert_called_once_with(late.path)
with open(late.path, "rb") as fh:
fixed = dialogue_starts(fh.read())
assert subtitle_timing.align(SPEECH, fixed).verdict == subtitle_timing.FIT
def test_shift_fixer_refuses_an_untrusted_reference(tmp_path):
from services.subtitle_health.fixers import shift_timing
late = _sidecar(tmp_path, ".en.srt", _srt(_translation_of(SPEECH, offset=6.0)))
with patch.object(
subtitle_timing, "find_reference", return_value=Reference(SPEECH, False, "sidecar")
):
result = shift_timing.apply_to_sidecar(late.path, "/media/v.mp4")
assert not result["changed"]
def test_sweep_repairs_confirmed_timing_findings_with_auto_fix_off(tmp_path):
from services.subtitle_health import sweep
findings = [
{"id": 1, "suggested_fix": "shift_timing", "target_kind": "sidecar", "target_path": "a"},
{"id": 2, "suggested_fix": None, "target_kind": "sidecar", "target_path": "b"},
{
"id": 3,
"suggested_fix": "quarantine_wrong_sidecar",
"target_kind": "sidecar",
"target_path": "c",
},
]
with (
patch.object(sweep, "_is_stable", return_value=True),
patch.object(sweep, "is_suspect_sidecar", return_value=False),
patch.object(sweep, "apply_fix", return_value={"changed": True}) as apply_fix,
):
sweep._repair_timing(findings, "/media/v.mp4")
assert [c.args for c in apply_fix.call_args_list] == [
(1, "shift_timing"),
(3, "quarantine_wrong_sidecar"),
]