diff options
Diffstat (limited to 'app/tests/test_tts.py')
| -rw-r--r-- | app/tests/test_tts.py | 379 |
1 files changed, 370 insertions, 9 deletions
diff --git a/app/tests/test_tts.py b/app/tests/test_tts.py index 6c245e5..45e3812 100644 --- a/app/tests/test_tts.py +++ b/app/tests/test_tts.py @@ -1,7 +1,9 @@ """Tests for the TTS client wrappers (language handling and payloads).""" +import base64 import io import json +import struct import tempfile import time import urllib.error @@ -45,6 +47,9 @@ from converter.clients import ( audiocpp_family_voice_policy, audiocpp_request_error, audiocpp_script_input, + allocation_log_note, + build_trimmed_voice_reference, + nvidia_device_memory_report, normalize_language, transcribe_reference_audio_detailed, whisper_backend_problem, @@ -1192,6 +1197,15 @@ class AudioCppFamilyVoicePolicyTests(unittest.TestCase): self.assertEqual(audiocpp_family_voice_policy("qwen3_tts"), AUDIOCPP_VOICE_REQUIRED) + def test_vevo2_is_required_despite_its_spec(self): + # Vevo2's spec lists tts/vc/svc but no "clone" task, yet its + # zero-shot TTS route refuses every request without a timbre + # reference: the explicit required set mirrors the server, so an + # "All" run sends the picked voice instead of failing every + # request with no voice at all. + self.assertEqual(audiocpp_family_voice_policy("vevo2"), + AUDIOCPP_VOICE_REQUIRED) + def test_unknown_family_keeps_the_conservative_default(self): self.assertEqual(audiocpp_family_voice_policy("brand_new_family"), AUDIOCPP_VOICE_REQUIRED) @@ -1240,7 +1254,8 @@ class AudioCppPlainTtsModeTests(unittest.TestCase): # 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 _client(self, family, task="tts", voice=None, captured=None, + instructions=None): def _dispatch(request, **_kwargs): url = request if isinstance(request, str) else request.full_url if url.endswith("/health"): @@ -1269,7 +1284,8 @@ class AudioCppPlainTtsModeTests(unittest.TestCase): patcher.start() self.addCleanup(patcher.stop) return AudioCppTTSClient(_DUMMY_CHUNKS, voice=voice, - model_id="model") + model_id="model", + instructions=instructions) def test_pure_tts_family_connects_in_plain_mode(self): client = self._client("supertonic") @@ -1317,6 +1333,30 @@ class AudioCppPlainTtsModeTests(unittest.TestCase): self.assertIn("--voice", message) self.assertIn("voice_preset", message) + def test_clone_only_instructions_alone_do_not_define_the_voice(self): + # The REQUIRED-policy refusal precedes the instruction-voice + # branch: clone-only (and Vevo2-style) families cannot take their + # voice from an instruction, so a voice-less run fails fast with + # the --voice fix instead of 500ing every request server-side. + with self.assertRaises(RuntimeError) as ctx: + self._client("chatterbox", task="clon", + instructions="Calm and steady.") + message = str(ctx.exception) + self.assertIn("--voice", message) + self.assertNotIn("instruction", message) + + def test_vevo2_without_voice_refuses_at_connect(self): + with self.assertRaises(RuntimeError) as ctx: + self._client("vevo2") + message = str(ctx.exception) + self.assertIn("--voice", message) + self.assertIn("vevo2", message) + + def test_vevo2_with_voice_connects_in_preset_mode(self): + client = self._client("vevo2", voice="narrator") + self.assertTrue(client.preset_mode) + self.assertFalse(client.plain_mode) + class AudioCppCloneOnlyErrorTests(unittest.TestCase): """The non-retryable classification of clone-only hosting 500s.""" @@ -1404,15 +1444,54 @@ class AudioCppDeterministicErrorTests(unittest.TestCase): self.assertIsInstance(exc, NonRetryableTTSError) self.assertIn("trim", str(exc)) - def test_allocation_failures_are_not_retryable(self): + def test_allocation_failures_are_not_retryable_with_a_hint(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) + # The hint names the server log (which records the exact attempted + # allocation size) and the DramaBox mem_saver session option. + for message in ("DramaBox vocoder backend buffer allocation failed", + "failed to allocate MOSS codec encoder forward graph"): + exc = self._error(message) + self.assertIsInstance(exc, NonRetryableTTSError) + self.assertIn("audiocpp-server.log", str(exc)) + self.assertIn("dramabox.mem_saver", str(exc)) + + def test_missing_companion_hint_names_the_configure_fix(self): + exc = self._error( + "model path does not exist: /tmp/audiocpp-gguf/MioCodec-25Hz" + "-44.1kHz-v2") + self.assertIn("companion package", str(exc)) + self.assertIn("Configure Backends", str(exc)) + + def test_stale_package_layout_hint_names_the_re_download(self): + exc = self._error("missing model package file 'tokenizer_merges'") + self.assertIn("Configure Backends", str(exc)) + self.assertIn("re-downloaded", str(exc)) + + def test_multi_gguf_directory_hint_names_the_hosting_fix(self): + exc = self._error("model directory contains 4 GGUF files: /m") + self.assertIn("several GGUFs", str(exc)) + self.assertIn("Configure Backends", str(exc)) + + def test_sample_capacity_hint_names_the_capacity_override(self): + exc = self._error("VoxCPM2 AudioVAE encoder sample capacity exceeded") + self.assertIn("encoder-sample capacity", str(exc)) + self.assertIn("Configure Backends", str(exc)) + + def test_allocation_log_note_is_appended_to_the_error(self): + exc = audiocpp_request_error( + 500, json.dumps({"error": {"message": + "DramaBox audio VAE backend buffer allocation failed"}}), + log_note=" The server's log (/x) records the failed allocation " + "as: allocating 12.5 MiB on device 0") + self.assertIn("allocating 12.5 MiB on device 0", str(exc)) + + def test_log_note_is_not_appended_to_unrelated_errors(self): + exc = audiocpp_request_error( + 500, json.dumps({"error": {"message": "model busy"}}), + log_note=" The server's log (/x) records the failed allocation " + "as: allocating 12.5 MiB on device 0") + self.assertNotIn("allocating 12.5 MiB", str(exc)) def test_max_tokens_before_eoc_stays_retryable(self): # Proven transient: a request that hit it has succeeded on retry. @@ -2467,3 +2546,285 @@ class BackendWiringTests(unittest.TestCase): 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.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.log = Path(self._tmp.name) / "audiocpp-server.log" + + def tearDown(self): + self._tmp.cleanup() + + def _note(self, message): + with patch("backends.servers.server_log_path", + return_value=self.log): + return allocation_log_note(message) + + def test_cuda_malloc_failure_line_is_surfaced(self): + self.log.write_text( + "I ... engine loaded\n" + "ggml_backend_cuda_buffer_type_alloc_buffer: allocating " + "1240.5 MiB on device 0: cudaMalloc failed: out of memory\n" + "server: request failed\n", encoding="utf-8") + note = self._note("DramaBox audio VAE backend buffer allocation " + "failed") + self.assertIn("1240.5 MiB", note) + self.assertIn("device 0", note) + self.assertIn(str(self.log), note) + + def test_non_allocation_message_gets_no_note(self): + self.log.write_text("allocating 1.0 MiB on device 0: cudaMalloc " + "failed: out of memory\n", encoding="utf-8") + self.assertEqual(self._note("model busy"), "") + + def test_missing_log_yields_no_note(self): + self.assertEqual( + self._note("failed to allocate MOSS codec encoder forward " + "graph"), "") + + def test_log_without_allocation_lines_yields_no_note(self): + self.log.write_text("unrelated\n", encoding="utf-8") + self.assertEqual(self._note("MOSS codec encoder forward graph " + "allocation failed"), "") + + +class DeviceMemoryWarningTests(unittest.TestCase): + """The one-time low-free-VRAM warning before the first request.""" + + def _warn(self, report, url="http://127.0.0.1:8080"): + client = AudioCppTTSClient.__new__(AudioCppTTSClient) + client.api_url = url + client.quiet = False + buf = io.StringIO() + with redirect_stdout(buf), \ + patch.object(audiocpp_client, "nvidia_device_memory_report", + return_value=report): + client._warn_low_device_memory() + return buf.getvalue() + + def test_low_free_memory_warns(self): + out = self._warn("0, 24576, 1024\n1, 24576, 23000") + self.assertIn("GPU 0", out) + self.assertIn("1024 MiB free of 24576 MiB", out) + self.assertNotIn("GPU 1", out) + + def test_healthy_memory_warns_nothing(self): + self.assertEqual(self._warn("0, 24576, 23000"), "") + + def test_missing_nvidia_smi_warns_nothing(self): + self.assertEqual(self._warn(None), "") + + def test_remote_host_skips_the_check(self): + with patch.object(audiocpp_client, "nvidia_device_memory_report", + side_effect=AssertionError("should not run")): + self.assertEqual(self._warn("0, 24576, 1024", + 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", {})) |