From d447db5ac9063ec12afb8f5bc7b3a0b05c75bad4 Mon Sep 17 00:00:00 2001 From: Craig Jennings Date: Fri, 25 Sep 2026 13:05:42 -0400 Subject: chore: stage the meeting transcription service for an install home This is a self-hosted transcription service: whisper.cpp for the words, pyannote for the speaker labels. It has been running on ratio since 2026-09-17, with velox as the offline fallback. A systemd path unit watches a filesystem queue and starts a oneshot worker per job. There is no network listener. ssh is the transport, systemd is the daemon, and the filesystem is the queue. It lands in working/ rather than its final home because two decisions come first. I haven't picked where the code lives in this repo. The Hugging Face token the diarization model needs on its first download also has to be handled, since anyone can read this repo. Neither blocks the service, which already runs. Both block the install path this repo owes it. The accompanying note lists what each machine needs. The torch venv is 1.3 GB and the whisper model is a separate download, so the note describes both rather than carrying them here. --- .../tests/test_diarize.py | 74 +++ .../tests/test_merge_transcript.py | 510 +++++++++++++++++++++ .../tests/test_ratio_transcribe.py | 263 +++++++++++ .../tests/test_transcribe_worker.py | 359 +++++++++++++++ 4 files changed, 1206 insertions(+) create mode 100644 working/meeting-transcription-service/tests/test_diarize.py create mode 100644 working/meeting-transcription-service/tests/test_merge_transcript.py create mode 100644 working/meeting-transcription-service/tests/test_ratio_transcribe.py create mode 100644 working/meeting-transcription-service/tests/test_transcribe_worker.py (limited to 'working/meeting-transcription-service/tests') 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/.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/.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() == [] -- cgit v1.2.3