From 975053f1789771ba5cb9dbe50ba7fe0aa396f0ab Mon Sep 17 00:00:00 2001 From: historia Date: Fri, 28 Aug 2026 00:59:47 -0400 Subject: fix: silently failing whisper transcription --- app/tests/test_backends_audiocpp.py | 163 ++++++++++++++++++++++++++++--- app/tests/test_tts.py | 190 ++++++++++++++++++++++++++++++++++++ 2 files changed, 342 insertions(+), 11 deletions(-) (limited to 'app/tests') diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 3e97b82..6f97438 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -1,5 +1,6 @@ """Tests for the audio.cpp backend setup module (backends/audiocpp.py).""" +import argparse import io import json import sys @@ -983,9 +984,10 @@ class TranscribeWavDirTests(unittest.TestCase): self._td.cleanup() def test_transcribes_to_stem_map_with_absolute_paths(self): - transcripts = {str(self.narrator): "First.", - str(self.other): "Second."} - with patch.object(make_server.voices, "transcribe_reference_audio", + transcripts = {str(self.narrator): ("First.", "ok"), + str(self.other): ("Second.", "ok")} + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", side_effect=lambda path, model_name="base": transcripts[path]): result = make_server.voices.transcribe_wav_dir( @@ -994,14 +996,27 @@ class TranscribeWavDirTests(unittest.TestCase): self.assertEqual(result["narrator"], "First.") def test_failed_transcription_keeps_empty_string(self): - with patch.object(make_server.voices, "transcribe_reference_audio", - return_value=None): + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=(None, "no speech detected")): result = make_server.voices.transcribe_wav_dir([self.narrator], "base") self.assertEqual(result["narrator"], "") + def test_failed_transcription_prints_the_reason(self): + buffer = io.StringIO() + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=(None, "faster_whisper heard no speech")), \ + redirect_stdout(buffer): + make_server.voices.transcribe_wav_dir([self.narrator], "base") + output = buffer.getvalue() + self.assertIn("No transcript for 'narrator'", output) + self.assertIn("faster_whisper heard no speech", output) + def test_whisper_model_name_passed_through(self): - with patch.object(make_server.voices, "transcribe_reference_audio", - return_value="text") as mock_transcribe: + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=("text", "ok")) as mock_transcribe: make_server.voices.transcribe_wav_dir([self.narrator], "large-v3") self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") @@ -1015,6 +1030,122 @@ class TranscribeWavDirTests(unittest.TestCase): self.assertIn("other|World.", text) +class TranscribePlanTests(unittest.TestCase): + """_transcribe: plan application and transcript wipe protection.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.folder = Path(self._td.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.args = argparse.Namespace(input_dir=self.folder, + whisper_model="base") + + def tearDown(self): + self._td.cleanup() + + def test_all_mode_retranscribes_everything(self): + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=("New words.", "ok")): + transcripts, write = make_server.voices._transcribe( + self.args, {"mode": "all", "missing": [], "existing": {}}) + self.assertTrue(write) + self.assertEqual(transcripts, {"narrator": "New words."}) + + def test_empty_retranscription_keeps_existing_transcript(self): + # A failed re-transcription must never overwrite known-good text + # with a blank: a blank prompt_text entry makes the server reject + # every clone request for that voice. + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=(None, "backend broken")), \ + redirect_stdout(io.StringIO()) as buffer: + transcripts, write = make_server.voices._transcribe( + self.args, {"mode": "all", "missing": [], + "existing": {"narrator": "Good words."}}) + self.assertTrue(write) + self.assertEqual(transcripts, {"narrator": "Good words."}) + self.assertIn("Kept the existing transcript for 'narrator'", + buffer.getvalue()) + + def test_missing_mode_merges_new_with_existing(self): + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=("Fresh text.", "ok")): + transcripts, _write = make_server.voices._transcribe( + self.args, {"mode": "missing", "missing": [self.narrator], + "existing": {}}) + self.assertEqual(transcripts, {"narrator": "Fresh text."}) + + def test_unusable_backend_warns_with_the_reason(self): + buffer = io.StringIO() + with patch.object(make_server.voices, "whisper_backend_problem", + return_value="faster_whisper is installed but " + "failed to import: boom"), \ + patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=("text", "ok")), \ + redirect_stdout(buffer): + make_server.voices._transcribe( + self.args, {"mode": "all", "missing": [], "existing": {}}) + output = buffer.getvalue() + self.assertIn("No usable Whisper backend", output) + self.assertIn("failed to import: boom", output) + + +class WizardTranscribeStepTests(unittest.TestCase): + """The setup lane's transcribe step: rc reflects unusable transcripts.""" + + def setUp(self): + self._td = tempfile.TemporaryDirectory() + self.folder = Path(self._td.name) + (self.folder / "narrator.wav").write_bytes(b"x") + self.args = argparse.Namespace(input_dir=None, whisper_model="base") + self.settings = { + "audiocpp_dir": self.folder, + "wav_dir": self.folder, + "include_clone": True, + "plan": {"mode": "all", "missing": [], "existing": {}}, + "build": None, + "model_entries": [], + } + + def tearDown(self): + self._td.cleanup() + + def _transcribe_step(self): + lanes = make_server.wizard._execute_lanes(self.settings, self.args) + return lanes[0].steps[0] + + def test_blank_transcripts_fail_the_step(self): + step = self._transcribe_step() + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=(None, "broken backend")), \ + redirect_stdout(io.StringIO()) as buffer: + rc = step.work(None, None) + self.assertEqual(rc, 1) + self.assertIn("No transcript for: narrator", buffer.getvalue()) + + def test_good_transcripts_pass_the_step(self): + step = self._transcribe_step() + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + return_value=("Words.", "ok")), \ + redirect_stdout(io.StringIO()): + self.assertEqual(step.work(None, None), 0) + + def test_no_clone_families_passes_without_transcribing(self): + self.settings["include_clone"] = False + self.settings["plan"] = None + step = self._transcribe_step() + with patch.object(make_server.voices, + "transcribe_reference_audio_detailed") as mock_transcribe: + self.assertEqual(step.work(None, None), 0) + mock_transcribe.assert_not_called() + + class DesignPackageTests(unittest.TestCase): """Voice-design package detection.""" @@ -1594,14 +1725,24 @@ class NonInteractiveMainTests(unittest.TestCase): argv = ["backends/audiocpp.py"] + argv transcribe_effect = transcribe if transcribe is not None \ else MagicMock() + + def detailed(path, model_name="base"): + result = transcribe_effect(path, model_name=model_name) + if isinstance(result, tuple): + return result + return (result, "ok" if result + else "faster_whisper is not installed (test stub)") + with patch.object(sys, "argv", argv), \ patch.object(make_server.build, "find_local_checkout", return_value=None if no_checkout else self.checkout), \ - patch.object(make_server.voices, "transcribe_reference_audio", - side_effect=transcribe_effect), \ - patch.object(make_server.voices, "whisper_backend_available", - return_value=whisper): + patch.object(make_server.voices, + "transcribe_reference_audio_detailed", + side_effect=detailed), \ + patch.object(make_server.voices, "whisper_backend_problem", + return_value=None if whisper else + "faster_whisper is not installed"): return make_server.wizard.main() def _args(self, *extra): diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index cf67c6c..ce2dbb6 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -39,7 +39,10 @@ from converter.clients import ( QwenTTSClient, audiocpp_entry_voice_capability, normalize_language, + transcribe_reference_audio_detailed, + whisper_backend_problem, ) +from converter.clients.base import NonRetryableTTSError from converter.converter import AudiobookConverter # Chunks folder handed to clients whose tests never write chunk files. @@ -1408,6 +1411,83 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): self.assertIn("500", str(ctx.exception)) self.assertIn("bad voice", str(ctx.exception)) + def test_reference_text_error_is_not_retryable(self): + # Qwen3-TTS Base cloning without a server-side transcript fails + # identically on every attempt: the error must carry the fix + # (prompt_text / x_vector_only_mode) and skip the retry budget. + client = self._make_client(preset_mode=True, voice="narrator") + error = urllib.error.HTTPError( + "http://127.0.0.1:8080/v1/audio/speech", 500, + "Server Error", {}, + io.BytesIO(b'{"error":{"message":"Qwen3 voice clone ICL mode ' + b'requires reference text","type":"server_error"}}')) + with patch("converter.clients.faster.urllib.request.urlopen", + side_effect=error): + with self.assertRaises(NonRetryableTTSError) as ctx: + client._request_wav("Hello.") + message = str(ctx.exception) + self.assertIn("requires reference text", message) + self.assertIn("'narrator'", message) + self.assertIn("prompt_text", message) + self.assertIn("x_vector_only_mode", message) + + def test_model_contract_error_is_not_retryable(self): + client = self._make_client(preset_mode=True, voice="narrator") + error = urllib.error.HTTPError( + "http://127.0.0.1:8080/v1/audio/speech", 500, + "Server Error", {}, + io.BytesIO(b'{"error":{"message":"model contract spec not found ' + b"for family 'qwen3_tts' (provide --model-spec-override)\"}}")) + with patch("converter.clients.faster.urllib.request.urlopen", + side_effect=error): + with self.assertRaises(NonRetryableTTSError) as ctx: + client._request_wav("Hello.") + message = str(ctx.exception) + self.assertIn("not retryable", message) + self.assertIn("model contract spec not found for family 'qwen3_tts'", + message) + self.assertIn("--model-spec-override", message) + + def test_unknown_model_id_error_is_not_retryable(self): + client = self._make_client(preset_mode=True, voice="narrator") + error = urllib.error.HTTPError( + "http://127.0.0.1:8080/v1/audio/speech", 500, + "Server Error", {}, + io.BytesIO(b'{"error":{"message":"unknown model id: nope"}}')) + with patch("converter.clients.faster.urllib.request.urlopen", + side_effect=error): + with self.assertRaises(NonRetryableTTSError) as ctx: + client._request_wav("Hello.") + message = str(ctx.exception) + self.assertIn("not retryable", message) + self.assertIn("unknown model id: nope", message) + + def test_unmatched_server_error_stays_retryable(self): + # Only known-deterministic fragments skip the retry budget; device + # hiccups, OOM, and anything unrecognized keep the plain error the + # retry loop has always retried. + client = self._make_client() + error = urllib.error.HTTPError( + "http://127.0.0.1:8080/v1/audio/speech", 500, + "Server Error", {}, + io.BytesIO(b'{"error":{"message":"CUDA error at ggml-cuda.cu"}}')) + with patch("converter.clients.faster.urllib.request.urlopen", + side_effect=error): + with self.assertRaises(RuntimeError) as ctx: + client._request_wav("Hello.") + self.assertNotIsInstance(ctx.exception, NonRetryableTTSError) + self.assertIn("CUDA error", str(ctx.exception)) + + def test_non_retryable_error_skips_remaining_attempts(self): + client = self._make_client() + with patch.object(client, "generate_chunk", + side_effect=NonRetryableTTSError("nope")) as mock_gen, \ + patch("converter.clients.base.time.sleep") as mock_sleep: + with self.assertRaises(NonRetryableTTSError): + client.process_chunk_with_retry(1, "Hello.") + self.assertEqual(mock_gen.call_count, 1) + mock_sleep.assert_not_called() + def test_transient_failure_fails_the_chunk_attempt(self): # Retrying is the chunk-level policy's job # (process_chunk_with_retry); one generate_chunk call makes one @@ -1474,6 +1554,116 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): self.assertEqual(remaining, ["chunk_0001.wav"]) +class TranscribeReasonTests(unittest.TestCase): + """transcribe_reference_audio_detailed: a reason for every empty result. + + The audio.cpp setup prints the reason per voice, so each failure class + must be distinguishable: missing package vs broken import vs transcribe + error vs a silent no-speech result. + """ + + def _transcribe_with_models(self, models, spec_present=True): + """Run one detailed transcription with _cached_model stubbed. + + MODELS maps backend name -> model object (or exception instance to + raise in its place). The whisper fallback sees its own entry or a + ModuleNotFoundError so no real package import ever happens; + importlib.util.find_spec is pinned so the not-installed vs + installed-but-broken distinction is deterministic in any env. + """ + def fake_cached(key, loader): + backend = key[0] + entry = models.get(backend) + if isinstance(entry, Exception): + raise entry + return entry + with patch("converter.clients.transcribe._cached_model", + side_effect=fake_cached), \ + patch("importlib.util.find_spec", + return_value=MagicMock() if spec_present else None): + return transcribe_reference_audio_detailed("clip.wav") + + def test_success_returns_text_and_ok(self): + model = MagicMock() + model.transcribe.return_value = (iter([MagicMock(text=" Hello. ")]), + MagicMock()) + text, reason = self._transcribe_with_models( + {"faster_whisper": model, "whisper": ModuleNotFoundError()}) + self.assertEqual(text, "Hello.") + self.assertEqual(reason, "ok") + + def test_missing_backend_is_not_called_broken(self): + text, reason = self._transcribe_with_models({ + "faster_whisper": ModuleNotFoundError( + "No module named 'faster_whisper'"), + "whisper": ModuleNotFoundError("No module named 'whisper'"), + }, spec_present=False) + self.assertIsNone(text) + self.assertIn("faster_whisper is not installed", reason) + self.assertIn("whisper is not installed", reason) + + def test_broken_import_is_distinguished_from_missing(self): + text, reason = self._transcribe_with_models({ + "faster_whisper": ImportError( + "Error loading shared library ld-linux-x86-64.so.2"), + "whisper": ModuleNotFoundError("No module named 'whisper'", + name="whisper"), + }) + self.assertIsNone(text) + self.assertIn("faster_whisper is installed but failed to import", + reason) + self.assertIn("ld-linux-x86-64.so.2", reason) + self.assertIn("whisper is not installed", reason) + + def test_transcribe_error_carries_the_exception(self): + model = MagicMock() + model.transcribe.side_effect = RuntimeError("decode failed") + text, reason = self._transcribe_with_models( + {"faster_whisper": model, + "whisper": ModuleNotFoundError("No module named 'whisper'")}) + self.assertIsNone(text) + self.assertIn("faster_whisper transcription failed: decode failed", + reason) + + def test_empty_result_reports_no_speech(self): + model = MagicMock() + model.transcribe.return_value = (iter([]), MagicMock()) + text, reason = self._transcribe_with_models( + {"faster_whisper": model, + "whisper": ModuleNotFoundError("No module named 'whisper'")}) + self.assertIsNone(text) + self.assertIn("faster_whisper heard no speech", reason) + + def test_whisper_fallback_used_when_faster_whisper_fails(self): + failing = MagicMock() + failing.transcribe.side_effect = RuntimeError("boom") + good = MagicMock() + # The openai-whisper interface returns a dict with "text". + good.transcribe.return_value = {"text": " Hi. "} + text, reason = self._transcribe_with_models( + {"faster_whisper": failing, "whisper": good}) + self.assertEqual(text, "Hi.") + self.assertEqual(reason, "ok") + + def test_backend_problem_reports_broken_import(self): + def fake_import(name, *args, **kwargs): + raise ImportError("lib load failure") + with patch("builtins.__import__", side_effect=fake_import), \ + patch("importlib.util.find_spec", return_value=MagicMock()): + problem = whisper_backend_problem() + self.assertIn("faster_whisper is installed but failed to import", + problem) + self.assertIn("whisper is installed but failed to import", problem) + + def test_backend_problem_none_when_a_backend_imports(self): + def fake_import(name, *args, **kwargs): + if name == "faster_whisper": + return MagicMock() + raise ImportError("should not be probed") + with patch("builtins.__import__", side_effect=fake_import): + self.assertIsNone(whisper_backend_problem()) + + class AudioCppHeartbeatTests(unittest.TestCase): """The heartbeat reports chunk progress while a request generates.""" -- cgit v1.2.3