diff options
Diffstat (limited to 'app/tests/test_tts.py')
| -rw-r--r-- | app/tests/test_tts.py | 279 |
1 files changed, 66 insertions, 213 deletions
diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index 45e3812..39b407d 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -41,14 +41,15 @@ from converter.clients import ( VOICE_MODES, AudioCppTTSClient, FasterTTSClient, + QWEN3_TTS_SPEAKERS, QwenTTSClient, audiocpp_entry_voice_capability, audiocpp_family_narrates, audiocpp_family_voice_policy, audiocpp_request_error, audiocpp_script_input, + audiocpp_voice_for_run, allocation_log_note, - build_trimmed_voice_reference, nvidia_device_memory_report, normalize_language, transcribe_reference_audio_detailed, @@ -2310,7 +2311,7 @@ class BackendWiringTests(unittest.TestCase): backend=BACKEND_FASTER, voice="narrator") mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", api_url=None, - quiet=False) + quiet=False, cancel=None) mock_qwen.assert_not_called() mock_audiocpp.assert_not_called() @@ -2327,7 +2328,7 @@ class BackendWiringTests(unittest.TestCase): instructions=None, request_options={}, api_url=None, quiet=False, - unload_models=None) + unload_models=None, cancel=None) mock_faster.assert_not_called() mock_qwen.assert_not_called() @@ -2341,7 +2342,7 @@ class BackendWiringTests(unittest.TestCase): instructions=None, request_options={}, api_url=None, quiet=False, - unload_models=None) + unload_models=None, cancel=None) def test_audiocpp_backend_model_id_is_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -2353,7 +2354,7 @@ class BackendWiringTests(unittest.TestCase): voice="narrator", language=config.LANGUAGE, model_id="higgs", instructions=None, request_options={}, api_url=None, quiet=False, - unload_models=None) + unload_models=None, cancel=None) def test_audiocpp_backend_instructions_and_options_are_wired_through(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -2368,7 +2369,8 @@ class BackendWiringTests(unittest.TestCase): model_id=None, instructions="A warm adult narrator", request_options={"emotion": "neutral", "speed": "1.1"}, - api_url=None, quiet=False, unload_models=None) + api_url=None, quiet=False, unload_models=None, + cancel=None) def test_qwen_backend_uses_qwen_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ @@ -2420,7 +2422,7 @@ class BackendWiringTests(unittest.TestCase): voice="narrator", language=config.LANGUAGE, model_id=None, instructions=None, request_options={}, api_url="http://10.0.0.5:8080", quiet=False, - unload_models=None) + unload_models=None, cancel=None) with patch("converter.converter.FasterTTSClient") as mock_faster: AudiobookConverter(voice_mode=VOICE_MODE_CLONE, backend=BACKEND_FASTER, voice="narrator", @@ -2428,7 +2430,7 @@ class BackendWiringTests(unittest.TestCase): mock_faster.assert_called_once_with(chunks_dir=converter_mod.CHUNKS_FOLDER, voice="narrator", api_url="http://10.0.0.5:8000", - quiet=False) + quiet=False, cancel=None) with patch("converter.converter.QwenTTSClient") as mock_qwen: AudiobookConverter(voice_mode=VOICE_MODE_CUSTOM, backend=BACKEND_QWEN, voice="Vivian", @@ -2438,7 +2440,8 @@ class BackendWiringTests(unittest.TestCase): voice_mode=VOICE_MODE_CUSTOM, voice_clone_ref_audio=None, voice_clone_ref_text=None, skip_transcription=False, language=config.LANGUAGE, instructions=None, - api_url="http://10.0.0.5:7860", quiet=False, voice="Vivian") + api_url="http://10.0.0.5:7860", quiet=False, voice="Vivian", + cancel=None) def test_audiocpp_clone_mode_does_not_require_reference(self): # Cloning is server-side for the audiocpp backend, so the @@ -2548,77 +2551,6 @@ if __name__ == "__main__": unittest.main() -class TrimmedVoiceReferenceTests(unittest.TestCase): - """build_trimmed_voice_reference: a bounded inline cloning reference.""" - - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.dir = Path(self._tmp.name) - - def tearDown(self): - self._tmp.cleanup() - - @staticmethod - def _wav_bytes(rate, channels, sampwidth, frames, fill): - if sampwidth == 1: - payload = bytes(fill & 0xFF for _ in range(frames * channels)) - else: - payload = fill.to_bytes(sampwidth, "little", signed=True) \ - * frames * channels - return (b"RIFF" + struct.pack("<I", 36 + len(payload)) + b"WAVEfmt " - + struct.pack("<IHHIIHH", 16, 1, channels, rate, - rate * channels * sampwidth, - channels * sampwidth, sampwidth * 8) - + b"data" + struct.pack("<I", len(payload)) + payload) - - def _write(self, name, rate, channels, sampwidth, seconds, fill=256): - path = self.dir / name - path.write_bytes(self._wav_bytes(rate, channels, sampwidth, - int(rate * seconds), fill)) - return path - - def test_sixty_second_stereo_reference_is_cut_to_thirty_seconds(self): - path = self._write("ref.wav", 44100, 2, 2, 60) - b64, seconds, name = build_trimmed_voice_reference(path) - self.assertEqual(name, "ref.wav") - self.assertLessEqual(seconds, 30.0 + 1e-6) - decoded = base64.b64decode(b64) - with wave.open(io.BytesIO(decoded)) as handle: - self.assertEqual(handle.getframerate(), 44100) - self.assertEqual(handle.getnchannels(), 1) - self.assertEqual(handle.getsampwidth(), 2) - self.assertLessEqual(handle.getnframes() / handle.getframerate(), - 30.0 + 1e-6) - self.assertLessEqual(len(b64), (5 * 1024 * 1024 + 2) // 3 * 4) - - def test_short_reference_is_sent_whole(self): - path = self._write("short.wav", 16000, 1, 2, 8) - _b64, seconds, _name = build_trimmed_voice_reference(path) - self.assertAlmostEqual(seconds, 8.0, places=2) - - def test_float_and_garbage_files_yield_none(self): - self.assertIsNone(build_trimmed_voice_reference(None)) - garbage = self.dir / "garbage.wav" - garbage.write_bytes(b"ID3 not a wav") - self.assertIsNone(build_trimmed_voice_reference(garbage)) - missing = self.dir / "missing.wav" - self.assertIsNone(build_trimmed_voice_reference(missing)) - - def test_sample_widths_beyond_s16_are_downconverted(self): - for name, width, seconds in (("s24.wav", 3, 40), ("s32.wav", 4, 40), - ("u8.wav", 1, 12)): - path = self._write(name, 44100, 1, width, seconds) - result = build_trimmed_voice_reference(path) - self.assertIsNotNone(result, name) - _b64, used, _name = result - self.assertLessEqual(used, min(seconds, 30.0) + 1e-6) - - def test_base64_payload_stays_under_the_server_limit(self): - path = self._write("long.wav", 48000, 2, 2, 120) - b64, _seconds, _name = build_trimmed_voice_reference(path) - self.assertLessEqual(len(base64.b64decode(b64)), 5 * 1024 * 1024) - - class AllocationLogNoteTests(unittest.TestCase): """allocation_log_note: the server log's exact allocation numbers.""" @@ -2695,136 +2627,57 @@ class DeviceMemoryWarningTests(unittest.TestCase): url="http://10.0.0.5:8080"), "") -class TrimmedReferenceRetryTests(unittest.TestCase): - """One trimmed-reference retry after an allocation-failure 500.""" - - def setUp(self): - self._tmp = tempfile.TemporaryDirectory() - self.dir = Path(self._tmp.name) - frames = 44100 * 60 - payload = (3000).to_bytes(2, "little", signed=True) * frames * 2 - wav = (b"RIFF" + struct.pack("<I", 36 + len(payload)) + b"WAVEfmt " - + struct.pack("<IHHIIHH", 16, 1, 2, 44100, 44100 * 2, 2, 16) - + b"data" + struct.pack("<I", len(payload)) + payload) - (self.dir / "obama.wav").write_bytes(wav) - - def tearDown(self): - self._tmp.cleanup() - - _WAV = b"RIFF\x04\x00\x00\x00WAVE" - - def _client(self, responses, captured, voice="obama", - family="moss_tts_local"): - client = AudioCppTTSClient.__new__(AudioCppTTSClient) - client.chunks_dir = self.dir - client.api_url = "http://127.0.0.1:8080" - client.model_id = "MOSS-TTS-Local-v1.5-GGUF" - client.preset_mode = True - client.voice = voice - client.language = "Auto" - client._seed = 42 - client.family = family - client.task = "tts" - client.profile = audiocpp_client.AUDIOCPP_DEFAULT_FAMILY_PROFILE - client.instructions = "" - client.request_options = {} - client.design_mode = False - client.instruction_voice = False - client.plain_mode = False - client._reference_trim_attempted = False - client._voice_ref_b64 = None - client._voice_ref_reference_text = None - client._voice_wav_path = lambda: self.dir / "obama.wav" - - calls = {"n": 0} - - def urlopen(request, **_kwargs): - index = calls["n"] - calls["n"] += 1 - if captured is not None: - captured.append(json.loads(request.data.decode("utf-8"))) - outcome = responses[index] if index < len(responses) \ - else responses[-1] - if isinstance(outcome, urllib.error.HTTPError): - raise outcome - response = MagicMock() - response.__enter__.return_value = response - response.read.return_value = outcome - return response - - patcher = patch("converter.clients.faster.urllib.request.urlopen", - side_effect=urlopen) - patcher.start() - self.addCleanup(patcher.stop) - return client - - @staticmethod - def _alloc_http_error(): - return urllib.error.HTTPError( - "http://127.0.0.1:8080", 500, "Internal Server Error", {}, - io.BytesIO(json.dumps({"error": {"message": - "failed to allocate MOSS codec encoder forward graph"}} - ).encode("utf-8"))) - - def test_allocation_failure_retries_once_with_trimmed_reference(self): - captured = [] - client = self._client([self._alloc_http_error(), self._WAV, self._WAV], - captured) - client._request_wav("Hello.") - client._request_wav("More text.") - self.assertEqual(len(captured), 3) - self.assertIn("voice", captured[0]) - self.assertNotIn("voice_ref", captured[0]) - self.assertIn("voice_ref", captured[1]) - self.assertEqual(captured[1]["voice_ref"]["type"], "base64") - self.assertLessEqual( - len(base64.b64decode(captured[1]["voice_ref"]["data"])), - 5 * 1024 * 1024) - self.assertNotIn("voice", captured[1]) - # The trimmed reference sticks for the rest of the run. - self.assertEqual(captured[2]["voice_ref"]["type"], "base64") - self.assertTrue(client._reference_trim_attempted) - - def test_no_local_wav_falls_through_to_the_error(self): - client = self._client([self._alloc_http_error()], None) - client._voice_wav_path = lambda: None - with self.assertRaises(NonRetryableTTSError): - client._request_wav("Hello.") - self.assertTrue(client._reference_trim_attempted) - - def test_unrelated_errors_are_not_retried(self): - captured = [] - boring = urllib.error.HTTPError( - "http://127.0.0.1:8080", 500, "Internal Server Error", {}, - io.BytesIO(json.dumps({"error": {"message": "model busy"}} - ).encode("utf-8"))) - client = self._client([boring], captured) - with self.assertRaises(RuntimeError): - client._request_wav("Hello.") - self.assertEqual(len(captured), 1) - self.assertIn("voice", captured[0]) - self.assertFalse(client._reference_trim_attempted) - - def test_transcript_is_carried_when_the_spec_accepts_it(self): - captured = [] - client = self._client([self._alloc_http_error(), self._WAV], captured) - with patch.object(audiocpp_client, "_family_spec", - return_value={"options": {"request": [ - {"name": "reference_text"}]}}), \ - patch.object(AudioCppTTSClient, "_voice_transcript", - return_value="The spoken reference text."): - client._request_wav("Hello.") - self.assertEqual(captured[1]["options"]["reference_text"], - "The spoken reference text.") - - def test_transcript_is_omitted_for_option_validating_families(self): - captured = [] - client = self._client([self._alloc_http_error(), self._WAV], captured, - family="dramabox") - with patch.object(audiocpp_client, "_family_spec", - return_value={"options": {"request": [ - {"name": "seed"}]}}), \ - patch.object(AudioCppTTSClient, "_voice_transcript", - return_value="The spoken reference text."): - client._request_wav("Hello.") - self.assertNotIn("reference_text", captured[1].get("options", {})) +class ErrorBodyClassificationTests(unittest.TestCase): + """Deterministic-error detection runs on the FULL HTTP error body.""" + + def test_fragment_beyond_200_chars_is_still_classified(self): + # The deterministic fragment sits deep in a long server message; + # truncating before matching would misclassify it as retryable. + filler = "x" * 300 + body = json.dumps({"error": {"message": + f"{filler} model contract spec not found for family"}}) + error = audiocpp_request_error(500, body) + self.assertIsInstance(error, NonRetryableTTSError) + self.assertIn("not retryable", str(error)) + + def test_quoted_message_is_truncated_for_display(self): + body = json.dumps({"error": {"message": "y" * 1000}}) + error = audiocpp_request_error(500, body) + self.assertNotIn("y" * 300, str(error)) + self.assertLess(len(str(error)), 1000) + + def test_http_error_body_helper_reads_whole_body(self): + body = b"z" * 500 + exc = urllib.error.HTTPError("http://x", 500, "ISE", {}, + io.BytesIO(body)) + self.assertEqual(audiocpp_client._http_error_body(exc), body.decode()) + + +class VoiceForRunTests(unittest.TestCase): + """audiocpp_voice_for_run: the "All"-run per-model voice resolution.""" + + def test_design_takes_no_voice(self): + self.assertIsNone(audiocpp_client.audiocpp_voice_for_run( + "voxcpm2", "vdes", "Vox", "narrator", ["narrator"])) + + def test_speaker_entry_takes_the_speaker_pick_or_first_speaker(self): + model = "Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF" + resolve = audiocpp_client.audiocpp_voice_for_run + self.assertEqual(resolve("qwen3_tts", "tts", model, "Vivian", []), + "Vivian") + self.assertEqual(resolve("qwen3_tts", "tts", model, "narrator", []), + QWEN3_TTS_SPEAKERS[0]) + + def test_clone_entry_takes_the_preset_pick_or_first_server_voice(self): + self.assertEqual(audiocpp_client.audiocpp_voice_for_run( + "higgs_audio_tts", "tts", "higgs", "narrator", + ["narrator", "other"]), "narrator") + self.assertEqual(audiocpp_client.audiocpp_voice_for_run( + "higgs_audio_tts", "tts", "higgs", "unknown", ["first", "x"]), + "first") + self.assertIsNone(audiocpp_client.audiocpp_voice_for_run( + "higgs_audio_tts", "tts", "higgs", "unknown", [])) + + def test_pure_tts_family_takes_no_voice(self): + self.assertIsNone(audiocpp_client.audiocpp_voice_for_run( + "supertonic", "tts", "supertonic", "narrator", ["narrator"])) |