"""Tests for dubtitle detection & selection (roadmap A4 / issue #146). Covers the pure similarity primitives, Tier-1 heuristic classification, the windowing maths, and the orchestrator's Tier-1/Tier-2 decision paths with audio sampling mocked out (no ffmpeg/Whisper needed). """ from __future__ import annotations import services.dubtitle.detector as det from services.dubtitle.detector import ( _classify_tier1, _compute_windows, _Cue, _cue_statistics, _english_subtitle_streams, _score_cues_against_windows, _TranscriptWindow, detect_dubtitle, validate_subtitle_against_audio, ) from services.dubtitle.text_similarity import ( containment_score, jaccard_score, normalize_to_tokens, ) # --------------------------------------------------------------------------- # text_similarity # --------------------------------------------------------------------------- class TestNormalizeToTokens: def test_strips_ass_override_blocks(self): assert normalize_to_tokens("{\\an8}hello {\\pos(1,2)}world") == ["hello", "world"] def test_strips_bracketed_sdh_by_default(self): assert normalize_to_tokens("[door creaks] run now") == ["run", "now"] def test_keeps_bracketed_when_disabled(self): toks = normalize_to_tokens("[door creaks] run", strip_bracketed=False) assert "door" in toks and "creaks" in toks def test_strips_speaker_labels_and_fillers(self): assert normalize_to_tokens("NARRATOR: uh well then") == ["well", "then"] def test_strips_music_notes_and_linebreaks(self): assert normalize_to_tokens("first\\Nsecond ♪") == ["first", "second"] def test_empty(self): assert normalize_to_tokens("") == [] assert normalize_to_tokens("{\\an8}♪♪") == [] class TestSimilarityMetrics: def test_containment_is_asymmetric_fraction(self): sub = {"the", "quick", "brown", "fox"} transcript = {"a", "quick", "brown", "fox", "jumps"} assert containment_score(sub, transcript) == 0.75 def test_containment_empty_sub_is_zero(self): assert containment_score(set(), {"a"}) == 0.0 def test_dubtitle_scores_far_above_fansub(self): # Dub transcript tokens transcript = set( ["i", "will", "protect", "everyone", "no", "matter", "what", "it", "takes"] ) dubtitle = set(["i", "will", "protect", "everyone", "no", "matter", "what"]) fansub = set(["the", "bonds", "we", "share", "transcend", "mere", "obligation"]) assert containment_score(dubtitle, transcript) >= 0.7 assert containment_score(fansub, transcript) <= 0.4 def test_jaccard_symmetric(self): assert jaccard_score({"a", "b"}, {"b", "c"}) == 1 / 3 assert jaccard_score(set(), set()) == 0.0 # --------------------------------------------------------------------------- # Tier-1 statistics + classification # --------------------------------------------------------------------------- def _dense_cues( n: int, *, gap_ms: int = 3000, text: str = "some spoken dialogue here" ) -> list[_Cue]: cues = [] t = 0 for _ in range(n): cues.append(_Cue(start_ms=t, end_ms=t + 2000, text=text)) t += gap_ms return cues class TestCueStatistics: def test_density_and_overlap(self): cues = _dense_cues(20) density, avg_cps, overlap = _cue_statistics(cues) assert density > 3.0 assert avg_cps > 0 assert overlap == 0.0 def test_detects_overlap(self): cues = [ _Cue(0, 5000, "one"), _Cue(3000, 8000, "two"), # starts before previous ends _Cue(9000, 10000, "three"), ] _, _, overlap = _cue_statistics(cues) assert overlap > 0 def test_empty(self): assert _cue_statistics([]) == (0.0, 0.0, 0.0) class TestTier1Classification: def _stream(self, *, sub_index=0, title="", forced=False, sdh=False, lang="eng"): disp = {"forced": 1 if forced else 0} if sdh: disp["hearing_impaired"] = 1 raw = { "index": 10 + sub_index, "tags": {"title": title, "language": lang}, "disposition": disp, } return { "sub_index": sub_index, "format": "ass", "language": lang, "_tags": raw["tags"], "_raw": raw, } def test_forced_disposition_is_signs_forced(self): c = _classify_tier1(self._stream(forced=True), _dense_cues(50)) assert c.tier1_label == "signs_forced" assert c.subtype == "forced" def test_signs_title_is_signs_forced(self): c = _classify_tier1(self._stream(title="Signs & Songs"), _dense_cues(50)) assert c.tier1_label == "signs_forced" def test_sparse_track_is_sparse(self): c = _classify_tier1(self._stream(), _dense_cues(5)) assert c.tier1_label == "sparse" def test_sdh_full_track_is_likely_dubtitle(self): c = _classify_tier1(self._stream(title="English CC", sdh=True), _dense_cues(80)) assert c.tier1_label == "likely_dubtitle" assert c.is_sdh is True def test_plain_full_track_is_candidate(self): c = _classify_tier1(self._stream(title="English"), _dense_cues(80)) assert c.tier1_label == "candidate" class TestEnglishStreamFilter: def test_filters_english_text_streams_only(self): probe = { "streams": [ {"codec_type": "audio", "codec_name": "aac", "tags": {"language": "eng"}}, {"codec_type": "subtitle", "codec_name": "ass", "tags": {"language": "eng"}}, {"codec_type": "subtitle", "codec_name": "subrip", "tags": {"language": "jpn"}}, { "codec_type": "subtitle", "codec_name": "hdmv_pgs_subtitle", "tags": {"language": "eng"}, }, {"codec_type": "subtitle", "codec_name": "srt", "tags": {"language": "und"}}, ] } out = _english_subtitle_streams(probe) langs = {s["language"] for s in out} # eng ass + und srt kept; jpn excluded; pgs (image) excluded assert langs == {"eng", "und"} # sub_index counts every subtitle stream, including the skipped pgs assert [s["sub_index"] for s in out] == [0, 3] # --------------------------------------------------------------------------- # Tier-2 windowing + scoring # --------------------------------------------------------------------------- class TestWindows: def test_three_windows_for_long_file(self): wins = _compute_windows(24 * 60 * 1000) assert len(wins) == 3 assert all(end > start for start, end in wins) # OP/ED-avoiding: first window not at absolute zero for a long file assert wins[0][0] > 0 def test_short_file_collapses(self): wins = _compute_windows(20 * 1000) assert len(wins) == 1 assert wins[0] == (0, 20 * 1000) def test_zero_duration(self): assert _compute_windows(0) == [] class TestScoreCuesAgainstWindows: def test_scores_only_in_window_cues(self): windows = [_TranscriptWindow(0, 10_000, {"hello", "world", "run"})] cues = [ _Cue(1000, 2000, "hello world"), # in window, full match _Cue(50_000, 51_000, "totally different text"), # out of window, ignored ] score = _score_cues_against_windows(cues, windows) assert score == 1.0 def test_none_when_no_cues_in_window(self): windows = [_TranscriptWindow(0, 10_000, {"hello"})] cues = [_Cue(50_000, 51_000, "later")] assert _score_cues_against_windows(cues, windows) is None # --------------------------------------------------------------------------- # Orchestrator # --------------------------------------------------------------------------- class TestDetectDubtitle: def _probe(self, langs_codecs): return { "streams": [ { "codec_type": "subtitle", "codec_name": codec, "tags": {"language": lang, "title": title}, } for lang, codec, title in langs_codecs ] } def test_no_english_tracks(self, monkeypatch): probe = { "streams": [ {"codec_type": "subtitle", "codec_name": "ass", "tags": {"language": "jpn"}} ] } res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.55) assert res.dubtitle_sub_index is None assert "No English" in res.message def test_single_full_text_track_tier1(self, monkeypatch): probe = self._probe([("eng", "ass", "English")]) monkeypatch.setattr(det, "_extract_and_load_cues", lambda *a, **k: _dense_cues(80)) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.55) assert res.method == "tier1" assert res.dubtitle_sub_index == 0 assert res.candidates[0].is_dubtitle is True def test_lone_sdh_wins_tier1(self, monkeypatch): probe = self._probe([("eng", "ass", "English"), ("eng", "ass", "English SDH")]) monkeypatch.setattr(det, "_extract_and_load_cues", lambda *a, **k: _dense_cues(80)) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.55) assert res.method == "tier1" # SDH track is sub_index 1 assert res.dubtitle_sub_index == 1 def test_two_full_text_tracks_trigger_tier2(self, monkeypatch): probe = self._probe( [("eng", "ass", "English [GroupA]"), ("eng", "ass", "English [GroupB]")] ) def fake_cues(video, sub_index, fmt): # Dense tracks (so Tier-1 = candidate), with distinct text in the # sampling window [720_000, 765_000]. window_text = ( "the bonds we share transcend obligation" if sub_index == 0 # fansub — diverges from the dub else "i will protect everyone no matter what" # dubtitle — matches ) cues = [] for i in range(80): t = i * 10_000 text = window_text if 720_000 <= t < 765_000 else "filler line here" cues.append(_Cue(t, t + 2_000, text)) return cues monkeypatch.setattr(det, "_extract_and_load_cues", fake_cues) transcript = set( ["i", "will", "protect", "everyone", "no", "matter", "what", "it", "takes"] ) monkeypatch.setattr( det, "_build_dub_transcript_windows", lambda video, dur: [_TranscriptWindow(720_000, 765_000, transcript)], ) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.55) assert res.method == "tier2" assert res.tier2_ran is True assert res.dubtitle_sub_index == 1 assert res.candidates[1].audio_score >= 0.55 assert res.candidates[0].audio_score < res.candidates[1].audio_score def test_tier2_unavailable_falls_back_gracefully(self, monkeypatch): probe = self._probe([("eng", "ass", "A"), ("eng", "ass", "B")]) monkeypatch.setattr(det, "_extract_and_load_cues", lambda *a, **k: _dense_cues(80)) monkeypatch.setattr(det, "_build_dub_transcript_windows", lambda video, dur: []) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.55) assert res.method == "tier1" assert res.dubtitle_sub_index is None assert res.whisper_fallback_available is True def _two_close_tracks(self, monkeypatch): """Two dense EN tracks whose audio scores are close (margin ~0.1).""" probe = self._probe([("eng", "ass", "A"), ("eng", "ass", "B")]) def fake_cues(video, sub_index, fmt): # sub 0 → 0.9 (one out-of-transcript token); sub 1 → 1.0 (all match) window_text = "a b c d e f g h i zzz" if sub_index == 0 else "a b c d e f g h i j" cues = [] for i in range(90): t = i * 10_000 text = window_text if 720_000 <= t < 765_000 else "a b c d e f g h i j" cues.append(_Cue(t, t + 2_000, text)) return cues monkeypatch.setattr(det, "_extract_and_load_cues", fake_cues) transcript = set(["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]) monkeypatch.setattr( det, "_build_dub_transcript_windows", lambda video, dur: [_TranscriptWindow(720_000, 765_000, transcript)], ) return probe def test_margin_and_runner_up_populated(self, monkeypatch): probe = self._two_close_tracks(monkeypatch) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.5) assert res.margin is not None assert res.runner_up_score is not None assert res.margin == round(1.0 - 0.9, 3) def test_on_demand_flags_best_despite_small_margin(self, monkeypatch): probe = self._two_close_tracks(monkeypatch) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.5, automated=False) assert res.dubtitle_sub_index == 1 # best (1.0) suggested for human def test_automated_blocks_on_small_margin(self, monkeypatch): probe = self._two_close_tracks(monkeypatch) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.5, automated=True) # margin 0.1 < default 0.15 → not auto-flagged assert res.dubtitle_sub_index is None assert "margin" in res.message.lower() def test_automated_cue_floor_excludes_thin_tracks(self, monkeypatch): # Two full-text tracks but each below the auto cue floor (70) → excluded, # so automated mode finds nothing to score. probe = self._probe([("eng", "ass", "A"), ("eng", "ass", "B")]) monkeypatch.setattr(det, "_extract_and_load_cues", lambda *a, **k: _dense_cues(50)) called = {"n": 0} monkeypatch.setattr( det, "_build_dub_transcript_windows", lambda v, d: called.__setitem__("n", 1) or [] ) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.5, automated=True) # 50-cue tracks pass on-demand floor (40) but fail auto floor (70): # both drop out of full_text → no full-text tracks → Tier-2 never runs. assert called["n"] == 0 assert res.dubtitle_sub_index is None def test_run_tier2_false_skips_audio(self, monkeypatch): probe = self._probe([("eng", "ass", "A"), ("eng", "ass", "B")]) monkeypatch.setattr(det, "_extract_and_load_cues", lambda *a, **k: _dense_cues(80)) called = {"n": 0} def _should_not_run(*a, **k): called["n"] += 1 return [] monkeypatch.setattr(det, "_build_dub_transcript_windows", _should_not_run) res = detect_dubtitle("/fake.mkv", probe_data=probe, min_score=0.55, run_tier2=False) assert called["n"] == 0 assert res.dubtitle_sub_index is None assert res.whisper_fallback_available is True class TestValidateSubtitleAgainstAudio: def test_pass_when_matches(self, monkeypatch, tmp_path): sub = tmp_path / "x.srt" sub.write_text( "1\n00:12:00,000 --> 00:12:02,000\ni will protect everyone no matter what\n", encoding="utf-8", ) transcript = set( ["i", "will", "protect", "everyone", "no", "matter", "what", "it", "takes"] ) monkeypatch.setattr( det, "_build_dub_transcript_windows", lambda video, dur: [_TranscriptWindow(720_000, 765_000, transcript)], ) res = validate_subtitle_against_audio("/fake.mkv", str(sub), min_score=0.55) assert res.passed is True assert res.score >= 0.55 def test_fail_when_mismatch(self, monkeypatch, tmp_path): sub = tmp_path / "x.srt" sub.write_text( "1\n00:12:00,000 --> 00:12:02,000\nthe bonds we share transcend obligation\n", encoding="utf-8", ) transcript = set( ["i", "will", "protect", "everyone", "no", "matter", "what", "it", "takes"] ) monkeypatch.setattr( det, "_build_dub_transcript_windows", lambda video, dur: [_TranscriptWindow(720_000, 765_000, transcript)], ) res = validate_subtitle_against_audio("/fake.mkv", str(sub), min_score=0.55) assert res.passed is False def test_unverifiable_does_not_block(self, monkeypatch, tmp_path): sub = tmp_path / "x.srt" sub.write_text("1\n00:12:00,000 --> 00:12:02,000\nhello\n", encoding="utf-8") monkeypatch.setattr(det, "_build_dub_transcript_windows", lambda video, dur: []) res = validate_subtitle_against_audio("/fake.mkv", str(sub), min_score=0.55) assert res.passed is False assert "skipped" in res.message.lower()