diff options
Diffstat (limited to 'app/tests/test_tts.py')
| -rw-r--r-- | app/tests/test_tts.py | 180 |
1 files changed, 177 insertions, 3 deletions
diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index 3538d8b..9067443 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -21,6 +21,9 @@ from converter.clients import ( AUDIOCPP_TASK_VDES, AUDIOCPP_VOICE_CLONE, AUDIOCPP_VOICE_DESIGN, + AUDIOCPP_VOICE_NONE, + AUDIOCPP_VOICE_OPTIONAL, + AUDIOCPP_VOICE_REQUIRED, AUDIOCPP_VOICE_SPEAKER, BACKEND_AUDIOCPP, BACKEND_FASTER, @@ -38,10 +41,13 @@ from converter.clients import ( FasterTTSClient, QwenTTSClient, audiocpp_entry_voice_capability, + audiocpp_family_voice_policy, + audiocpp_request_error, normalize_language, transcribe_reference_audio_detailed, whisper_backend_problem, ) +from converter.clients import audiocpp as audiocpp_client from converter.clients.base import NonRetryableTTSError from converter.converter import AudiobookConverter @@ -1028,13 +1034,16 @@ class AudioCppFamilyDetectionTests(unittest.TestCase): self.assertEqual(client.profile.language_style, AUDIOCPP_LANG_OMIT) 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 + # fast instead of guessing. client = None try: client = self._client(voice=None, models={"data": [ - {"id": _AUDIOCPP_MODEL_ID, "family": "voxcpm2"}]}) + {"id": _AUDIOCPP_MODEL_ID, "family": "some_new_family"}]}) except RuntimeError as exc: message = str(exc) - self.assertIn("voxcpm2", message) + self.assertIn("some_new_family", message) self.assertIn("--voice", message) self.assertIn("no built-in speakers", message) self.assertIsNone(client) @@ -1116,10 +1125,174 @@ class AudiocppEntryVoiceCapabilityTests(unittest.TestCase): # The "customvoice" substring only marks a speaker for the qwen3_tts # family; another family with a lookalike id stays clone-only. self.assertEqual(self._cap("future_tts", "tts", - "Qwen3-TTS-CustomVoice"), + "Qwen3-TTS-CustomVoice"), AUDIOCPP_VOICE_CLONE) +class AudioCppFamilyVoicePolicyTests(unittest.TestCase): + """The per-family voice policy resolver (required/optional/none).""" + + def setUp(self): + # Seed the spec cache instead of reading the (gitignored, setup- + # downloaded) checkout's model_specs, so the tests are hermetic. + cache = audiocpp_client._FAMILY_SPEC_TASKS + cache.clear() + cache.update({ + "higgs_audio_tts": {"tts", "clone"}, + "supertonic": {"tts"}, + "confucius4_tts": {"clone"}, + }) + self.addCleanup(cache.clear) + + def test_pure_tts_family_needs_no_voice(self): + self.assertEqual(audiocpp_family_voice_policy("supertonic"), + AUDIOCPP_VOICE_NONE) + + def test_mixed_family_has_an_optional_voice(self): + self.assertEqual(audiocpp_family_voice_policy("higgs_audio_tts"), + AUDIOCPP_VOICE_OPTIONAL) + + def test_clone_only_spec_is_required(self): + self.assertEqual(audiocpp_family_voice_policy("confucius4_tts"), + AUDIOCPP_VOICE_REQUIRED) + + def test_chatterbox_is_required_despite_its_spec(self): + # The chatterbox spec wrongly lists "tts": the explicit + # clone-only set wins so the binary's rejection is mirrored. + self.assertEqual(audiocpp_family_voice_policy("chatterbox"), + AUDIOCPP_VOICE_REQUIRED) + + def test_qwen3_tts_is_entry_typed_and_stays_required(self): + # Qwen3-TTS is decided per entry (speaker/clone/design), so the + # family policy never loosens its voice requirement. + self.assertEqual(audiocpp_family_voice_policy("qwen3_tts"), + AUDIOCPP_VOICE_REQUIRED) + + def test_unknown_family_keeps_the_conservative_default(self): + self.assertEqual(audiocpp_family_voice_policy("brand_new_family"), + AUDIOCPP_VOICE_REQUIRED) + + +class AudioCppPlainTtsModeTests(unittest.TestCase): + """Plain-TTS runs: families that synthesize without a reference voice.""" + + @staticmethod + def _json_response(payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + # Minimal WAV: _request_wav only validates the RIFF/WAVE header. + _WAV = b"RIFF\x04\x00\x00\x00WAVE" + + def _client(self, family, task="tts", voice=None, captured=None): + def _dispatch(request, **_kwargs): + url = request if isinstance(request, str) else request.full_url + if url.endswith("/health"): + return self._json_response({"status": "ok"}) + if url.endswith("/v1/models"): + return self._json_response( + {"data": [{"id": "model", "family": family, + "task": task}]}) + if "/v1/audio/voices" in url: + return self._json_response({"voices": ["narrator"]}) + if url.endswith("/v1/audio/speech"): + if captured is not None: + captured.append(json.loads(request.data.decode("utf-8"))) + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = self._WAV + return response + if url.endswith("/unload_all_models"): + return self._json_response({"unloaded": []}) + raise AssertionError(f"unexpected URL: {url}") + + # The patch stays up for the whole test so _request_wav calls land + # on the dispatch too (capturing the speech payload). + patcher = patch("converter.clients.faster.urllib.request.urlopen", + side_effect=_dispatch) + patcher.start() + self.addCleanup(patcher.stop) + return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, + model_id="model") + + def test_pure_tts_family_connects_in_plain_mode(self): + client = self._client("supertonic") + self.assertTrue(client.plain_mode) + self.assertFalse(client.preset_mode) + self.assertFalse(client.design_mode) + + def test_mixed_family_without_voice_runs_plain(self): + client = self._client("higgs_audio_tts") + self.assertTrue(client.plain_mode) + + def test_plain_mode_omits_the_voice_field(self): + captured = [] + client = self._client("supertonic", captured=captured) + client._request_wav("Hello world.") + self.assertNotIn("voice", captured[0]) + self.assertEqual(captured[0]["input"], "Hello world.") + + def test_clone_only_family_without_voice_still_raises(self): + # Unknown families keep the conservative clone-only default. + with self.assertRaises(RuntimeError) as ctx: + self._client("some_new_family") + self.assertIn("--voice", str(ctx.exception)) + + def test_clone_only_family_hosted_as_tts_fails_fast(self): + # A Chatterbox entry hosted with task "tts" fails every request at + # session-creation time: refuse at connect with the re-host hint + # instead of 500ing each chunk. + with self.assertRaises(RuntimeError) as ctx: + self._client("chatterbox") + message = str(ctx.exception) + self.assertIn("chatterbox", message) + self.assertIn('"clon"', message) + self.assertIn("Configure Backends", message) + + def test_clone_only_family_hosted_as_tts_fails_fast_with_voice(self): + with self.assertRaises(RuntimeError) as ctx: + self._client("chatterbox", task="tts", voice="narrator") + self.assertIn('"clon"', str(ctx.exception)) + + def test_clone_only_family_hosted_as_clon_needs_a_voice(self): + with self.assertRaises(RuntimeError) as ctx: + self._client("chatterbox", task="clon") + message = str(ctx.exception) + self.assertIn("--voice", message) + self.assertIn("voice_preset", message) + + +class AudioCppCloneOnlyErrorTests(unittest.TestCase): + """The non-retryable classification of clone-only hosting 500s.""" + + def _error(self, message): + return audiocpp_request_error( + 500, json.dumps({"error": {"message": message}})) + + def test_chatterbox_hosting_error_is_not_retryable(self): + exc = self._error( + "Chatterbox supports VoiceCloning and VoiceConversion") + self.assertIsInstance(exc, NonRetryableTTSError) + self.assertIn("VoiceCloning and VoiceConversion", str(exc)) + self.assertIn('"clon"', str(exc)) + + def test_confucius_hosting_error_is_not_retryable(self): + exc = self._error("Confucius4-TTS supports the VoiceCloning task") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_echo_hosting_error_is_not_retryable(self): + exc = self._error("Echo-TTS only supports offline voice cloning") + self.assertIsInstance(exc, NonRetryableTTSError) + + def test_unrelated_error_stays_retryable(self): + exc = self._error("model busy") + self.assertNotIsInstance(exc, NonRetryableTTSError) + self.assertIn("model busy", str(exc)) + + + class AudioCppTTSClientRequestTests(unittest.TestCase): """The /v1/audio/speech payload and response validation.""" @@ -1150,6 +1323,7 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): client.instructions = instructions or "" client.request_options = dict(request_options or {}) client.design_mode = task == AUDIOCPP_TASK_VDES + client.plain_mode = False # Mirrors the connect-time rule: an instruction-defined voice on a # clone-capable entry with no --voice (design mode takes precedence). capability = audiocpp_entry_voice_capability( |