import json import os from pathlib import Path import shutil import signal import subprocess import tempfile import time import unittest from test_chunks import make_audio ROOT = Path(__file__).resolve().parent TOOLS = ("bash", "ffmpeg", "ffprobe", "jq", "base64", "mktemp", "realpath") @unittest.skipUnless(all(shutil.which(tool) for tool in TOOLS), "CLI media dependencies missing") class TranscribeTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory() self.addCleanup(self.directory.cleanup) self.path = Path(self.directory.name) self.bin = self.path / "bin" self.bin.mkdir() for name in ("op", "curl"): fake = self.bin / name shutil.copyfile(ROOT / "tests" / "fake_provider.py", fake) fake.chmod(0o755) self.log = self.path / "requests.jsonl" self.env = dict(os.environ, PATH=f"{self.bin}{os.pathsep}{os.environ['PATH']}", TRANSCRIBE_TEST_LOG=str(self.log), TRANSCRIBE_TEST_MODE="ok", OPENROUTER_API_KEY="test-openrouter-key", PYANNOTE_API_KEY="test-pyannote-key", PYANNOTE_API_KEY_REF="") self.audio = self.path / "input with spaces.wav" def run_cli(self, *args, mode="ok", duration=70, env_overrides=None): make_audio(self.audio, duration) command = [str(ROOT / "transcribe"), *args, str(self.audio)] process = subprocess.Popen( command, env=dict(self.env, TRANSCRIBE_TEST_MODE=mode, **(env_overrides or {})), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True, ) try: stdout, stderr = process.communicate(timeout=45) except subprocess.TimeoutExpired: # Let the CLI clean its worker process groups before temporary # fixtures disappear, including a poll stuck on a simulated error. process.terminate() try: process.communicate(timeout=5) except subprocess.TimeoutExpired: process.kill() process.communicate(timeout=5) raise return subprocess.CompletedProcess(command, process.returncode, stdout, stderr) def requests(self): return sorted((json.loads(line) for line in self.log.read_text().splitlines()), key=lambda request: request["index"]) if self.log.exists() else [] def events(self): path = self.log.with_suffix(".events.jsonl") return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] def reset_logs(self): for suffix in (".jsonl", ".events.jsonl", ".pyannote.jsonl", ".credentials.jsonl"): self.log.with_suffix(suffix).unlink(missing_ok=True) def pyannote_requests(self): path = self.log.with_suffix(".pyannote.jsonl") return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] def speaker_request(self): return next(request["request"] for request in self.pyannote_requests() if request["stage"] == "submit") def assert_automatic_speaker_detection(self): request = self.speaker_request() self.assertEqual(request["model"], "precision-2") self.assertNotIn("numSpeakers", request) self.assertIs(request["exclusive"], True) self.assertIs(request["transcription"], False) def assert_processes_stopped(self): for event in self.events(): for pid in (event["pid"], event.get("child_pid")): if pid is None: continue try: # A killed descendant can await reaping by init; it must # not still be running or holding our temporary files. state = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()[0] self.assertEqual(state, "Z", f"Process {pid} survived worker cleanup") except FileNotFoundError: pass def assert_cleaned(self): self.assertTrue(all(not Path(r["staging"]).exists() for r in self.requests() + self.pyannote_requests() + self.events())) def test_parallel_default_and_job_limit_with_ordered_output(self): for options, limit, duration in (((), 8, 500), (("--jobs", "2"), 2, 180), (("--jobs", "1"), 1, 130)): with self.subTest(options=options): self.reset_logs() result = self.run_cli( *options, "--no-diarization", "--chunk-minutes", "1", mode="parallel", duration=duration, env_overrides={"TRANSCRIBE_TEST_BARRIER": str(limit)}, ) self.assertEqual(result.returncode, 0, result.stderr) requests = self.requests() self.assertEqual(result.stdout.split(), [f"Part{i}." for i in range(len(requests))]) active = maximum = 0 events = sorted(self.events(), key=lambda event: event["time"]) for event in events: active += 1 if event["event"] == "start" else -1 maximum = max(maximum, active) self.assertEqual(active, 0) self.assertEqual(maximum, limit) self.assertEqual(len({r["work"] for r in requests}), len(requests)) if limit > 1: first_finished = next(e for e in events if e["event"] == "finish") self.assertNotEqual(first_finished["index"], 0) refilled = next(e["time"] for e in events if e["event"] == "start" and e["index"] == limit) slow_finished = next(e["time"] for e in events if e["event"] == "finish" and e["index"] == 0) self.assertLess(refilled, slow_finished) self.assert_processes_stopped() self.assert_cleaned() def test_parallel_srt_matches_sequential_output(self): serial = self.run_cli("--no-diarization", "--jobs", "1", "--chunk-minutes", "1", "--srt", mode="parallel", duration=130) self.assertEqual(serial.returncode, 0, serial.stderr) self.reset_logs() parallel = self.run_cli( "--no-diarization", "--jobs", "3", "--chunk-minutes", "1", "--srt", mode="parallel", duration=130, env_overrides={"TRANSCRIBE_TEST_BARRIER": "3"}, ) self.assertEqual(parallel.returncode, 0, parallel.stderr) self.assertEqual(parallel.stdout, serial.stdout) self.assertEqual([block.splitlines()[0] for block in parallel.stdout.strip().split("\n\n")], ["1", "2", "3"]) self.assert_cleaned() def test_parallel_failure_cancels_workers_and_their_children(self): result = self.run_cli( "--jobs", "2", "--chunk-minutes", "1", mode="parallel-error", duration=180, env_overrides={"TRANSCRIBE_TEST_BARRIER": "2"}, ) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertIn("simulated concurrent failure", result.stderr) self.assertIn("stopping remaining workers", result.stderr) self.assertEqual({e["index"] for e in self.events() if e["event"] == "start"}, {0, 1}) self.assertTrue(any(e["event"] == "child" for e in self.events())) self.assert_processes_stopped() self.assert_cleaned() def test_interruptions_stop_workers_before_removing_staging(self): for signum in (signal.SIGINT, signal.SIGTERM): with self.subTest(signal=signum): self.reset_logs() make_audio(self.audio, 180) process = subprocess.Popen( [str(ROOT / "transcribe"), "--jobs", "2", "--chunk-minutes", "1", str(self.audio)], env=dict(self.env, TRANSCRIBE_TEST_MODE="blocked"), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True, ) try: deadline = time.monotonic() + 5 while time.monotonic() < deadline: events = self.events() if (sum(e["event"] == "start" for e in events) >= 2 and any(e["event"] == "child" for e in events)): break time.sleep(0.01) else: self.fail("Workers did not start") process.send_signal(signum) stdout, stderr = process.communicate(timeout=5) self.assertEqual(process.returncode, 128 + signum, stderr) self.assertEqual(stdout, "") self.assert_processes_stopped() self.assertTrue(all(not Path(e["staging"]).exists() for e in self.events())) finally: if process.poll() is None: process.kill() for event in self.events(): try: os.killpg(event["group"], signal.SIGKILL) except ProcessLookupError: pass process.communicate(timeout=5) def test_immediate_encoder_failure_is_not_lost_by_worker_waiting(self): encoder = self.bin / "ffmpeg" encoder.write_text( "#!/usr/bin/env python3\nimport os, sys\n" "if '-ss' in sys.argv:\n sys.exit(42)\n" f"os.execv({shutil.which('ffmpeg')!r}, ['ffmpeg', *sys.argv[1:]])\n" ) encoder.chmod(0o755) result = self.run_cli("--jobs", "64", "--chunk-minutes", "1", duration=500) self.assertEqual(result.returncode, 42, result.stderr) self.assertEqual(result.stdout, "") self.assertIn("stopping remaining workers", result.stderr) self.assertFalse(self.log.exists()) def test_invalid_jobs_fail_before_provider_access(self): for value in ("0", "65", "-1", "1.5", "abc", "9999999999999999999999"): with self.subTest(value=value): result = self.run_cli("--jobs", value, duration=1) self.assertNotEqual(result.returncode, 0) self.assertIn("integer between 1 and 64", result.stderr) result = subprocess.run([str(ROOT / "transcribe"), "--jobs"], env=self.env, capture_output=True, text=True, timeout=5) self.assertNotEqual(result.returncode, 0) self.assertIn("integer between 1 and 64", result.stderr) self.assertFalse(self.log.exists()) def assert_mai_diarization_disabled(self): for request in self.requests(): self.assertEqual(request["provider"], { "options": {"azure": {"diarization": {"enabled": False}}}, }) self.assertIs(request["provider"]["options"]["azure"]["diarization"]["enabled"], False) def test_long_text_uploads_bounded_mp3_and_stitches_once(self): result = self.run_cli("--chunk-minutes", "1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "[Speaker 0]\nHello.\n\n[Speaker 0]\nboundary continues.\n\n[Speaker 0]\nLast.\n" )) self.assertIn("chunk 2/2", result.stderr) requests = self.requests() self.assertEqual(len(requests), 2) self.assertTrue(all(r["upload_duration"] <= 60 for r in requests)) self.assertAlmostEqual(requests[0]["chunk"]["end"] - requests[1]["chunk"]["start"], 6) self.assert_mai_diarization_disabled() self.assert_cleaned() def test_whole_recording_upload_is_lossless_and_follows_all_transcription(self): result = self.run_cli("--chunk-minutes", "1", mode="reconcile") self.assertEqual(result.returncode, 0, result.stderr) requests = self.pyannote_requests() self.assertEqual([request["stage"] for request in requests], ["media", "upload", "submit", "poll"]) upload = requests[1] self.assertAlmostEqual(upload["upload_duration"], 70, places=3) self.assertEqual(upload["stream"], {"codec_name": "flac", "sample_rate": "16000", "channels": 1}) self.assertNotIn("authorization", upload["headers"]) self.assert_automatic_speaker_detection() events = self.events() last_transcription = max(e["time"] for e in events if e["index"] >= 0 and e["event"] == "finish") first_diarization = min(e["time"] for e in events if e["index"] == -1) self.assertGreater(first_diarization, last_transcription) self.assert_cleaned() def test_long_srt_uses_original_timeline_and_continuous_numbers(self): result = self.run_cli("--srt", "--chunk-minutes", "1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "1\n00:00:01,000 --> 00:00:01,400\nHello.\n\n" "2\n00:00:56,700 --> 00:00:57,800\nboundary continues.\n\n" "3\n00:01:05,000 --> 00:01:05,400\nLast.\n\n" )) self.assert_mai_diarization_disabled() self.assert_cleaned() def test_short_text_without_diarization_retains_provider_line_breaks(self): result = self.run_cli("--no-diarization", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "Provider's original formatting.\nNext line.\n") self.assertEqual(len(self.requests()), 1) self.assertIsNone(self.requests()[0]["response_format"]) self.assert_mai_diarization_disabled() self.assertEqual(self.pyannote_requests(), []) self.assert_cleaned() def test_short_text_requests_timestamps_and_uses_exclusive_speaker_timeline(self): result = self.run_cli(mode="speakers", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "[Speaker 0]\nHello\n\n[Speaker 1]\nthere\n") self.assertEqual(self.requests()[0]["response_format"], "verbose_json") self.assertEqual(self.requests()[0]["timestamp_granularities"], ["segment", "word"]) self.assert_automatic_speaker_detection() self.assert_mai_diarization_disabled() self.assert_cleaned() def test_chunked_text_has_speaker_markers_without_chunk_details(self): result = self.run_cli("--chunk-minutes", "1", mode="speakers") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "[Speaker 0]\nHello.\n\n" "[Speaker 0]\nboundary\n\n" "[Speaker 1]\ncontinues.\n\n" "[Speaker 1]\nLast.\n" )) self.assertNotIn("Chunk", result.stdout) self.assert_cleaned() def test_absent_and_partial_mai_speaker_labels_do_not_control_final_identity(self): for mode, expected in (("ok", "[Speaker 0]\nHello.\n"), ("partial-speakers", "[Speaker 0]\nHello\n\n[Speaker 1]\nthere\n")): with self.subTest(mode=mode): self.reset_logs() result = self.run_cli(mode=mode, duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, expected) self.assertNotIn("returned no speaker labels", result.stderr) self.assertNotIn("some words without speaker labels", result.stderr) self.assert_mai_diarization_disabled() self.assert_automatic_speaker_detection() def test_short_srt_still_requests_word_timestamps(self): result = self.run_cli("--srt", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "1\n00:00:01,000 --> 00:00:01,400\nHello.\n\n") self.assertEqual(self.requests()[0]["response_format"], "verbose_json") self.assertEqual(self.requests()[0]["timestamp_granularities"], ["segment", "word"]) self.assert_mai_diarization_disabled() def test_no_diarization_applies_to_text_srt_and_every_chunk(self): for output in ((), ("--srt",)): for chunked in (False, True): with self.subTest(output=output, chunked=chunked): self.reset_logs() options = ("--chunk-minutes", "1") if chunked else () result = self.run_cli( *output, *options, "--no-diarization", duration=70 if chunked else 5, ) self.assertEqual(result.returncode, 0, result.stderr) self.assertTrue(result.stdout) self.assertEqual(len(self.requests()), 2 if chunked else 1) self.assert_mai_diarization_disabled() self.assertEqual(self.pyannote_requests(), []) for request in self.requests(): if output or chunked: self.assertEqual(request["response_format"], "verbose_json") self.assertEqual(request["timestamp_granularities"], ["segment", "word"]) else: self.assertIsNone(request["response_format"]) self.assertIsNone(request["timestamp_granularities"]) self.assertLessEqual(request["upload_duration"], 60) self.assert_cleaned() self.reset_logs() result = self.run_cli("--no-diarization", "--chunk-minutes", "1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(len(self.requests()), 2) self.assert_mai_diarization_disabled() self.assert_cleaned() def test_no_diarization_removes_unsolicited_labels_and_skips_pyannote(self): for output in ((), ("--srt",)): for options in (("--no-diarization", "--chunk-minutes", "1", "--jobs", "1"), ("--chunk-minutes", "1", "--jobs", "2", "--no-diarization")): with self.subTest(output=output, options=options): self.reset_logs() result = self.run_cli(*output, *options, mode="reconcile") self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("AlphaLate.", result.stdout) self.assertNotIn("[Speaker", result.stdout) self.assertEqual(len(self.requests()), 2) self.assert_mai_diarization_disabled() self.assertEqual(self.pyannote_requests(), []) self.assert_cleaned() def test_default_transcription_upload_limit_is_thirty_minutes_in_both_modes(self): for options in ((), ("--no-diarization",)): with self.subTest(options=options): self.reset_logs() result = self.run_cli(*options, duration=1810) self.assertEqual(result.returncode, 0, result.stderr) requests = self.requests() self.assertEqual(len(requests), 2) self.assertGreaterEqual(requests[0]["upload_duration"], 1799.9) self.assertTrue(all(r["upload_duration"] <= 1800 for r in requests)) self.assert_mai_diarization_disabled() if not options: upload = next(request for request in self.pyannote_requests() if request["stage"] == "upload") self.assertAlmostEqual(upload["upload_duration"], 1810, places=3) self.assert_cleaned() def test_diarization_breaks_srt_at_speaker_changes(self): result = self.run_cli("--srt", mode="speakers", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "1\n00:00:01,000 --> 00:00:01,400\nHello\n\n" "2\n00:00:01,500 --> 00:00:01,900\nthere\n\n" )) self.assert_mai_diarization_disabled() self.reset_logs() result = self.run_cli("--no-diarization", "--srt", mode="speakers", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "1\n00:00:01,000 --> 00:00:01,900\nHello there\n\n") self.assert_mai_diarization_disabled() self.assertEqual(self.pyannote_requests(), []) def test_failed_later_chunk_has_no_partial_stdout_or_pyannote_job_and_cleans_up(self): for mode in ("http-error", "bad-json", "api-error", "missing-words"): with self.subTest(mode=mode): self.reset_logs() result = self.run_cli("--chunk-minutes", "1", mode=mode) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertEqual(self.pyannote_requests(), []) self.assertTrue(result.stderr) self.assert_cleaned() def test_invalid_limits_fail_before_provider_access(self): for value in ("0", "31", "abc", "1.5", "-1"): with self.subTest(value=value): result = self.run_cli("--chunk-minutes", value, duration=1) self.assertNotEqual(result.returncode, 0) self.assertIn("integer between 1 and 30", result.stderr) self.assertFalse(self.log.exists()) self.assertEqual(self.pyannote_requests(), []) def test_whole_timeline_merges_host_fragments_despite_conflicting_mai_ids(self): result = self.run_cli("--speakers", "2", "--chunk-minutes", "1", mode="reconcile") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "[Speaker 0]\nAlpha.\n\n[Speaker 0]\nAgain.\n\n[Speaker 1]\nBeta.\n\n" "[Speaker 0]\nAlphaLate.\n\n[Speaker 1]\nBetaLate.\n" )) self.assertEqual(self.speaker_request()["model"], "precision-2") self.assertEqual(self.speaker_request()["numSpeakers"], 2) self.assertEqual(len(self.requests()), 2) self.assert_mai_diarization_disabled() self.assert_cleaned() def test_bare_command_detects_speakers_without_a_known_count(self): for duration in (30, 1811): with self.subTest(duration=duration): self.reset_logs() result = self.run_cli(mode="reconcile", duration=duration) self.assertEqual(result.returncode, 0, result.stderr) expected = "[Speaker 0]\nAlpha.\n\n[Speaker 0]\nAgain.\n\n[Speaker 1]\nBeta.\n" if duration > 1800: expected += "\n[Speaker 0]\nAlphaLate.\n\n[Speaker 1]\nBetaLate.\n" self.assertEqual(result.stdout, expected) self.assert_automatic_speaker_detection() self.assertEqual(len(self.requests()), 2 if duration > 1800 else 1) self.assert_mai_diarization_disabled() self.assert_cleaned() def test_default_srt_detects_voices_without_speaker_options(self): result = self.run_cli("--srt", mode="reconcile", duration=30) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "1\n00:00:01,000 --> 00:00:03,000\nAlpha.\n\n" "2\n00:00:06,000 --> 00:00:08,000\nAgain.\n\n" "3\n00:00:12,000 --> 00:00:14,000\nBeta.\n\n" )) self.assert_automatic_speaker_detection() self.assertEqual(len(self.requests()), 1) self.assert_mai_diarization_disabled() self.assert_cleaned() def test_supported_speaker_model_and_count_overrides(self): for model, count in (("community-1", ()), ("precision-2", ("--speakers", "2"))): with self.subTest(model=model): self.reset_logs() result = self.run_cli(*count, "--speaker-model", model, mode="reconcile", duration=30) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "[Speaker 0]\nAlpha.\n\n[Speaker 0]\nAgain.\n\n[Speaker 1]\nBeta.\n" )) self.assertEqual(self.speaker_request()["model"], model) if count: self.assertEqual(self.speaker_request()["numSpeakers"], 2) else: self.assertNotIn("numSpeakers", self.speaker_request()) self.assert_cleaned() def test_pyannote_srt_preserves_words_and_original_timestamps(self): serial = self.run_cli("--jobs", "1", "--speakers", "2", "--srt", "--chunk-minutes", "1", mode="reconcile") self.assertEqual(serial.returncode, 0, serial.stderr) self.assertEqual(serial.stdout, ( "1\n00:00:01,000 --> 00:00:03,000\nAlpha.\n\n" "2\n00:00:06,000 --> 00:00:08,000\nAgain.\n\n" "3\n00:00:12,000 --> 00:00:14,000\nBeta.\n\n" "4\n00:01:00,000 --> 00:01:02,000\nAlphaLate.\n\n" "5\n00:01:05,000 --> 00:01:07,000\nBetaLate.\n\n" )) self.assert_cleaned() self.reset_logs() parallel = self.run_cli("--jobs", "2", "--speakers", "2", "--srt", "--chunk-minutes", "1", mode="reconcile") self.assertEqual(parallel.returncode, 0, parallel.stderr) self.assertEqual(parallel.stdout, serial.stdout) self.assert_cleaned() def test_empty_transcription_and_empty_speaker_timeline_are_valid(self): result = self.run_cli(mode="silence", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "") self.assert_automatic_speaker_detection() self.assert_cleaned() def test_partial_timeline_leaves_unsupported_words_unknown(self): result = self.run_cli(mode="pyannote-partial-timeline") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "[Speaker 0]\nHello.\n\n" "[Speaker unknown]\nboundary continues.\n\n" "[Speaker unknown]\nLast.\n" )) self.assertIn("ambiguous or missing speaker timing", result.stderr) self.assert_cleaned() def test_pyannote_created_running_succeeded_lifecycle(self): result = self.run_cli(mode="pyannote-lifecycle", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "[Speaker 0]\nHello.\n") polls = [request for request in self.pyannote_requests() if request["stage"] == "poll"] self.assertEqual([request["response"]["status"] for request in polls], ["created", "running", "succeeded"]) self.assert_cleaned() def test_transient_poll_failure_retries_the_same_job_without_resubmitting(self): result = self.run_cli(mode="pyannote-transient-poll", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, "[Speaker 0]\nHello.\n") requests = self.pyannote_requests() self.assertEqual(sum(request["stage"] == "submit" for request in requests), 1) polls = [request for request in requests if request["stage"] == "poll"] self.assertEqual(len(polls), 2) self.assertEqual(polls[0]["url"], polls[1]["url"]) self.assertIn("retrying the same job", result.stderr) self.assert_cleaned() def test_failed_or_invalid_pyannote_result_emits_no_partial_transcript_and_cleans_up(self): for mode in ("pyannote-media-http-error", "pyannote-upload-http-error", "pyannote-submit-http-error", "pyannote-poll-http-error", "pyannote-transport-error", "pyannote-bad-json", "pyannote-failed", "pyannote-canceled", "pyannote-missing-timeline", "pyannote-malformed-timeline", "pyannote-empty-timeline", "pyannote-wrong-job", "pyannote-output-error"): with self.subTest(mode=mode): self.reset_logs() result = self.run_cli(mode=mode, duration=5) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertTrue(self.pyannote_requests()) self.assertTrue(result.stderr) self.assert_cleaned() def test_invalid_speaker_options_fail_before_credentials_or_provider_access(self): options = [("--speakers", value) for value in ("0", "65", "-1", "1.5", "abc", "99999999999999999999")] options += [("--speaker-model", value) for value in ("", "-bad", "bad model", "a\nb", "google/gemini-3.8-flash", "test/audio")] options += [("--no-diarization", "--speakers", "2"), ("--speakers", "2", "--no-diarization"), ("--no-diarization", "--speaker-model", "precision-2"), ("--speaker-model", "precision-2", "--no-diarization")] for args in options: with self.subTest(options=args): result = self.run_cli(*args, duration=1) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertIn("--speaker", result.stderr) for option in ("--speakers", "--speaker-model"): result = subprocess.run([str(ROOT / "transcribe"), option], env=self.env, capture_output=True, text=True, timeout=5) self.assertNotEqual(result.returncode, 0) self.assertIn(option, result.stderr) self.assertFalse(self.log.with_suffix(".credentials.jsonl").exists()) self.assertFalse(self.log.exists()) self.assertEqual(self.pyannote_requests(), []) def test_explicit_api_keys_are_used_without_secret_manager_access(self): result = self.run_cli(duration=5, env_overrides={"TRANSCRIBE_TEST_OP_FAIL": "1"}) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.requests()[0]["authorization"], "Bearer test-openrouter-key") self.assertFalse(self.log.with_suffix(".credentials.jsonl").exists()) self.assertNotIn("test-openrouter-key", result.stdout + result.stderr) self.assertNotIn("test-pyannote-key", result.stdout + result.stderr) def test_missing_pyannote_key_fails_before_transcription(self): result = self.run_cli(duration=5, env_overrides={"PYANNOTE_API_KEY": "", "PYANNOTE_API_KEY_REF": ""}) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertIn("PYANNOTE_API_KEY", result.stderr) self.assertEqual(self.requests(), []) self.assertEqual(self.pyannote_requests(), []) def test_blank_or_multiline_keys_fail_before_provider_access(self): for key in ("PYANNOTE_API_KEY", "OPENROUTER_API_KEY"): for value in (" ", "\t", "key\nInjected: header", "key\rInjected: header"): with self.subTest(key=key, value=value): self.reset_logs() result = self.run_cli(duration=1, env_overrides={key: value}) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertIn(key, result.stderr) self.assertEqual(self.requests(), []) self.assertEqual(self.pyannote_requests(), []) def test_no_diarization_does_not_require_pyannote_credentials(self): result = self.run_cli("--no-diarization", duration=5, env_overrides={"PYANNOTE_API_KEY": "", "PYANNOTE_API_KEY_REF": ""}) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(self.pyannote_requests(), []) self.assertFalse(self.log.with_suffix(".credentials.jsonl").exists()) def test_configured_secret_references_are_read_only_when_key_is_missing(self): result = self.run_cli(duration=5, env_overrides={ "OPENROUTER_API_KEY": "", "PYANNOTE_API_KEY": "", "PYANNOTE_API_KEY_REF": "op://test/pyannote/credential", }) self.assertEqual(result.returncode, 0, result.stderr) access = [json.loads(line)["args"] for line in self.log.with_suffix(".credentials.jsonl").read_text().splitlines()] self.assertCountEqual(access, [ ["read", "op://API Keys/openrouter-mai-transcribe-2/credential"], ["read", "op://test/pyannote/credential"], ]) self.assertEqual(self.requests()[0]["authorization"], "Bearer test-op-api-key") self.assertNotIn("test-op-api-key", result.stdout + result.stderr) def test_failed_credential_lookup_has_no_provider_side_effects(self): for missing in ("OPENROUTER_API_KEY", "PYANNOTE_API_KEY"): with self.subTest(missing=missing): self.reset_logs() result = self.run_cli(duration=5, env_overrides={ missing: "", "PYANNOTE_API_KEY_REF": "op://test/pyannote/credential", "TRANSCRIBE_TEST_OP_FAIL": "1", }) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertEqual(self.requests(), []) self.assertEqual(self.pyannote_requests(), []) def test_diagnostics_support_review_and_rerender_without_audio_credentials_or_urls(self): diagnostics = self.path / "review artifacts" result = self.run_cli("--diagnostics", str(diagnostics), "--chunk-minutes", "1", mode="reconcile-diagnostics-metadata") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual({path.name for path in diagnostics.iterdir()}, {"chunks.json", "transcript.json", "diarization.json", "labeled.json"}) transcript = json.loads((diagnostics / "transcript.json").read_text()) labeled = json.loads((diagnostics / "labeled.json").read_text()) saved_job = json.loads((diagnostics / "diarization.json").read_text()) self.assertEqual(set(saved_job), {"jobId", "status", "output"}) self.assertEqual(set(saved_job["output"]), {"diarization", "exclusiveDiarization"}) self.assertEqual(saved_job["output"]["exclusiveDiarization"][0]["confidence"], {"voice_a": 99.0}) for timeline in saved_job["output"].values(): for segment in timeline: self.assertLessEqual(set(segment), {"start", "end", "speaker", "confidence"}) for original, assigned in zip(transcript["words"], labeled["words"], strict=True): self.assertEqual({key: original[key] for key in ("word", "start", "end")}, {key: assigned[key] for key in ("word", "start", "end")}) saved = "\n".join(path.read_text() for path in diagnostics.iterdir()) for forbidden in ("test-openrouter-key", "test-pyannote-key", "fixture-signed-url", "https://", "media://"): self.assertNotIn(forbidden, saved) rerender = subprocess.run(["python3", "-B", str(ROOT / "transcript.py"), str(diagnostics / "labeled.json")], capture_output=True, text=True, timeout=5) self.assertEqual(rerender.returncode, 0, rerender.stderr) self.assertEqual(rerender.stdout, result.stdout) self.assert_cleaned() def test_diagnostics_refuses_existing_directory_before_provider_access(self): diagnostics = self.path / "already exists" diagnostics.mkdir() original = diagnostics / "keep.txt" original.write_text("do not replace") result = self.run_cli("--diagnostics", str(diagnostics), duration=5) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") self.assertEqual(original.read_text(), "do not replace") self.assertEqual(self.requests(), []) self.assertEqual(self.pyannote_requests(), []) def test_interrupting_pyannote_stops_descendants_and_cleans_up(self): make_audio(self.audio, 5) process = subprocess.Popen( [str(ROOT / "transcribe"), str(self.audio)], env=dict(self.env, TRANSCRIBE_TEST_MODE="pyannote-blocked"), stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, start_new_session=True, ) try: deadline = time.monotonic() + 8 while time.monotonic() < deadline: if any(e["event"] == "child" and e["index"] == -1 for e in self.events()): break time.sleep(0.01) else: self.fail("Pyannote job polling did not start") process.send_signal(signal.SIGTERM) stdout, stderr = process.communicate(timeout=5) self.assertEqual(process.returncode, 128 + signal.SIGTERM, stderr) self.assertEqual(stdout, "") self.assert_processes_stopped() self.assert_cleaned() finally: if process.poll() is None: process.kill() for event in self.events(): try: os.killpg(event["group"], signal.SIGKILL) except ProcessLookupError: pass process.communicate(timeout=5) if __name__ == "__main__": unittest.main()