import json import subprocess import tempfile import unittest from datetime import datetime, timezone from pathlib import Path from unittest.mock import Mock, patch from scripts import transcribe_meeting from src.meeting_lab.transcription.whisper import ( TranscriptionError, compact_transcript, transcribe_audio, ) RAW_RESULT = { "params": {"language": "de"}, "transcription": [ { "timestamps": {"from": "00:00:00,000", "to": "00:00:01,200"}, "offsets": {"from": 0, "to": 1200}, "text": " Hallo ", }, { "timestamps": {"from": "00:00:01,200", "to": "00:00:02,500"}, "offsets": {"from": 1200, "to": 2500}, "text": "Welt", }, ], } class FakeRunner: def __init__(self, *, returncode: int = 0, write_raw: bool = True) -> None: self.returncode = returncode self.write_raw = write_raw self.commands: list[list[str]] = [] def __call__(self, command, **kwargs): command = list(command) self.commands.append(command) if self.write_raw: prefix = Path(command[command.index("-of") + 1]) prefix.with_suffix(".json").write_text( json.dumps(RAW_RESULT, ensure_ascii=False), encoding="utf-8" ) return subprocess.CompletedProcess( command, self.returncode, stdout="", stderr="backend failure" ) class TranscriptionTests(unittest.TestCase): def create_inputs(self, root: Path) -> tuple[Path, Path]: audio = root / "meeting.wav" model = root / "ggml-model.bin" audio.write_bytes(b"audio") model.write_bytes(b"model") return audio, model def test_missing_audio_file_is_rejected_before_execution(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) model = root / "model.bin" model.write_bytes(b"model") runner = FakeRunner() with self.assertRaisesRegex(TranscriptionError, "Audio file does not exist"): transcribe_audio(root / "missing.wav", model, root / "out", runner=runner) self.assertEqual(runner.commands, []) def test_invalid_parameters_are_rejected(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) with self.assertRaisesRegex(TranscriptionError, "model does not exist"): transcribe_audio(audio, root / "missing.bin", root / "out") with self.assertRaisesRegex(TranscriptionError, "Language must"): transcribe_audio(audio, model, root / "out", " ") with self.assertRaisesRegex(TranscriptionError, "executable must"): transcribe_audio(audio, model, root / "out", executable="") with self.assertRaisesRegex(TranscriptionError, "Threads must"): transcribe_audio(audio, model, root / "out", threads="zero") def test_auto_threads_use_detected_physical_core_count_and_flash_attention(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) runner = FakeRunner() result = transcribe_audio( audio, model, root / "output", runner=runner, thread_detector=lambda: 6, ) command = runner.commands[0] self.assertEqual(command[command.index("-t") + 1], "6") self.assertIn("-fa", command) metadata = json.loads(result.runtime_metadata.read_text()) self.assertEqual(metadata["threads"], 6) self.assertEqual(metadata["threads_option"], "auto") self.assertTrue(metadata["flash_attention"]) self.assertEqual(metadata["whisper_executable"], "whisper-cli") def test_explicit_thread_count_is_forwarded_unchanged(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) runner = FakeRunner() result = transcribe_audio( audio, model, root / "output", threads=12, runner=runner ) command = runner.commands[0] self.assertEqual(command[command.index("-t") + 1], "12") metadata = json.loads(result.runtime_metadata.read_text()) self.assertEqual(metadata["threads"], 12) self.assertEqual(metadata["threads_option"], "12") def test_cli_forwards_explicit_thread_count_to_wrapper(self) -> None: fake_result = Mock( transcript_json=Path("output/transcript.json"), runtime_seconds=1.0 ) with patch.object( transcribe_meeting, "transcribe_audio", return_value=fake_result ) as transcribe: code = transcribe_meeting.main( [ "meeting.wav", "--model", "model.bin", "--output-dir", "output", "--threads", "6", ] ) self.assertEqual(code, 0) self.assertEqual(transcribe.call_args.kwargs["threads"], "6") def test_compact_transcript_has_established_schema(self) -> None: self.assertEqual( compact_transcript(RAW_RESULT), { "text": "Hallo Welt", "segments": [ {"id": 0, "start": 0.0, "end": 1.2, "text": "Hallo"}, {"id": 1, "start": 1.2, "end": 2.5, "text": "Welt"}, ], }, ) def test_success_preserves_raw_output_and_creates_metadata(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) runner = FakeRunner() times = iter([10.0, 12.25]) result = transcribe_audio( audio, model, root / "output", "de", runner=runner, monotonic=lambda: next(times), now=lambda: datetime(2026, 8, 21, 8, 30, tzinfo=timezone.utc), ) self.assertEqual(json.loads(result.raw_output.read_text()), RAW_RESULT) self.assertEqual(json.loads(result.transcript_json.read_text())["text"], "Hallo Welt") self.assertEqual( result.transcript_text.read_text(), "[00:00:00.000 - 00:00:01.200] Hallo\n" "[00:00:01.200 - 00:00:02.500] Welt\n", ) metadata = json.loads(result.runtime_metadata.read_text()) self.assertEqual(metadata["backend"], "whisper.cpp") self.assertEqual(metadata["language"], "de") self.assertEqual(metadata["duration_seconds"], 2.5) self.assertEqual(metadata["runtime_seconds"], 2.25) self.assertEqual(metadata["timestamp"], "2026-08-21T08:30:00+00:00") self.assertEqual(metadata["input_file"], str(audio.resolve())) self.assertEqual(metadata["model"], str(model.resolve())) def test_output_paths_are_deterministic(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) output = root / "chosen-output" result = transcribe_audio(audio, model, output, runner=FakeRunner()) self.assertEqual(result.raw_output, output / "whisper_raw.json") self.assertEqual(result.transcript_json, output / "transcript.json") self.assertEqual(result.transcript_text, output / "transcript.txt") self.assertEqual(result.runtime_metadata, output / "runtime_metadata.json") self.assertEqual( set(json.loads(result.runtime_metadata.read_text())["output_files"].values()), {"whisper_raw.json", "transcript.json", "transcript.txt", "runtime_metadata.json"}, ) def test_backend_failure_raises_and_preserves_available_raw_output(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) output = root / "output" with self.assertRaisesRegex(TranscriptionError, "exit code 7.*backend failure"): transcribe_audio(audio, model, output, runner=FakeRunner(returncode=7)) self.assertEqual(json.loads((output / "whisper_raw.json").read_text()), RAW_RESULT) self.assertFalse((output / "transcript.json").exists()) self.assertFalse((output / "runtime_metadata.json").exists()) def test_missing_backend_output_is_reported(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) audio, model = self.create_inputs(root) with self.assertRaisesRegex(TranscriptionError, "without producing JSON"): transcribe_audio( audio, model, root / "output", runner=FakeRunner(write_raw=False) ) if __name__ == "__main__": unittest.main()