ddidderr
3042b7743b
Sampled Gemini voice matching could leave one person under separate IDs across transcription chunks. Use hosted Precision-2 on the complete recording and align its exclusive speaker timeline with MAI word timing. Preserve transcription text, order, and timestamps; leave ambiguous word assignments unknown instead of merging identities to satisfy a count. Upload lossless FLAC, bound polling, retry transient status reads against the same job, and validate results before emitting stdout. Keep optional speaker-count and hosted model overrides. Disable MAI diarization and use 30-minute transcription uploads in both modes. Respect environment keys, keep authentication out of process arguments, and save optional diagnostics without audio, credentials, or signed upload URLs. Test Plan: - python3 -B -m unittest -v: all 101 regression tests passed, including mocked provider upload, polling, alignment, failure, and cleanup paths. - bash -n transcribe and shellcheck transcribe: passed. - git diff --cached --check: passed. - Prepared the 74-minute input locally and verified identical decoded PCM hashes for the normalized WAV and full-recording lossless FLAC. - Live API testing was explicitly deferred by the user; mocked results do not establish speaker accuracy on the recording. BREAKING CHANGE: Default diarization requires PYANNOTE_API_KEY or its 1Password reference. --speaker-model accepts hosted precision-2 or community-1 instead of OpenRouter audio-chat model IDs.
706 lines
37 KiB
Python
706 lines
37 KiB
Python
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()
|