diff options
Diffstat (limited to 'working/meeting-transcription-service/tests')
4 files changed, 1206 insertions, 0 deletions
diff --git a/working/meeting-transcription-service/tests/test_diarize.py b/working/meeting-transcription-service/tests/test_diarize.py new file mode 100644 index 0000000..ef009b8 --- /dev/null +++ b/working/meeting-transcription-service/tests/test_diarize.py @@ -0,0 +1,74 @@ +"""Tests for diarize's pure parts. The pyannote pipeline itself is not loaded here.""" + +import sys +from collections import namedtuple +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +import diarize # noqa: E402 + +Segment = namedtuple("Segment", "start end") + + +class TestTurnsFromTracks: + def test_diarize_turns_from_tracks_converts_and_sorts(self): + """Normal: (segment, track, label) triples become sorted turn dicts.""" + tracks = [(Segment(5.0, 9.25), "_", "SPEAKER_01"), (Segment(0.5, 4.0), "_", "SPEAKER_00")] + assert diarize.turns_from_tracks(tracks) == [ + {"start": 0.5, "end": 4.0, "speaker": "SPEAKER_00"}, + {"start": 5.0, "end": 9.25, "speaker": "SPEAKER_01"}, + ] + + def test_diarize_turns_from_tracks_rounds_to_milliseconds(self): + """Boundary: float noise from the model is rounded away.""" + tracks = [(Segment(0.03096875, 1.9980000000000002), "_", "SPEAKER_00")] + assert diarize.turns_from_tracks(tracks) == [ + {"start": 0.031, "end": 1.998, "speaker": "SPEAKER_00"} + ] + + def test_diarize_turns_from_tracks_drops_empty_segments(self): + """Boundary: zero or negative length segments carry no speech.""" + tracks = [(Segment(2.0, 2.0), "_", "SPEAKER_00"), (Segment(3.0, 2.5), "_", "SPEAKER_00")] + assert diarize.turns_from_tracks(tracks) == [] + + def test_diarize_turns_from_tracks_empty_input_is_empty_list(self): + """Boundary: no tracks.""" + assert diarize.turns_from_tracks([]) == [] + + +class TestParseArgs: + def test_diarize_parse_args_speakers_sets_exact_count(self): + """Normal: --speakers pins the count.""" + args = diarize.parse_args(["a.wav", "out.json", "--speakers", "3"]) + assert (args.audio, args.out, args.speakers) == ("a.wav", "out.json", 3) + + def test_diarize_parse_args_defaults_let_the_model_estimate(self): + """Normal: no count given.""" + args = diarize.parse_args(["a.wav", "out.json"]) + assert args.speakers is None and args.min_speakers is None and args.max_speakers is None + + @pytest.mark.parametrize("argv", [ + ["a.wav", "out.json", "--speakers", "0"], + ["a.wav", "out.json", "--speakers", "-2"], + ["a.wav", "out.json", "--speakers", "three"], + ["a.wav", "out.json", "--speakers", "3", "--max-speakers", "5"], + ["a.wav", "out.json", "--min-speakers", "4", "--max-speakers", "2"], + ["a.wav"], + ]) + def test_diarize_parse_args_rejects_bad_counts(self, argv): + """Error: non-positive, non-numeric, contradictory or missing arguments.""" + with pytest.raises(SystemExit): + diarize.parse_args(argv) + + +class TestPipelineKwargs: + def test_diarize_pipeline_kwargs_only_passes_what_was_given(self): + """Normal: unset options are not forwarded to the model.""" + args = diarize.parse_args(["a.wav", "o.json", "--min-speakers", "2", "--max-speakers", "4"]) + assert diarize.pipeline_kwargs(args) == {"min_speakers": 2, "max_speakers": 4} + args = diarize.parse_args(["a.wav", "o.json", "--speakers", "3"]) + assert diarize.pipeline_kwargs(args) == {"num_speakers": 3} + assert diarize.pipeline_kwargs(diarize.parse_args(["a.wav", "o.json"])) == {} diff --git a/working/meeting-transcription-service/tests/test_merge_transcript.py b/working/meeting-transcription-service/tests/test_merge_transcript.py new file mode 100644 index 0000000..a584043 --- /dev/null +++ b/working/meeting-transcription-service/tests/test_merge_transcript.py @@ -0,0 +1,510 @@ +"""Tests for merge_transcript: whisper words + speaker turns -> transcript lines.""" + +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) + +import merge_transcript as mt # noqa: E402 +from merge_transcript import Turn, Unit # noqa: E402 + + +def words(start, *tokens, step=0.5): + """Evenly spaced word units beginning at ``start`` seconds.""" + return [ + Unit(start + i * step, start + (i + 1) * step, f" {token}") + for i, token in enumerate(tokens) + ] + + +class TestMergeNormal: + def test_merge_transcript_merge_two_speakers_yields_one_line_per_turn(self): + """Normal: words fall into the turn that contains them.""" + units = words(0.0, "Good", "morning.") + words(6.0, "Sounds", "good.") + turns = [Turn(0.0, 5.0, "SPEAKER_00"), Turn(5.5, 9.0, "SPEAKER_01")] + assert mt.merge(units, turns) == [ + "00:00:00 Speaker A: Good morning.", + "00:00:06 Speaker B: Sounds good.", + ] + + def test_merge_transcript_merge_letters_follow_order_of_first_speech(self): + """Normal: Speaker A is whoever talks first, whatever pyannote called them.""" + units = words(0.0, "First.") + words(4.0, "Second.") + words(8.0, "Again.") + turns = [ + Turn(0.0, 3.0, "SPEAKER_02"), + Turn(3.5, 7.0, "SPEAKER_00"), + Turn(7.5, 10.0, "SPEAKER_02"), + ] + assert mt.merge(units, turns) == [ + "00:00:00 Speaker A: First.", + "00:00:04 Speaker B: Second.", + "00:00:08 Speaker A: Again.", + ] + + def test_merge_transcript_merge_same_speaker_across_turns_stays_one_line(self): + """Normal: pyannote splits a speaker's run into turns; the line does not.""" + units = words(0.0, "One", "two") + words(1.2, "three.") + turns = [Turn(0.0, 1.0, "SPEAKER_00"), Turn(1.1, 2.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["00:00:00 Speaker A: One two three."] + + def test_merge_transcript_merge_long_pause_starts_a_new_line(self): + """Normal: a pause past max_gap_s breaks the line so timestamps stay useful.""" + units = words(0.0, "Before.") + words(10.0, "After.") + turns = [Turn(0.0, 12.0, "SPEAKER_00")] + assert mt.merge(units, turns, max_gap_s=3.0) == [ + "00:00:00 Speaker A: Before.", + "00:00:10 Speaker A: After.", + ] + + def test_merge_transcript_merge_word_in_a_gap_goes_to_nearest_turn(self): + """Normal: whisper's timing drifts; a word between turns joins the closer one.""" + units = [Unit(4.6, 4.8, " late"), Unit(5.2, 5.4, " early")] + turns = [Turn(0.0, 4.5, "SPEAKER_00"), Turn(5.5, 9.0, "SPEAKER_01")] + assert mt.merge(units, turns) == [ + "00:00:04 Speaker A: late", + "00:00:05 Speaker B: early", + ] + + def test_merge_transcript_merge_overlapping_turns_pick_the_larger_overlap(self): + """Normal: with overlapped speech, the word goes where most of it sits.""" + # The anchor word fixes SPEAKER_00 as A, so the contested word's label + # actually shows which turn won: 0.2 s of overlap with A, 0.9 s with B. + units = [Unit(1.0, 1.5, " anchor"), Unit(4.0, 5.0, " contested")] + turns = [Turn(0.0, 4.2, "SPEAKER_00"), Turn(4.1, 9.0, "SPEAKER_01")] + assert mt.merge(units, turns) == [ + "00:00:01 Speaker A: anchor", + "00:00:04 Speaker B: contested", + ] + + +class TestMergeBoundary: + def test_merge_transcript_merge_timestamp_past_an_hour_is_floored(self): + """Boundary: 3725.9 s renders as 01:02:05.""" + units = [Unit(3725.9, 3726.4, " Late.")] + turns = [Turn(3700.0, 3800.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["01:02:05 Speaker A: Late."] + + def test_merge_transcript_merge_zero_length_and_blank_units_are_handled(self): + """Boundary: whisper emits empty units and zero-length words.""" + units = [Unit(0.0, 0.0, ""), Unit(2.18, 2.18, " we"), Unit(2.18, 2.54, " have")] + turns = [Turn(0.0, 5.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["00:00:02 Speaker A: we have"] + + def test_merge_transcript_merge_twenty_seventh_speaker_gets_a_number(self): + """Boundary: letters run out after Z.""" + units, turns = [], [] + for i in range(27): + units += [Unit(i * 10.0, i * 10.0 + 1, f" s{i}")] + turns += [Turn(i * 10.0, i * 10.0 + 5, f"SPEAKER_{i:02d}")] + lines = mt.merge(units, turns) + assert lines[25] == "00:04:10 Speaker Z: s25" + assert lines[26] == "00:04:20 Speaker 26: s26" + + def test_merge_transcript_merge_unicode_and_inner_spacing_survive(self): + """Boundary: non-ASCII text; whitespace collapses to single spaces.""" + units = [Unit(0.0, 0.5, " Բարև,"), Unit(0.5, 1.0, " Երևան"), Unit(1.0, 1.5, " — café.")] + turns = [Turn(0.0, 2.0, "SPEAKER_00")] + assert mt.merge(units, turns) == ["00:00:00 Speaker A: Բարև, Երևան — café."] + + def test_merge_transcript_merge_unsorted_input_is_sorted_first(self): + """Boundary: neither list has to arrive in time order.""" + units = words(6.0, "Second.") + words(0.0, "First.") + turns = [Turn(5.0, 9.0, "SPEAKER_01"), Turn(0.0, 4.0, "SPEAKER_00")] + assert mt.merge(units, turns) == [ + "00:00:00 Speaker A: First.", + "00:00:06 Speaker B: Second.", + ] + + +class TestMergeError: + def test_merge_transcript_merge_no_turns_raises(self): + """Error: words with no diarization cannot be labelled.""" + with pytest.raises(ValueError, match="no speaker turns"): + mt.merge(words(0.0, "Hello."), []) + + def test_merge_transcript_merge_no_words_raises(self): + """Error: an empty transcription is a failure, not an empty transcript.""" + with pytest.raises(ValueError, match="no speech"): + mt.merge([Unit(0.0, 0.0, " ")], [Turn(0.0, 5.0, "SPEAKER_00")]) + + def test_merge_transcript_merge_negative_gap_setting_raises(self): + """Error: max_gap_s must not be negative.""" + with pytest.raises(ValueError, match="max_gap_s"): + mt.merge(words(0.0, "Hi."), [Turn(0.0, 1.0, "SPEAKER_00")], max_gap_s=-1) + + +class TestLoaders: + def test_merge_transcript_load_whisper_json_reads_millisecond_offsets(self, tmp_path): + """Normal: whisper-cli -oj stores offsets in milliseconds.""" + path = tmp_path / "w.json" + path.write_text(json.dumps({"transcription": [ + {"offsets": {"from": 0, "to": 190}, "text": " if"}, + {"offsets": {"from": 190, "to": 590}, "text": " there's"}, + ]})) + assert mt.load_whisper_json(path) == [Unit(0.0, 0.19, " if"), Unit(0.19, 0.59, " there's")] + + def test_merge_transcript_load_turns_json_reads_seconds(self, tmp_path): + """Normal: the diarizer writes start/end in seconds.""" + path = tmp_path / "t.json" + path.write_text(json.dumps([{"start": 0.5, "end": 4.25, "speaker": "SPEAKER_00"}])) + assert mt.load_turns_json(path) == [Turn(0.5, 4.25, "SPEAKER_00")] + + @pytest.mark.parametrize("content", ["not json", "{}", '{"transcription": [{"text": "x"}]}']) + def test_merge_transcript_load_whisper_json_malformed_raises(self, tmp_path, content): + """Error: a truncated or foreign file is rejected with its path named.""" + path = tmp_path / "w.json" + path.write_text(content) + with pytest.raises(ValueError, match="w.json"): + mt.load_whisper_json(path) + + @pytest.mark.parametrize("content", ["not json", "{}", '[{"start": 1}]']) + def test_merge_transcript_load_turns_json_malformed_raises(self, tmp_path, content): + """Error: same for the turns file.""" + path = tmp_path / "t.json" + path.write_text(content) + with pytest.raises(ValueError, match="t.json"): + mt.load_turns_json(path) + + +class TestLoadersMissingFile: + def test_merge_transcript_load_whisper_json_missing_file_raises_value_error(self, tmp_path): + """Error: a missing whisper file is a clean ValueError naming the path.""" + with pytest.raises(ValueError, match="gone.json"): + mt.load_whisper_json(tmp_path / "gone.json") + + def test_merge_transcript_load_turns_json_missing_file_raises_value_error(self, tmp_path): + """Error: a missing turns file (the diarizer failed upstream) is a clean ValueError.""" + with pytest.raises(ValueError, match="gone.json"): + mt.load_turns_json(tmp_path / "gone.json") + + def test_merge_transcript_main_missing_turns_file_exits_one_without_traceback(self, tmp_path, capsys): + """Error: the CLI reports it in one line and prints nothing on stdout.""" + w = tmp_path / "w.json" + w.write_text('{"transcription": [{"offsets": {"from": 0, "to": 500}, "text": " Hi."}]}') + assert mt.main([str(w), str(tmp_path / "gone.json")]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "gone.json" in captured.err and "Traceback" not in captured.err + + +class TestCli: + def _files(self, tmp_path, transcription, turns): + w = tmp_path / "w.json" + t = tmp_path / "t.json" + w.write_text(json.dumps({"transcription": transcription})) + t.write_text(json.dumps(turns)) + return str(w), str(t) + + def test_merge_transcript_main_prints_transcript_and_returns_zero(self, tmp_path, capsys): + """Normal: the CLI writes the lines to stdout.""" + w, t = self._files( + tmp_path, + [{"offsets": {"from": 0, "to": 500}, "text": " Hello."}], + [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}], + ) + assert mt.main([w, t]) == 0 + assert capsys.readouterr().out == "00:00:00 Speaker A: Hello.\n" + + def test_merge_transcript_main_failure_prints_nothing_on_stdout(self, tmp_path, capsys): + """Error: a failed merge exits 1 with the reason on stderr only.""" + w, t = self._files(tmp_path, [], [{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}]) + assert mt.main([w, t]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "no speech" in captured.err + + def test_merge_transcript_main_wrong_argument_count_returns_two(self, capsys): + """Error: usage.""" + assert mt.main([]) == 2 + assert "usage" in capsys.readouterr().err.lower() + + +def tok(start_ms, end_ms, text): + return {"text": text, "offsets": {"from": start_ms, "to": end_ms}} + + +class TestLoadWhisperTokens: + """whisper-cli -ojf keeps normal segments and adds per-token offsets inside each.""" + + def _write(self, tmp_path, segments): + path = tmp_path / "full.json" + path.write_text(json.dumps({"transcription": segments})) + return path + + def test_merge_transcript_load_whisper_json_prefers_tokens_over_segments(self, tmp_path): + """Normal: with tokens present, units are words, not whole segments.""" + path = self._write(tmp_path, [{ + "offsets": {"from": 0, "to": 2000}, "text": " we basically", + "tokens": [tok(0, 0, "[_BEG_]"), tok(10, 400, " we"), tok(500, 1900, " basically"), tok(2000, 2000, "[_TT_100]")], + }]) + assert mt.load_whisper_json(path) == [Unit(0.01, 0.4, " we"), Unit(0.5, 1.9, " basically")] + + def test_merge_transcript_load_whisper_json_joins_subword_and_punctuation_tokens(self, tmp_path): + """Normal: a token with no leading space continues the previous word.""" + path = self._write(tmp_path, [{ + "offsets": {"from": 0, "to": 3000}, "text": " Saturday, yes.", + "tokens": [tok(0, 300, " Sat"), tok(300, 700, "urday"), tok(700, 750, ","), tok(900, 1300, " yes"), tok(1300, 1350, ".")], + }]) + assert mt.load_whisper_json(path) == [Unit(0.0, 0.75, " Saturday,"), Unit(0.9, 1.35, " yes.")] + + def test_merge_transcript_load_whisper_json_segment_without_tokens_falls_back(self, tmp_path): + """Boundary: plain -oj output, or a segment whose token list is empty.""" + path = self._write(tmp_path, [ + {"offsets": {"from": 0, "to": 1000}, "text": " First.", "tokens": []}, + {"offsets": {"from": 1000, "to": 2000}, "text": " Second."}, + ]) + assert mt.load_whisper_json(path) == [Unit(0.0, 1.0, " First."), Unit(1.0, 2.0, " Second.")] + + def test_merge_transcript_load_whisper_json_leading_continuation_token_stands_alone(self, tmp_path): + """Boundary: the very first token has nothing to attach to.""" + path = self._write(tmp_path, [{ + "offsets": {"from": 0, "to": 500}, "text": "ing on", + "tokens": [tok(0, 200, "ing"), tok(200, 500, " on")], + }]) + assert mt.load_whisper_json(path) == [Unit(0.0, 0.2, "ing"), Unit(0.2, 0.5, " on")] + + def test_merge_transcript_load_whisper_json_token_without_offsets_raises(self, tmp_path): + """Error: a malformed token is rejected, naming the file.""" + path = self._write(tmp_path, [{"offsets": {"from": 0, "to": 1}, "text": " x", "tokens": [{"text": " x"}]}]) + with pytest.raises(ValueError, match="full.json"): + mt.load_whisper_json(path) + + +class TestFindRepetition: + def test_merge_transcript_find_repetition_clean_text_returns_none(self): + """Normal: ordinary speech, even with stock phrases scattered about.""" + text = " ".join(f"point {i} and i don't know if that works for us" for i in range(8)) + assert mt.find_repetition(text) is None + + def test_merge_transcript_find_repetition_back_to_back_phrase_is_reported(self): + """Normal: whisper's hallucination loop, the same phrase again and again.""" + text = "okay so " + "We don't know where we're going to be. " * 6 + "anyway moving on" + hit = mt.find_repetition(text) + assert hit is not None + phrase, count = hit + assert count >= 6 + assert "where we're going to be" in phrase.lower() + + def test_merge_transcript_find_repetition_three_repeats_is_tolerated(self): + """Boundary: people do say a thing three times; four in a row is the line.""" + assert mt.find_repetition("go back to this area " * 3) is None + assert mt.find_repetition("go back to this area " * 4) is not None + + def test_merge_transcript_find_repetition_short_fillers_are_not_loops(self): + """Boundary: 'yeah yeah yeah yeah yeah' is speech, not a loop.""" + assert mt.find_repetition("yeah " * 9 + "no no no no no") is None + + def test_merge_transcript_find_repetition_empty_text_returns_none(self): + """Boundary: nothing to scan.""" + assert mt.find_repetition("") is None + + def test_merge_transcript_main_loop_in_transcript_fails(self, tmp_path, capsys): + """Error: a looping transcription exits 1 with nothing on stdout.""" + words_ = ("we don't know where we're going to be " * 5).split() + # 40 words at 3 s each: two minutes of loop, far past the 30 s a short stutter gets + segs = [{"offsets": {"from": i * 3_000, "to": i * 3_000 + 2_500}, "text": f" {w}"} for i, w in enumerate(words_)] + w = tmp_path / "w.json" + w.write_text(json.dumps({"transcription": segs})) + t = tmp_path / "t.json" + t.write_text(json.dumps([{"start": 0.0, "end": 130.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(w), str(t)]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "repeat" in captured.err.lower() + + +class TestInvertedTokenTimes: + """whisper-cli sometimes clamps a token's start to its segment, leaving end < start.""" + + def test_merge_transcript_load_whisper_json_inverted_token_becomes_zero_length(self, tmp_path): + """Boundary: an end before the start is treated as a zero-length word at the start.""" + path = tmp_path / "inv.json" + path.write_text(json.dumps({"transcription": [{ + "offsets": {"from": 13120, "to": 16160}, "text": " which you", + "tokens": [tok(13120, 9580, " which"), tok(13120, 10140, " you")], + }]})) + assert mt.load_whisper_json(path) == [Unit(13.12, 13.12, " which"), Unit(13.12, 13.12, " you")] + + def test_merge_transcript_merge_inverted_tokens_do_not_split_a_sentence(self, tmp_path): + """Normal: the real case, after a pause longer than max_gap_s.""" + path = tmp_path / "inv.json" + path.write_text(json.dumps({"transcription": [ + {"offsets": {"from": 5360, "to": 8640}, "text": " blue pixels", + "tokens": [tok(5360, 7000, " blue"), tok(7000, 8640, " pixels")]}, + {"offsets": {"from": 13120, "to": 16160}, "text": " which you know", + "tokens": [tok(13120, 9580, " which"), tok(13120, 10140, " you"), tok(13120, 10890, " know")]}, + ]})) + turns = [Turn(0.0, 8.8, "SPEAKER_00"), Turn(13.3, 16.0, "SPEAKER_00")] + assert mt.merge(mt.load_whisper_json(path), turns) == [ + "00:00:05 Speaker A: blue pixels", + "00:00:13 Speaker A: which you know", + ] + + +class TestDropOutsideSpeech: + """Whisper invents words in silence; the diarizer knows where the speech is.""" + + def test_merge_transcript_drop_outside_speech_removes_words_far_from_any_turn(self): + """Normal: a hallucinated run in a silent stretch goes; real words stay.""" + real = words(0.0, "Good", "morning.") + invented = words(60.0, "Thank", "you.", "Thank", "you.", step=5.0) + kept, dropped = mt.drop_outside_speech(real + invented, [Turn(0.0, 3.0, "SPEAKER_00")]) + assert kept == real + assert dropped == 4 + + def test_merge_transcript_drop_outside_speech_keeps_words_between_close_turns(self): + """Normal: a word in a short gap between two turns is speech the diarizer clipped.""" + units = words(0.0, "One.") + words(3.2, "and") + words(4.0, "two.") + turns = [Turn(0.0, 3.0, "SPEAKER_00"), Turn(4.0, 6.0, "SPEAKER_01")] + assert mt.drop_outside_speech(units, turns) == (units, 0) + + def test_merge_transcript_drop_outside_speech_word_exactly_at_the_margin_is_kept(self): + """Boundary: the margin is inclusive.""" + unit = Unit(5.0, 5.5, " edge") + assert mt.drop_outside_speech([unit], [Turn(0.0, 3.0, "S")], margin_s=2.0) == ([unit], 0) + + def test_merge_transcript_drop_outside_speech_word_just_past_the_margin_is_dropped(self): + """Boundary: one millisecond further and it goes.""" + unit = Unit(5.001, 5.5, " edge") + assert mt.drop_outside_speech([unit], [Turn(0.0, 3.0, "S")], margin_s=2.0) == ([], 1) + + def test_merge_transcript_drop_outside_speech_zero_margin_needs_contact_with_a_turn(self): + """Boundary: margin 0 keeps a word touching a turn and drops one that is not.""" + touching, apart = Unit(3.0, 3.4, " touch"), Unit(3.5, 3.9, " apart") + kept, dropped = mt.drop_outside_speech([touching, apart], [Turn(0.0, 3.0, "S")], margin_s=0.0) + assert (kept, dropped) == ([touching], 1) + + def test_merge_transcript_drop_outside_speech_before_the_first_turn_counts_too(self): + """Boundary: silence at the start of a recording.""" + early = Unit(1.0, 1.5, " Thanks.") + assert mt.drop_outside_speech([early], [Turn(30.0, 40.0, "S")]) == ([], 1) + + def test_merge_transcript_drop_outside_speech_no_units_is_a_no_op(self): + """Boundary: nothing in, nothing out.""" + assert mt.drop_outside_speech([], [Turn(0.0, 3.0, "S")]) == ([], 0) + + def test_merge_transcript_drop_outside_speech_no_turns_raises(self): + """Error: without turns there is no way to tell speech from silence.""" + with pytest.raises(ValueError, match="no speaker turns"): + mt.drop_outside_speech(words(0.0, "Hello."), []) + + def test_merge_transcript_drop_outside_speech_negative_margin_raises(self): + """Error: a negative margin is a caller bug.""" + with pytest.raises(ValueError, match="margin"): + mt.drop_outside_speech(words(0.0, "Hello."), [Turn(0.0, 3.0, "S")], margin_s=-1.0) + + def test_merge_transcript_main_silence_hallucinations_do_not_trip_the_loop_guard(self, tmp_path, capsys): + """Normal: a run of invented thank-yous in silence is dropped, not reported as a loop.""" + transcription = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}] + [ + {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."} + for i in range(12) + ] + whisper = tmp_path / "w.json" + whisper.write_text(json.dumps({"transcription": transcription})) + turns = tmp_path / "t.json" + turns.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(whisper), str(turns)]) == 0 + assert capsys.readouterr().out == "00:00:00 Speaker A: Good morning.\n" + + def test_merge_transcript_main_a_loop_inside_speech_still_fails(self, tmp_path, capsys): + """Error: a real repetition loop happens while someone is talking, and is still caught.""" + transcription = [ + {"offsets": {"from": i * 1_000, "to": i * 1_000 + 900}, "text": " where we're going to be"} + for i in range(60) # a full minute of the same phrase + ] + whisper = tmp_path / "w.json" + whisper.write_text(json.dumps({"transcription": transcription})) + turns = tmp_path / "t.json" + turns.write_text(json.dumps([{"start": 0.0, "end": 70.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(whisper), str(turns)]) == 1 + assert "looped" in capsys.readouterr().err + + +class TestCollapseRepetitions: + """A short stutter is collapsed to one occurrence; a long loop still fails the job.""" + + def test_merge_transcript_collapse_repetitions_short_loop_keeps_one_copy(self): + """Normal: a six-word phrase said four times in ten seconds becomes one phrase.""" + units = words(0.0, "So", "anyway,") + words(1.0, *("fair, it's not going to be".split() * 4), step=0.4) + words(12.0, "done.") + kept, collapsed = mt.collapse_repetitions(units) + assert " ".join(u.text.strip() for u in kept) == "So anyway, fair, it's not going to be done." + assert collapsed == [("fair, it's not going to be", 4)] + + def test_merge_transcript_collapse_repetitions_clean_units_are_untouched(self): + """Normal: ordinary speech passes through with nothing collapsed.""" + units = words(0.0, "We", "have", "detection", "today,", "and", "we", "have", "a", "plan.") + assert mt.collapse_repetitions(units) == (units, []) + + def test_merge_transcript_collapse_repetitions_two_loops_both_collapse(self): + """Normal: separate stutters are each collapsed and each reported.""" + units = (words(0.0, *("go back to this area".split() * 4), step=0.3) + + words(10.0, "then") + + words(11.0, *("where we're going to be".split() * 5), step=0.3)) + kept, collapsed = mt.collapse_repetitions(units) + assert " ".join(u.text.strip() for u in kept) == "go back to this area then where we're going to be" + assert collapsed == [("go back to this area", 4), ("where we're going to be", 5)] + + def test_merge_transcript_collapse_repetitions_three_repeats_are_speech(self): + """Boundary: three repeats is emphasis, not a loop, and stays.""" + units = words(0.0, *("this is the thing".split() * 3)) + assert mt.collapse_repetitions(units) == (units, []) + + def test_merge_transcript_collapse_repetitions_single_word_runs_stay(self): + """Boundary: "yeah yeah yeah yeah" is ordinary speech.""" + units = words(0.0, *(["yeah"] * 8)) + assert mt.collapse_repetitions(units) == (units, []) + + def test_merge_transcript_collapse_repetitions_loop_at_the_limit_is_collapsed(self): + """Boundary: a loop lasting exactly max_loop_s is still a short one.""" + units = words(0.0, *("we do not know where".split() * 4), step=1.5) # 20 words, 30.0 s + kept, collapsed = mt.collapse_repetitions(units, max_loop_s=30.0) + assert collapsed == [("we do not know where", 4)] and len(kept) == 5 + + def test_merge_transcript_collapse_repetitions_loop_past_the_limit_raises(self): + """Error: a loop longer than max_loop_s means real speech was lost, so the job fails.""" + units = words(0.0, *("we do not know where".split() * 4), step=1.6) # 32.0 s + with pytest.raises(ValueError, match="looped"): + mt.collapse_repetitions(units, max_loop_s=30.0) + + def test_merge_transcript_collapse_repetitions_segment_units_collapse_too(self): + """Boundary: whisper's segment fallback puts a whole phrase in one unit.""" + units = [Unit(i * 1.0, i * 1.0 + 0.9, " where we're going to be") for i in range(6)] + kept, collapsed = mt.collapse_repetitions(units) + assert len(kept) == 1 and collapsed == [("where we're going to be", 6)] + + def test_merge_transcript_collapse_repetitions_negative_limit_raises(self): + """Error: a negative limit is a caller bug.""" + with pytest.raises(ValueError, match="max_loop_s"): + mt.collapse_repetitions(words(0.0, "hi"), max_loop_s=-1.0) + + def test_merge_transcript_main_short_loop_is_collapsed_not_fatal(self, tmp_path, capsys): + """Normal: the CLI prints the collapsed transcript and exits 0.""" + words_ = "we have detection today " .split() + ("fair, it's not going to be " * 4).split() + "easy.".split() + segs = [{"offsets": {"from": i * 300, "to": i * 300 + 250}, "text": f" {w}"} for i, w in enumerate(words_)] + w = tmp_path / "w.json"; w.write_text(json.dumps({"transcription": segs})) + t = tmp_path / "t.json"; t.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}])) + assert mt.main([str(w), str(t)]) == 0 + assert capsys.readouterr().out == "00:00:00 Speaker A: we have detection today fair, it's not going to be easy.\n" + + +class TestCollapseRepetitionsInsideUnits: + """The repeat can live inside one unit's text, which is what whisper's segment fallback emits.""" + + def test_merge_transcript_collapse_repetitions_loop_inside_one_unit_is_collapsed(self): + """Error case turned regression: a single unit holding the phrase four times must not hang.""" + unit = Unit(0.0, 8.0, " " + " ".join(["we do not know where"] * 4)) + kept, collapsed = mt.collapse_repetitions(unit and [unit]) + assert [u.text for u in kept] == [" we do not know where"] + assert collapsed == [("we do not know where", 4)] + + def test_merge_transcript_collapse_repetitions_partial_unit_keeps_its_other_words(self): + """Boundary: a unit holding the last repeat and real words after it keeps the real words.""" + units = words(0.0, *("go back to this area".split() * 3), step=0.5) + [ + Unit(7.5, 9.0, " go back to this area and then we stopped.") + ] + kept, collapsed = mt.collapse_repetitions(units) + assert " ".join(u.text.strip() for u in kept) == "go back to this area and then we stopped." + assert collapsed == [("go back to this area", 4)] diff --git a/working/meeting-transcription-service/tests/test_ratio_transcribe.py b/working/meeting-transcription-service/tests/test_ratio_transcribe.py new file mode 100644 index 0000000..1a10b3e --- /dev/null +++ b/working/meeting-transcription-service/tests/test_ratio_transcribe.py @@ -0,0 +1,263 @@ +"""Tests for ratio-transcribe, the client. + +ssh and scp are replaced by fakes that act on a temp directory standing in for the +remote home, so the client's real logic (job ids, upload-then-rename, polling, +collecting, the local fallback) runs against a filesystem it can't tell from the host. +""" + +import json +import os +import subprocess +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parent.parent / "src" / "ratio-transcribe" +STATE = ".local/state/meeting-transcribe" + +FAKE_SSH = r"""#!/usr/bin/env bash +# ssh [-o k=v]... host command... -> run the command with HOME at the fake remote +[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255 +no_stdin="" +while [[ "$1" == -* ]]; do [[ "$1" == "-n" ]] && no_stdin=1; [[ "$1" == "-o" ]] && shift; shift; done +shift # host +printf '%s\n' "$*" >> "$FAKE_SSH_LOG" +if [[ -n "$no_stdin" ]]; then + ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" < /dev/null ) + rc=$? +else + ( cd "$FAKE_REMOTE_HOME" && HOME="$FAKE_REMOTE_HOME" bash -c "$*" ) + rc=$? + cat > /dev/null # like the real ssh, drain whatever stdin the command left behind +fi +if [[ "$*" == *"incoming/"* && "$*" == mv* && -n "${FAKE_ON_SUBMIT:-}" ]]; then + ( cd "$FAKE_REMOTE_HOME" && bash -c "$FAKE_ON_SUBMIT" ) +fi +exit $rc +""" + +FAKE_SCP = r"""#!/usr/bin/env bash +[[ -n "${FAKE_SSH_DOWN:-}" ]] && exit 255 +while [[ "$1" == -* ]]; do [[ "$1" == "-o" ]] && shift; shift; done +printf '%s -> %s\n' "$1" "$2" >> "$FAKE_SCP_LOG" +cp "$1" "$FAKE_REMOTE_HOME/${2#*:}" +""" + +# What the worker would do, compressed: finish or fail whatever sits in incoming/. +WORKER_OK = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d"); + printf '00:00:00 Speaker A: Hello from the host.\\n' > done/$id.txt; rm -rf "$d"; done""" +WORKER_FAIL = f"""cd {STATE}; mkdir -p failed; for d in incoming/*/; do id=$(basename "$d"); + printf 'stage: whisper\\nexit 3\\n' > failed/$id.log; rm -rf "$d"; done""" +WORKER_EMPTY = f"""cd {STATE}; mkdir -p done; for d in incoming/*/; do id=$(basename "$d"); + : > done/$id.txt; rm -rf "$d"; done""" + + +class Rig: + def __init__(self, tmp_path): + self.tmp = tmp_path + self.remote = tmp_path / "remote-home" + self.remote.mkdir() + self.local_home = tmp_path / "local-home" + self.local_home.mkdir() + self.bin = tmp_path / "bin" + self.bin.mkdir() + for name, body in (("ssh", FAKE_SSH), ("scp", FAKE_SCP)): + path = self.bin / name + path.write_text(body) + path.chmod(0o755) + self.ssh_log = tmp_path / "ssh.log" + self.scp_log = tmp_path / "scp.log" + self.audio = tmp_path / "meeting.m4a" + self.audio.write_bytes(b"pretend audio") + + def run(self, args=None, on_submit=WORKER_OK, **env_extra): + env = { + "PATH": f"{self.bin}:{os.environ['PATH']}", + "HOME": str(self.local_home), + "FAKE_REMOTE_HOME": str(self.remote), + "FAKE_SSH_LOG": str(self.ssh_log), + "FAKE_SCP_LOG": str(self.scp_log), + "TRANSCRIBE_HOST": "testhost", + "TRANSCRIBE_POLL": "0", + "TRANSCRIBE_TIMEOUT": "5", + } + if on_submit: + env["FAKE_ON_SUBMIT"] = on_submit + env.update({k: str(v) for k, v in env_extra.items()}) + if args is None: + args = [str(self.audio)] + return subprocess.run([str(SCRIPT), *args], env=env, capture_output=True, text=True, timeout=60) + + def state(self, *parts): + return self.remote.joinpath(STATE, *parts) + + def uploads(self): + return self.scp_log.read_text().splitlines() if self.scp_log.exists() else [] + + +@pytest.fixture +def rig(tmp_path): + return Rig(tmp_path) + + +class TestClientNormal: + def test_ratio_transcribe_new_recording_is_uploaded_and_transcript_printed(self, rig): + """Normal: upload, wait, print. stdout is the transcript and nothing else.""" + result = rig.run() + assert result.returncode == 0, result.stderr + assert result.stdout == "00:00:00 Speaker A: Hello from the host.\n" + assert len(rig.uploads()) == 1 + + def test_ratio_transcribe_job_file_carries_language_speakers_and_name(self, rig): + """Normal: the options reach the host inside job.json.""" + rig.run(args=[str(rig.audio), "es"], on_submit=None, SPEAKERS=3, TRANSCRIBE_TIMEOUT=0) + jobs = list(rig.state("incoming").glob("*/job.json")) + assert len(jobs) == 1 + assert json.loads(jobs[0].read_text()) == { + "language": "es", "speakers": 3, "original_name": "meeting.m4a", + } + assert (jobs[0].parent / "audio.m4a").read_bytes() == b"pretend audio" + + def test_ratio_transcribe_finished_job_is_collected_without_uploading_again(self, rig): + """Normal: rerunning after a dropped connection costs one ssh round trip.""" + first = rig.run() + rig.scp_log.unlink() + second = rig.run(on_submit=None) + assert second.returncode == 0 + assert second.stdout == first.stdout + assert rig.uploads() == [] + + def test_ratio_transcribe_job_id_depends_on_audio_and_options(self, rig): + """Normal: same file with a different speaker count is a different job.""" + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0, SPEAKERS=3) + assert len(list(rig.state("incoming").iterdir())) == 2 + + def test_ratio_transcribe_earlier_failure_is_retried(self, rig): + """Normal: a stale failure log does not block a fresh attempt.""" + assert rig.run(on_submit=WORKER_FAIL).returncode == 1 + result = rig.run(on_submit=WORKER_OK) + assert result.returncode == 0 + assert "Hello from the host" in result.stdout + assert list(rig.state("failed").glob("*.log")) == [] + + +class TestClientBoundary: + def test_ratio_transcribe_job_still_queued_is_not_uploaded_twice(self, rig): + """Boundary: a second run while the first job waits just joins the wait.""" + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + assert len(rig.uploads()) == 1 + + def test_ratio_transcribe_upload_lands_in_incoming_only_by_rename(self, rig): + """Boundary: the audio is copied into uploading/, never straight into incoming/.""" + rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + assert "/uploading/" in rig.uploads()[0] + assert list(rig.state("uploading").iterdir()) == [] + + def test_ratio_transcribe_awkward_filename_survives(self, rig): + """Boundary: spaces and quotes in the recording's name.""" + odd = rig.tmp / "someone's \"weekly\" sync.m4a" + odd.write_bytes(b"x") + result = rig.run(args=[str(odd)], on_submit=None, TRANSCRIBE_TIMEOUT=0) + job = json.loads(next(rig.state("incoming").glob("*/job.json")).read_text()) + assert job["original_name"] == "someone's \"weekly\" sync.m4a", result.stderr + + def test_ratio_transcribe_colon_in_a_relative_filename_still_uploads(self, rig): + """Boundary: scp reads "standup 14:30.m4a" as host:path; the client hands it an absolute path.""" + (rig.tmp / "standup 14:30.m4a").write_bytes(b"x") + env = { + "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home), + "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log), + "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK, + "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5", + } + result = subprocess.run( + [str(SCRIPT), "standup 14:30.m4a"], cwd=rig.tmp, env=env, capture_output=True, text=True, timeout=60, + ) + assert result.returncode == 0, result.stderr + assert rig.uploads()[0].startswith("/") + + def test_ratio_transcribe_host_unreachable_falls_back_to_local_worker(self, rig): + """Boundary: offline, the same queue and worker run on this machine.""" + worker = rig.tmp / "fake-worker" + worker.write_text( + "#!/usr/bin/env bash\n" + f"cd \"$HOME/{STATE}\" && mkdir -p done && for d in incoming/*/; do id=$(basename \"$d\");\n" + "printf '00:00:00 Speaker A: Local fallback.\\n' > done/$id.txt; rm -rf \"$d\"; done\n" + ) + worker.chmod(0o755) + result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(worker)) + assert result.returncode == 0, result.stderr + assert result.stdout == "00:00:00 Speaker A: Local fallback.\n" + assert "local" in result.stderr.lower() + + +class TestClientStdin: + def test_ratio_transcribe_leaves_the_callers_stdin_alone(self, rig): + """Boundary: in a `while read` loop the client must not eat the loop's input.""" + env = { + "PATH": f"{rig.bin}:{os.environ['PATH']}", "HOME": str(rig.local_home), + "FAKE_REMOTE_HOME": str(rig.remote), "FAKE_SSH_LOG": str(rig.ssh_log), + "FAKE_SCP_LOG": str(rig.scp_log), "FAKE_ON_SUBMIT": WORKER_OK, + "TRANSCRIBE_HOST": "testhost", "TRANSCRIBE_POLL": "0", "TRANSCRIBE_TIMEOUT": "5", + } + result = subprocess.run( + ["bash", "-c", '"$0" "$1" > /dev/null 2>&1; cat', str(SCRIPT), str(rig.audio)], + env=env, input="next line of the caller's loop\n", capture_output=True, text=True, timeout=60, + ) + assert result.stdout == "next line of the caller's loop\n" + + +class TestClientError: + def test_ratio_transcribe_no_arguments_prints_usage(self, rig): + """Error: usage.""" + result = rig.run(args=[]) + assert result.returncode == 1 and "Usage: ratio-transcribe" in result.stderr and result.stdout == "" + + def test_ratio_transcribe_missing_file_fails_before_any_ssh(self, rig): + """Error: no such recording.""" + result = rig.run(args=[str(rig.tmp / "nope.m4a")]) + assert result.returncode == 1 and "not found" in result.stderr + assert not rig.ssh_log.exists() + + @pytest.mark.parametrize("env", [{"SPEAKERS": "0"}, {"SPEAKERS": "three"}, {"MIN_SPEAKERS": "-1"}, + {"MIN_SPEAKERS": "4", "MAX_SPEAKERS": "2"}, {"SPEAKERS": "3", "MAX_SPEAKERS": "5"}]) + def test_ratio_transcribe_bad_speaker_settings_are_rejected(self, rig, env): + """Error: non-positive, non-numeric or contradictory counts.""" + result = rig.run(**env) + assert result.returncode == 1 and "speaker" in result.stderr.lower() + assert not rig.ssh_log.exists() + + def test_ratio_transcribe_bad_language_is_rejected(self, rig): + """Error: the language ends up in a job file and a command line.""" + result = rig.run(args=[str(rig.audio), "en; rm -rf /"]) + assert result.returncode == 1 and "language" in result.stderr.lower() + assert not rig.ssh_log.exists() + + def test_ratio_transcribe_failed_job_reports_the_log_and_prints_nothing(self, rig): + """Error: the host's failure log comes back on stderr.""" + result = rig.run(on_submit=WORKER_FAIL) + assert result.returncode == 1 + assert result.stdout == "" + assert "stage: whisper" in result.stderr + + def test_ratio_transcribe_timeout_says_the_job_is_still_running(self, rig): + """Error: giving up waiting is not the job failing.""" + result = rig.run(on_submit=None, TRANSCRIBE_TIMEOUT=0) + assert result.returncode == 1 + assert result.stdout == "" + assert "again" in result.stderr.lower() + + def test_ratio_transcribe_empty_transcript_is_a_failure(self, rig): + """Error: a zero-byte result is never passed off as a transcript.""" + result = rig.run(on_submit=WORKER_EMPTY, TRANSCRIBE_TIMEOUT=1) + assert result.returncode == 1 + assert result.stdout == "" + + def test_ratio_transcribe_offline_without_local_worker_names_what_is_missing(self, rig): + """Error: no host and nothing installed locally.""" + result = rig.run(FAKE_SSH_DOWN=1, TRANSCRIBE_WORKER=str(rig.tmp / "absent")) + assert result.returncode == 1 + assert "absent" in result.stderr diff --git a/working/meeting-transcription-service/tests/test_transcribe_worker.py b/working/meeting-transcription-service/tests/test_transcribe_worker.py new file mode 100644 index 0000000..bfffd31 --- /dev/null +++ b/working/meeting-transcription-service/tests/test_transcribe_worker.py @@ -0,0 +1,359 @@ +"""Tests for transcribe-worker: the queue drain on the transcription host. + +ffmpeg, whisper-cli and the diarizer are replaced by small fake executables (the +process boundary). The queue handling, the merge and the loop guard run for real. +""" + +import importlib.machinery +import importlib.util +import json +import os +import stat +import sys +from pathlib import Path + +import pytest + +SRC = Path(__file__).resolve().parent.parent / "src" +sys.path.insert(0, str(SRC)) + + +def _load_worker(): + loader = importlib.machinery.SourceFileLoader("transcribe_worker", str(SRC / "transcribe-worker")) + spec = importlib.util.spec_from_loader("transcribe_worker", loader) + assert spec is not None + module = importlib.util.module_from_spec(spec) + sys.modules["transcribe_worker"] = module # dataclasses looks the module up by name + loader.exec_module(module) + return module + + +worker = _load_worker() + +FAKE_FFMPEG = """#!/usr/bin/env bash +# last argument is the output; the one after -i is the input +while [[ $# -gt 1 ]]; do [[ "$1" == "-i" ]] && in="$2"; shift; done +cp "$in" "$1" +exit "${FAKE_FFMPEG_EXIT:-0}" +""" + +FAKE_WHISPER = """#!/usr/bin/env bash +printf '%s\\n' "$*" >> "$FAKE_LOG" +while [[ $# -gt 0 ]]; do [[ "$1" == "-of" ]] && prefix="$2"; shift; done +[[ "${FAKE_WHISPER_EXIT:-0}" == "0" ]] && cp "$FAKE_WHISPER_JSON" "$prefix.json" +exit "${FAKE_WHISPER_EXIT:-0}" +""" + +FAKE_DIARIZE = """#!/usr/bin/env bash +printf 'diarize %s\\n' "$*" >> "$FAKE_LOG" +[[ "${FAKE_DIARIZE_EXIT:-0}" == "0" ]] && cp "$FAKE_TURNS_JSON" "$2" +echo "fake diarizer says hello" >&2 +exit "${FAKE_DIARIZE_EXIT:-0}" +""" + + +def whisper_json(*words, step_ms=400): + return {"transcription": [ + {"offsets": {"from": i * step_ms, "to": i * step_ms + 300}, "text": f" {w}"} + for i, w in enumerate(words) + ]} + + +class Rig: + def __init__(self, tmp_path, monkeypatch): + self.home = tmp_path / "state" + self.bin = tmp_path / "bin" + self.bin.mkdir() + for name, body in (("ffmpeg", FAKE_FFMPEG), ("whisper-cli", FAKE_WHISPER), ("fake-diarize", FAKE_DIARIZE)): + path = self.bin / name + path.write_text(body) + path.chmod(path.stat().st_mode | stat.S_IXUSR) + self.log = tmp_path / "calls.log" + self.whisper_json = tmp_path / "whisper.json" + self.turns_json = tmp_path / "turns.json" + self.set_whisper(whisper_json("Good", "morning.")) + self.turns_json.write_text(json.dumps([{"start": 0.0, "end": 60.0, "speaker": "SPEAKER_00"}])) + monkeypatch.setenv("PATH", f"{self.bin}:{os.environ['PATH']}") + monkeypatch.setenv("FAKE_LOG", str(self.log)) + monkeypatch.setenv("FAKE_WHISPER_JSON", str(self.whisper_json)) + monkeypatch.setenv("FAKE_TURNS_JSON", str(self.turns_json)) + self.monkeypatch = monkeypatch + self.config = worker.Config( + home=self.home, + whisper_model=tmp_path / "model.bin", + diarize_cmd=[str(self.bin / "fake-diarize")], + threads=2, + ) + + def set_whisper(self, data): + self.whisper_json.write_text(json.dumps(data)) + + def submit(self, job_id, job=None, audio=b"audio", where="incoming"): + folder = self.home / where / job_id + folder.mkdir(parents=True) + (folder / "audio.m4a").write_bytes(audio) + if job is not False: + (folder / "job.json").write_text(json.dumps(job if job is not None else {"language": "en"})) + return folder + + def calls(self): + return self.log.read_text().splitlines() if self.log.exists() else [] + + +@pytest.fixture +def rig(tmp_path, monkeypatch): + return Rig(tmp_path, monkeypatch) + + +class TestDrainNormal: + def test_transcribe_worker_drain_one_job_writes_transcript_to_done(self, rig): + """Normal: a job in incoming/ ends as done/<id>.txt and leaves nothing behind.""" + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n" + assert not (rig.home / "incoming" / "job1").exists() + assert not (rig.home / "work" / "job1").exists() + assert not (rig.home / "failed" / "job1.log").exists() + + def test_transcribe_worker_drain_writes_metadata_beside_the_transcript(self, rig): + """Normal: done/<id>.json records what ran, for the client and for debugging.""" + rig.submit("job1", {"language": "en", "speakers": 3, "original_name": "standup.mkv"}) + worker.drain(rig.config) + meta = json.loads((rig.home / "done" / "job1.json").read_text()) + assert meta["original_name"] == "standup.mkv" + assert meta["speakers_found"] == 1 + assert meta["lines"] == 1 + assert meta["seconds"] >= 0 + + def test_transcribe_worker_drain_passes_language_and_speaker_count_through(self, rig): + """Normal: job options reach whisper and the diarizer.""" + rig.submit("job1", {"language": "es", "speakers": 3}) + worker.drain(rig.config) + whisper_call = next(c for c in rig.calls() if not c.startswith("diarize")) + diarize_call = next(c for c in rig.calls() if c.startswith("diarize")) + assert "-l es" in whisper_call + assert "-mc 0" in whisper_call and "-ojf" in whisper_call + assert diarize_call.endswith("--speakers 3") + + def test_transcribe_worker_drain_min_and_max_speakers_are_forwarded(self, rig): + """Normal: a range instead of an exact count.""" + rig.submit("job1", {"min_speakers": 2, "max_speakers": 5}) + worker.drain(rig.config) + diarize_call = next(c for c in rig.calls() if c.startswith("diarize")) + assert "--min-speakers 2" in diarize_call and "--max-speakers 5" in diarize_call + + def test_transcribe_worker_drain_processes_every_job_oldest_first(self, rig): + """Normal: the queue drains completely, in arrival order.""" + first = rig.submit("b-first") + rig.submit("a-second") + os.utime(first, (1, 1)) + assert worker.drain(rig.config) == {"done": 2, "failed": 0} + diarize_calls = [c for c in rig.calls() if c.startswith("diarize")] + assert "b-first" in diarize_calls[0] and "a-second" in diarize_calls[1] + + +class TestDrainBoundary: + def test_transcribe_worker_drain_empty_queue_is_a_no_op(self, rig): + """Boundary: nothing to do, and the folders get created.""" + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "incoming").is_dir() and (rig.home / "done").is_dir() + + def test_transcribe_worker_drain_ignores_uploads_still_in_flight(self, rig): + """Boundary: a dot-prefixed folder is left alone (second line of defence behind uploading/).""" + rig.submit(".tmp-job9") + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "incoming" / ".tmp-job9").exists() + + def test_transcribe_worker_drain_job_already_done_is_not_rerun(self, rig): + """Boundary: resubmitting finished work costs nothing.""" + (rig.home / "done").mkdir(parents=True) + (rig.home / "done" / "job1.txt").write_text("already here\n") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "already here\n" + assert not (rig.home / "incoming" / "job1").exists() + assert rig.calls() == [] + + def test_transcribe_worker_drain_missing_job_file_uses_defaults(self, rig): + """Boundary: audio with no job.json still transcribes, in English, count estimated.""" + rig.submit("job1", job=False) + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + + def test_transcribe_worker_drain_retries_a_job_left_in_work_by_a_crash(self, rig): + """Boundary: a job stranded in work/ is picked up again.""" + rig.submit("job1", where="work") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + + def test_transcribe_worker_drain_gives_up_on_a_job_that_keeps_crashing(self, rig): + """Boundary: the second stranding is a failure, not a loop.""" + rig.submit("job1", {"language": "en", "attempts": 2}, where="work") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "attempt" in (rig.home / "failed" / "job1.log").read_text() + + +class TestDrainError: + def test_transcribe_worker_drain_whisper_failure_is_logged_with_its_stage(self, rig): + """Error: whisper exits non-zero.""" + rig.monkeypatch.setenv("FAKE_WHISPER_EXIT", "3") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + log = (rig.home / "failed" / "job1.log").read_text() + assert "whisper" in log + assert not (rig.home / "done" / "job1.txt").exists() + assert not (rig.home / "work" / "job1").exists() + + def test_transcribe_worker_drain_diarizer_failure_keeps_its_stderr(self, rig): + """Error: the diarizer fails; its own words land in the log.""" + rig.monkeypatch.setenv("FAKE_DIARIZE_EXIT", "1") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + log = (rig.home / "failed" / "job1.log").read_text() + assert "diarize" in log and "fake diarizer says hello" in log + + def test_transcribe_worker_drain_looping_transcription_fails(self, rig): + """Error: whisper's repetition loop is a failed job, never a transcript.""" + rig.set_whisper(whisper_json(*("we do not know where we are going".split() * 5), step_ms=1000)) # 40 s + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "looped" in (rig.home / "failed" / "job1.log").read_text() + assert not (rig.home / "done" / "job1.txt").exists() + + def test_transcribe_worker_drain_short_loop_is_collapsed_and_recorded(self, rig): + """Normal: a stutter of a few seconds is collapsed to one copy and noted in the metadata.""" + rig.set_whisper(whisper_json("Okay,", *("fair, it's not going to be".split() * 4), "easy.")) + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Okay, fair, it's not going to be easy.\n" + meta = json.loads((rig.home / "done" / "job1.json").read_text()) + assert meta["loops_collapsed"] == [["fair, it's not going to be", 4]] + + def test_transcribe_worker_drain_malformed_job_file_fails_that_job_only(self, rig): + """Error: one bad job does not stop the queue.""" + bad = rig.submit("bad") + (bad / "job.json").write_text("{not json") + os.utime(bad, (1, 1)) + rig.submit("good") + assert worker.drain(rig.config) == {"done": 1, "failed": 1} + assert (rig.home / "done" / "good.txt").exists() + assert "job.json" in (rig.home / "failed" / "bad.log").read_text() + + def test_transcribe_worker_drain_job_without_audio_fails(self, rig): + """Error: a job folder holding no audio file.""" + folder = rig.submit("job1") + (folder / "audio.m4a").unlink() + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "audio" in (rig.home / "failed" / "job1.log").read_text() + + @pytest.mark.parametrize("bad", [{"speakers": 0}, {"speakers": "three"}, {"language": "en; rm -rf"}]) + def test_transcribe_worker_drain_rejects_bad_option_values(self, rig, bad): + """Error: options are validated before they reach a command line.""" + rig.submit("job1", bad) + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert rig.calls() == [] + + +class TestDrainSilenceHallucinations: + def _silent_stretch(self, rig): + spoken = [{"offsets": {"from": 0, "to": 900}, "text": " Good morning."}] + invented = [ + {"offsets": {"from": 60_000 + i * 5_000, "to": 64_000 + i * 5_000}, "text": " Thank you."} + for i in range(12) + ] + rig.set_whisper({"transcription": spoken + invented}) + rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])) + + def test_transcribe_worker_drain_words_invented_in_silence_are_dropped_not_failed(self, rig): + """Normal: a quiet meeting transcribes; the invented run never reaches the transcript.""" + self._silent_stretch(rig) + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 1, "failed": 0} + assert (rig.home / "done" / "job1.txt").read_text() == "00:00:00 Speaker A: Good morning.\n" + + def test_transcribe_worker_drain_metadata_counts_the_dropped_words(self, rig): + """Normal: the count is on record, so a transcript that lost a lot is visible.""" + self._silent_stretch(rig) + rig.submit("job1") + worker.drain(rig.config) + assert json.loads((rig.home / "done" / "job1.json").read_text())["dropped_outside_speech"] == 12 + + def test_transcribe_worker_drain_recording_with_no_speech_at_all_fails_cleanly(self, rig): + """Error: everything whisper produced sits in silence.""" + rig.set_whisper({"transcription": [{"offsets": {"from": 60_000, "to": 61_000}, "text": " Thank you."}]}) + rig.turns_json.write_text(json.dumps([{"start": 0.0, "end": 2.0, "speaker": "SPEAKER_00"}])) + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "no speech" in (rig.home / "failed" / "job1.log").read_text() + + +class TestRunStdin: + def test_transcribe_worker_run_gives_tools_no_stdin(self, tmp_path): + """Boundary: ffmpeg reads stdin when it can, which would eat a calling loop's input.""" + out = tmp_path / "stdin-target" + read_end, write_end = os.pipe() + saved = os.dup(0) + os.dup2(read_end, 0) + try: + worker._run("probe", ["bash", "-c", f"readlink /proc/self/fd/0 > {out}"]) + finally: + os.dup2(saved, 0) + for fd in (saved, read_end, write_end): + os.close(fd) + assert out.read_text().strip() == "/dev/null" + + +class TestDrainResilience: + def test_transcribe_worker_drain_null_attempts_counts_as_a_first_try(self, rig): + """Error: a job.json with "attempts": null must not take the worker down; it is a first try.""" + odd = rig.submit("odd", {"language": "en", "attempts": None}) + os.utime(odd, (1, 1)) + rig.submit("good") + assert worker.drain(rig.config) == {"done": 2, "failed": 0} + assert (rig.home / "done" / "odd.txt").exists() + assert (rig.home / "done" / "good.txt").exists() + + def test_transcribe_worker_drain_unexpected_error_in_one_job_is_logged_and_the_queue_goes_on(self, rig): + """Error: an exception the pipeline never anticipated lands in failed/ with its type, not on the run.""" + first = rig.submit("first") + os.utime(first, (1, 1)) + rig.submit("second") + + def explode(job: dict) -> list[str]: + raise RuntimeError("boom") + + rig.monkeypatch.setattr(worker, "_diarize_options", explode) + assert worker.drain(rig.config) == {"done": 0, "failed": 2} # both reached, neither crashed the run + for job_id in ("first", "second"): + log = (rig.home / "failed" / f"{job_id}.log").read_text() + assert "RuntimeError" in log and "boom" in log + assert not (rig.home / "work" / job_id).exists() + + def test_transcribe_worker_drain_unwritable_done_dir_is_a_failed_job_not_a_crash(self, rig): + """Error: an OSError while writing the transcript lands in failed/, and the run survives.""" + if os.geteuid() == 0: + pytest.skip("root ignores directory permissions") + (rig.home / "done").mkdir(parents=True) + (rig.home / "done").chmod(0o500) + try: + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "job1" in "".join(p.name for p in (rig.home / "failed").iterdir()) + finally: + (rig.home / "done").chmod(0o700) + + def test_transcribe_worker_drain_ffmpeg_failure_is_logged_with_its_stage(self, rig): + """Error: ffmpeg exits non-zero.""" + rig.monkeypatch.setenv("FAKE_FFMPEG_EXIT", "2") + rig.submit("job1") + assert worker.drain(rig.config) == {"done": 0, "failed": 1} + assert "ffmpeg" in (rig.home / "failed" / "job1.log").read_text() + + def test_transcribe_worker_drain_returns_at_once_when_another_worker_holds_the_lock(self, rig): + """Boundary: a second drain does not touch the queue while the first holds the lock.""" + import fcntl + rig.home.mkdir(parents=True, exist_ok=True) + rig.submit("job1") + with open(rig.home / "lock", "w", encoding="utf-8") as held: + fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB) + assert worker.drain(rig.config) == {"done": 0, "failed": 0} + assert (rig.home / "incoming" / "job1").exists() + assert rig.calls() == [] |