diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_audiobook_cli.py | 8 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 64 | ||||
| -rw-r--r-- | app/tests/test_runview.py | 132 | ||||
| -rw-r--r-- | app/tests/test_tts.py | 149 |
4 files changed, 353 insertions, 0 deletions
diff --git a/app/tests/test_audiobook_cli.py b/app/tests/test_audiobook_cli.py index f5dddf4..f1eb8e8 100644 --- a/app/tests/test_audiobook_cli.py +++ b/app/tests/test_audiobook_cli.py @@ -494,7 +494,15 @@ class AllModelsConvertTests(unittest.TestCase): self.assertEqual(events[4], {"kind": "book", "index": 2, "total": 2, "name": "book.txt", "model": "m2"}) + # Passthrough events are stamped with the model too, so the run + # view can attribute a chunk failure to its model. + self.assertEqual(events[1], + {"kind": "chunks", "total": 3, "model": "m1"}) + self.assertEqual(events[2], + {"kind": "chunk_done", "chunk": 1, "total": 3, + "model": "m1"}) self.assertEqual(events[3]["model"], "m1") + self.assertEqual(events[5]["model"], "m2") self.assertEqual(events[7]["model"], "m2") self.assertEqual(events[8], {"kind": "done", "ok": 2, "total": 2, diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index b1a91ed..80a97d0 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -1218,6 +1218,52 @@ class ConvertFlowTests(unittest.TestCase): "beta": [("book.txt", "planned")]}) self.assertNotIn("planned", kwargs) + def test_all_pick_skips_non_narrating_families_with_a_notice(self): + # Speech-to-speech-only families (PersonaPlex) cannot narrate text: + # every request would fail, so the All pick drops them, records a + # run notice naming what was skipped, and keeps them single-pickable. + spec_cache = audiocpp_client._FAMILY_SPECS + spec_cache["personaplex"] = {"tasks": ["s2s"]} + self.addCleanup(spec_cache.pop, "personaplex", None) + self._patch_remote( + [{"id": "alpha", "family": "higgs_audio_tts", "task": "tts"}, + {"id": "plex", "family": "personaplex", "task": "tts"}], + voices=["narrator"]) + self._mock_preflight() + self._answer_form(backend="audiocpp-remote", + model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="narrator", instructions="") + cmd = self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + kwargs = cmd[2] + self.assertEqual(kwargs["model_ids"], ["alpha"]) + self.assertEqual(kwargs["model_voices"], {"alpha": "narrator"}) + self.assertEqual(kwargs["run_notice"], + "skipped non-TTS model(s): plex") + + def test_all_pick_refuses_when_every_model_is_non_narrating(self): + # With nothing left to generate with after the skip, the All pick + # is refused up front instead of starting a doomed run. + spec_cache = audiocpp_client._FAMILY_SPECS + spec_cache["personaplex"] = {"tasks": ["s2s"]} + self.addCleanup(spec_cache.pop, "personaplex", None) + self._patch_remote( + [{"id": "plex-1", "family": "personaplex", "task": "tts"}, + {"id": "plex-2", "family": "personaplex", "task": "tts"}], + voices=["narrator"]) + self._mock_preflight() + self._answer_form(backend="audiocpp-remote", + model_id=hub.AUDIOCPP_MODEL_ALL, + audiocpp_voice="narrator", instructions="") + self._convert(None, [self._remote("audiocpp", "audio.cpp")]) + fields = self.tui.forms_seen[0][1] + voice_field = self._field("audiocpp_voice") + self._field("model_id")["value"] = hub.AUDIOCPP_MODEL_ALL + error = voice_field["validate"]("narrator") + self.assertIsNotNone(error) + self.assertIn("plex-1", error) + self.assertIn("plex-2", error) + self.assertIn("synthesize text", error) + def test_all_voice_falls_back_per_capability(self): # A CustomVoice entry cannot clone: it synthesizes with a built-in # speaker (the pick when it names one, the first speaker when it @@ -2448,6 +2494,24 @@ class PrepareRunConfigTests(unittest.TestCase): self.assertIn("clon", cfg.notice) self.assertNotIn("restarted", cfg.notice) + def test_run_notice_is_popped_and_joined_with_the_server_notice(self): + # The form's pre-flight warning (e.g. the All run's skipped + # non-narrating models) rides to the run view's notice line, and + # multiple notices accumulate instead of overwriting each other. + spec = self._spec("audiocpp", "http://127.0.0.1:8080") + status = BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=True, servers=[spec]) + kwargs = {"run_notice": "skipped non-TTS model(s): plex", + "restart_server": "audiocpp", "audiocpp_rehost": True} + with patch.object(hub, "detect_all", return_value=[status]), \ + patch("backends.common.server_running", + return_value=True), \ + patch.object(hub.servers, "alive", return_value=True): + cfg = hub._prepare_run_config("audiocpp", kwargs) + self.assertNotIn("run_notice", kwargs) + self.assertIn("skipped non-TTS model(s): plex", cfg.notice) + self.assertIn("re-hosted clone-only", cfg.notice) + def test_stop_and_exit_travels_on_the_config_not_the_kwargs(self): # The run-view toggle is not a converter kwarg: it moves onto the # config (and defaults to off when the form did not send it). diff --git a/app/tests/test_runview.py b/app/tests/test_runview.py index f8ab452..4e3f9fe 100644 --- a/app/tests/test_runview.py +++ b/app/tests/test_runview.py @@ -118,6 +118,97 @@ class StateTransitionTests(_FakeTui, unittest.TestCase): self.assertEqual(view.phase, "error") self.assertTrue(view.error_message) + def test_chunk_failure_message_names_the_model_and_reason(self): + # An "All" run stamps chunk_failed with the generating model and + # the server's error detail; the message carries both, and the + # results row keeps the reason (book_done(ok=False) has none). + view, _ = self.make_view() + view.handle_event({"kind": "book", "index": 1, "total": 3, + "name": "b.txt", "model": "m1"}) + view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1, + "error": "vocoder backend buffer allocation " + "failed", "model": "m1"}) + self.assertEqual(view.error_message, + "m1: chunk 1/1 failed — vocoder backend buffer " + "allocation failed") + view.handle_event({"kind": "book_done", "name": "b.txt", + "ok": False, "model": "m1"}) + self.assertEqual(view.book_results, + [("b.txt", False, [], + "vocoder backend buffer allocation failed", + "m1")]) + + def test_chunk_failure_without_detail_keeps_the_count(self): + # No server error detail: the message (and the row reason) still + # say which chunk of how many failed. + view, _ = self.make_view() + view.handle_event({"kind": "book", "index": 1, "total": 1, + "name": "b", "model": "m1"}) + view.handle_event({"kind": "chunk_failed", "chunk": 2, "total": 4, + "model": "m1"}) + self.assertEqual(view.error_message, "m1: chunk 2/4 failed") + view.handle_event({"kind": "book_done", "name": "b", "ok": False, + "model": "m1"}) + self.assertEqual(view.book_results[0][3], "chunk 2/4 failed") + + def test_new_book_clears_a_stale_failure_message(self): + # The heart of the "Chunk 1/1 failed persisted for every later + # model" bug: a failure message from one book of an "All" run must + # not linger under the next book's progress. + view, _ = self.make_view() + view.handle_event({"kind": "book", "index": 1, "total": 2, + "name": "b.txt", "model": "m1"}) + view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1, + "error": "boom", "model": "m1"}) + view.handle_event({"kind": "book_done", "name": "b.txt", + "ok": False, "model": "m1"}) + self.assertTrue(view.error_message) + view.handle_event({"kind": "book", "index": 2, "total": 2, + "name": "b.txt", "model": "m2"}) + self.assertEqual(view.error_message, "") + self.assertEqual(view._book_error, "") + + def test_done_message_lists_the_failed_models(self): + # The terminal "done" screen names what failed instead of showing + # the last chunk failure (or a bare count). + view, _ = self.make_view() + for index, (model, ok) in enumerate( + [("m1", False), ("m2", True), ("m3", False)], 1): + view.handle_event({"kind": "book", "index": index, "total": 3, + "name": "b.txt", "model": model}) + view.handle_event({"kind": "book_done", "name": "b.txt", + "ok": ok, "model": model}) + view.handle_event({"kind": "done", "ok": 1, "total": 3}) + self.assertEqual(view.phase, "error") + self.assertEqual(view.error_message, + "2 of 3 book(s) failed: m1, m3") + + def test_done_message_caps_the_failed_model_list(self): + # More failures than fit the two detail lines: the list is capped + # with the leftover count (the rows carry the full list). + view, _ = self.make_view() + for index in range(1, 8): + view.handle_event({"kind": "book", "index": index, "total": 7, + "name": "b.txt", "model": f"m{index}"}) + view.handle_event({"kind": "book_done", "name": "b.txt", + "ok": False, "model": f"m{index}"}) + view.handle_event({"kind": "done", "ok": 0, "total": 7}) + self.assertEqual( + view.error_message, + "7 of 7 book(s) failed: m1, m2, m3, m4, m5, … +2 more") + + def test_done_all_failed_rows_only_names_models_of_failures(self): + # A model that cannot even start (audiobook.py's constructor + # failure path) reports book_failed without a model field: the + # name stands in for the model in the summary. + view, _ = self.make_view() + view.handle_event({"kind": "book_failed", "name": "m1", + "error": "voice 'x' is not available", + "files": []}) + view.handle_event({"kind": "done", "ok": 0, "total": 1}) + self.assertEqual(view.error_message, + "1 of 1 book(s) failed: m1") + def test_server_exit_during_boot_is_error(self): view, _ = self.make_view() view.handle_event({"kind": "starting", "name": "audiocpp"}) @@ -273,6 +364,47 @@ class RenderTests(_FakeTui, unittest.TestCase): text = self._strings(screen) self.assertIn("book.txt — m1", text) + def test_summary_shows_failed_rows_first_with_their_reason(self): + # With dozens of "All"-run results, the failures must be visible + # without scrolling past the successes, and each failed row says + # why it failed. + view, screen = self.make_view() + view.handle_event({"kind": "book", "index": 1, "total": 3, + "name": "book.txt", "model": "m1"}) + view.handle_event({"kind": "book_done", "name": "book.txt", + "ok": True, "model": "m1"}) + view.handle_event({"kind": "book", "index": 2, "total": 3, + "name": "book.txt", "model": "m2"}) + view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1, + "error": "missing model root: dac", + "model": "m2"}) + view.handle_event({"kind": "book_done", "name": "book.txt", + "ok": False, "model": "m2"}) + view.handle_event({"kind": "book", "index": 3, "total": 3, + "name": "book.txt", "model": "m3"}) + view.handle_event({"kind": "book_done", "name": "book.txt", + "ok": True, "model": "m3"}) + view.handle_event({"kind": "done", "ok": 2, "total": 3}) + view.render() + text = self._strings(screen) + fail_pos = text.index("[FAIL]") + ok_pos = text.index("[OK]") + self.assertLess(fail_pos, ok_pos) + self.assertIn("book.txt — m2: missing model root: dac", text) + + def test_progress_line_shows_the_model_of_a_chunk_failure(self): + # While the run is live, the message line names the model that + # failed (not just the chunk counters). + view, screen = self.make_view() + view.handle_event({"kind": "book", "index": 4, "total": 30, + "name": "book.txt", "model": "DramaBox-GGUF"}) + view.handle_event({"kind": "chunk_failed", "chunk": 1, "total": 1, + "error": "vocoder backend buffer allocation " + "failed", "model": "DramaBox-GGUF"}) + view.render() + text = self._strings(screen) + self.assertIn("DramaBox-GGUF: chunk 1/1 failed", text) + def test_stopping_status_shows_elapsed(self): view, screen = self.make_view() view.handle_event({"kind": "book", "index": 1, "total": 1, diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index 3c039ab..8403913 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -41,8 +41,10 @@ from converter.clients import ( FasterTTSClient, QwenTTSClient, audiocpp_entry_voice_capability, + audiocpp_family_narrates, audiocpp_family_voice_policy, audiocpp_request_error, + audiocpp_script_input, normalize_language, transcribe_reference_audio_detailed, whisper_backend_problem, @@ -1033,6 +1035,26 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): self.assertIs(client.profile, AUDIOCPP_DEFAULT_FAMILY_PROFILE) self.assertEqual(client.profile.language_style, AUDIOCPP_LANG_OMIT) + def test_speech_to_speech_only_family_rejected_at_connect(self): + # A family whose model spec has no text-synthesis task + # (PersonaPlex, s2s-only) cannot narrate regardless of its hosted + # task: the run fails at connect with a pointer at the other + # entries instead of a mid-run 500 on every request. + cache = audiocpp_client._FAMILY_SPECS + cache.clear() + cache.update({"personaplex": {"tasks": ["s2s"]}}) + self.addCleanup(cache.clear) + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": _AUDIOCPP_MODEL_ID, "family": "personaplex", + "task": "tts"}, + {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]}) + message = str(ctx.exception) + self.assertIn("personaplex", message) + self.assertIn("speech-to-speech", message) + self.assertIn(_AUDIOCPP_MODEL_ID, message) + self.assertIn("tts-1", message) + def test_speaker_mode_rejected_for_clone_only_family(self): # An unknown family (no spec, no built-in speakers) keeps the # conservative clone-only default: without --voice the run fails @@ -1173,6 +1195,36 @@ class AudioCppFamilyVoicePolicyTests(unittest.TestCase): AUDIOCPP_VOICE_REQUIRED) +class AudioCppFamilyNarratesTests(unittest.TestCase): + """The per-family "can this model narrate text at all" resolver.""" + + def setUp(self): + # Seed the spec cache like AudioCppFamilyVoicePolicyTests: the + # tests stay hermetic without a downloaded checkout. + cache = audiocpp_client._FAMILY_SPECS + cache.clear() + cache.update({ + "personaplex": {"tasks": ["s2s"]}, + "vibevoice": {"tasks": ["tts"]}, + "glm_tts": {"tasks": ["tts", "clone"]}, + }) + self.addCleanup(cache.clear) + + def test_speech_to_speech_only_family_cannot_narrate(self): + self.assertFalse(audiocpp_family_narrates("personaplex")) + + def test_tts_family_narrates(self): + self.assertTrue(audiocpp_family_narrates("vibevoice")) + + def test_mixed_family_narrates(self): + self.assertTrue(audiocpp_family_narrates("glm_tts")) + + def test_unlisted_family_is_never_hidden(self): + # No local spec for the family: conservatively narrating (None), + # so an unknown family is never silently dropped from the menu. + self.assertIsNone(audiocpp_family_narrates("brand_new_family")) + + class AudioCppPlainTtsModeTests(unittest.TestCase): """Plain-TTS runs: families that synthesize without a reference voice.""" @@ -1292,6 +1344,79 @@ class AudioCppCloneOnlyErrorTests(unittest.TestCase): self.assertIn("model busy", str(exc)) +class AudioCppDeterministicErrorTests(unittest.TestCase): + """Deterministic audio.cpp failures are not retried. + + The fragments come from real 500 bodies (incomplete model packages, + s2s-only families, unresolvable voices, VRAM exhaustion); retrying + the identical request cannot succeed, so in an "All" run every broken + model must be skipped in one attempt with its reason on screen. + """ + + def _error(self, message): + return audiocpp_request_error( + 500, json.dumps({"error": {"message": message}})) + + def test_missing_model_package_file_is_not_retryable(self): + exc = self._error( + "failed to load model resources using builtin model spec for " + "family 'glm_tts' source 'safetensors': missing model package " + "file 'tokenizer_merges': /models/GLM-TTS_Q8") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_missing_model_root_is_not_retryable(self): + exc = self._error( + "failed to select safetensors source from builtin model spec " + "for family 'outetts': missing model root: dac=/models/OuteTTS") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_ambiguous_gguf_directory_is_not_retryable(self): + exc = self._error( + "model directory contains 4 GGUF files: /models/MiniMax-H3-Q4-" + "GGUF; found: audio_vae_folded_f16.gguf, dit.gguf, ...") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_missing_companion_model_path_is_not_retryable(self): + exc = self._error( + "model path does not exist: /tmp/audiocpp-gguf/" + "MioCodec-25Hz-44.1kHz-v2") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_speech_to_speech_only_family_is_not_retryable_with_a_hint(self): + exc = self._error("PersonaPlex supports only speech-to-speech sessions") + self.assertIsInstance(exc, NonRetryableTTSError) + self.assertIn("cannot generate audiobooks", str(exc)) + + def test_unresolvable_clone_voice_is_not_retryable_with_a_hint(self): + exc = self._error("Vevo2 requires target_voice or voice speaker audio") + self.assertIsInstance(exc, NonRetryableTTSError) + self.assertIn("reference audio", str(exc)) + + def test_unscripted_vibevoice_prompt_is_not_retryable(self): + exc = self._error("VibeVoice prompt has no valid Speaker N: lines") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_reference_over_encoder_capacity_is_not_retryable_with_a_hint(self): + exc = self._error("VoxCPM2 AudioVAE encoder sample capacity exceeded") + self.assertIsInstance(exc, NonRetryableTTSError) + self.assertIn("trim", str(exc)) + + def test_allocation_failures_are_not_retryable(self): + # VRAM does not change between attempts of a sequential run (the + # "All" loop unloads models between books, not between retries). + self.assertIsInstance( + self._error("DramaBox vocoder backend buffer allocation failed"), + NonRetryableTTSError) + self.assertIsInstance( + self._error("failed to allocate MOSS codec encoder forward graph"), + NonRetryableTTSError) + + def test_max_tokens_before_eoc_stays_retryable(self): + # Proven transient: a request that hit it has succeeded on retry. + exc = self._error("Higgs TTS generation reached max_tokens before EOC") + self.assertNotIsInstance(exc, NonRetryableTTSError) + + class AudioCppTTSClientRequestTests(unittest.TestCase): """The /v1/audio/speech payload and response validation.""" @@ -1514,6 +1639,30 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): with self.assertRaises(RuntimeError): client._request_wav("Hello.") + def test_vibevoice_prompt_is_flattened_into_one_script_line(self): + # VibeVoice parses the prompt line by line and silently drops + # every line without a "Speaker N:" prefix, so the client formats + # each sub-request as one Speaker-1 line (the server maps the + # lowest speaker to the cloned reference voice). + client = self._make_client(preset_mode=True, voice="narrator", + family="vibevoice") + with patch("converter.clients.faster.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello world.\n\nSecond paragraph here.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["input"], + "Speaker 1: Hello world. Second paragraph here.") + + def test_other_families_keep_the_raw_text(self): + # The script formatting is the vibevoice profile's alone: every + # other family sends the text untouched. + self.assertIsNone( + AUDIOCPP_FAMILY_PROFILES.get("higgs_audio_tts", + AUDIOCPP_DEFAULT_FAMILY_PROFILE) + .script_prefix) + self.assertEqual(audiocpp_script_input("Speaker 1", "a\nb"), + "Speaker 1: a b") + def test_http_error_body_surfaced(self): import urllib.error client = self._make_client() |