import json import os from pathlib import Path import shutil import subprocess import tempfile 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") self.audio = self.path / "input with spaces.wav" def run_cli(self, *args, mode="ok", duration=70): make_audio(self.audio, duration) return subprocess.run([str(ROOT / "transcribe"), *args, str(self.audio)], env=dict(self.env, TRANSCRIBE_TEST_MODE=mode), capture_output=True, text=True, timeout=30) def requests(self): return [json.loads(line) for line in self.log.read_text().splitlines()] def assert_cleaned(self): self.assertTrue(all(not Path(r["staging"]).exists() for r in self.requests())) def assert_diarization(self, enabled): for request in self.requests(): self.assertEqual(request["provider"], { "options": {"azure": {"diarization": {"enabled": enabled}}}, }) self.assertIs( request["provider"]["options"]["azure"]["diarization"]["enabled"], enabled, ) 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, "Hello.\n\nboundary continues.\n\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_diarization(True) 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_diarization(True) 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_diarization(False) self.assert_cleaned() def test_short_text_requests_and_prints_speaker_labels(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_diarization(True) self.assertEqual(result.stderr, "") self.assert_cleaned() def test_chunked_text_prints_speaker_ids_with_request_scope(self): result = self.run_cli("--chunk-minutes", "1", mode="speakers") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(result.stdout, ( "[Chunk 1, Speaker 0]\nHello.\n\n" "[Chunk 1, Speaker 0]\nboundary\n\n" "[Chunk 2, Speaker 1]\ncontinues.\n\n" "[Chunk 2, Speaker 1]\nLast.\n" )) self.assertNotIn("Warning", result.stderr) self.assert_cleaned() def test_missing_speaker_labels_warn_instead_of_implying_detection_worked(self): for output in ((), ("--srt",)): with self.subTest(output=output): result = self.run_cli(*output, duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("returned no speaker labels", result.stderr) self.assertNotIn("[Speaker", result.stdout) result = self.run_cli(mode="partial-speakers", duration=5) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("some words without speaker labels", result.stderr) self.assertEqual(result.stdout, "[Speaker 0]\nHello\n\n[Speaker unknown]\nthere\n") 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_diarization(True) 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.log.unlink(missing_ok=True) 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_diarization(False) 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() # The opt-out must also preserve a limit specified after it. self.log.unlink() result = self.run_cli("--no-diarization", "--chunk-minutes", "1") self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual(len(self.requests()), 2) self.assert_diarization(False) self.assert_cleaned() def test_default_upload_limits_depend_on_diarization(self): for options, limit in (((), 600), (("--no-diarization",), 1800)): with self.subTest(options=options): self.log.unlink(missing_ok=True) result = self.run_cli(*options, duration=limit + 10) self.assertEqual(result.returncode, 0, result.stderr) requests = self.requests() self.assertEqual(len(requests), 2) self.assertGreaterEqual(requests[0]["upload_duration"], limit - 0.1) self.assertTrue(all(r["upload_duration"] <= limit for r in requests)) self.assert_diarization(not options) 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_diarization(True) self.log.unlink() 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_diarization(False) def test_failed_later_chunk_has_no_partial_stdout_and_cleans_up(self): for mode in ("http-error", "bad-json", "api-error", "missing-words"): with self.subTest(mode=mode): result = self.run_cli("--chunk-minutes", "1", mode=mode) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") 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()) if __name__ == "__main__": unittest.main()