diff options
Diffstat (limited to 'tests/test_tts.py')
| -rw-r--r-- | tests/test_tts.py | 254 |
1 files changed, 249 insertions, 5 deletions
diff --git a/tests/test_tts.py b/tests/test_tts.py index 89248f2..a2df07f 100644 --- a/tests/test_tts.py +++ b/tests/test_tts.py @@ -654,6 +654,144 @@ class AudioCppTTSClientHealthTests(unittest.TestCase): self.assertIn("pocket-tts", message) +class AudioCppTaskDetectionTests(unittest.TestCase): + """Task auto-detection (tts/clon/vdes) and voice design validation.""" + + @staticmethod + def _json_response(payload): + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = json.dumps(payload).encode("utf-8") + return response + + def _client(self, voice=None, instructions=None, request_options=None, + models=None): + if models is None: + models = {"data": [{"id": config.AUDIOCPP_MODEL_ID, + "family": "qwen3_tts"}]} + + 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(models) + if "/v1/audio/voices" in url: + return self._json_response({"voices": ["narrator"]}) + raise AssertionError(f"unexpected URL: {url}") + + with patch("converter.tts.urllib.request.urlopen", + side_effect=_dispatch): + return AudioCppTTSClient(voice=voice, instructions=instructions, + request_options=request_options) + + def test_missing_task_falls_back_to_tts(self): + # Servers that predate the task field hosted plain TTS models. + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts"}]}) + self.assertEqual(client.task, tts.AUDIOCPP_TASK_TTS) + self.assertFalse(client.design_mode) + + def test_task_detected_from_models_endpoint(self): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertEqual(client.task, tts.AUDIOCPP_TASK_VDES) + self.assertTrue(client.design_mode) + + def test_clon_task_entry_connects_in_preset_mode(self): + client = self._client(voice="narrator", models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "chatterbox", + "task": "clon"}]}) + self.assertEqual(client.task, "clon") + self.assertFalse(client.design_mode) + self.assertTrue(client.preset_mode) + + def test_unsupported_task_rejected_with_available_entries(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_asr", + "task": "asr"}, + {"id": "tts-1", "family": "qwen3_tts", "task": "tts"}]}, + instructions="unused") + message = str(ctx.exception) + self.assertIn("'asr'", message) + self.assertIn("--model", message) + self.assertIn("tts-1", message) + + def test_vdes_without_instructions_requires_description(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}) + message = str(ctx.exception) + self.assertIn("voice design", message) + self.assertIn("--instructions", message) + + def test_vdes_with_voice_rejected(self): + with self.assertRaises(RuntimeError) as ctx: + self._client(voice="narrator", models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertIn("--voice", str(ctx.exception)) + self.assertIn("--instructions", str(ctx.exception)) + + def test_vdes_with_instructions_connects_in_design_mode(self): + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="A warm adult narrator") + self.assertTrue(client.design_mode) + self.assertEqual(client.instructions, "A warm adult narrator") + out = buf.getvalue() + self.assertIn("voice design", out) + self.assertIn("A warm adult narrator", out) + + def test_instructions_without_voice_on_generic_family_connects(self): + # Families without built-in speakers can get their voice from the + # instruction alone (e.g. OmniVoice voice design). + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "omnivoice", + "task": "tts"}]}, + instructions="female, young adult, moderate pitch") + self.assertFalse(client.design_mode) + self.assertTrue(client.instruction_voice) + self.assertIn("instruction voice", buf.getvalue()) + + def test_instructions_with_builtin_speaker_family_stays_speaker_mode(self): + buf = io.StringIO() + with redirect_stdout(buf): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "tts"}]}, + instructions="Very happy.") + self.assertFalse(client.design_mode) + self.assertFalse(client.instruction_voice) + self.assertIn("speaker 'Vivian'", buf.getvalue()) + + def test_config_instructions_used_when_flag_omitted(self): + with patch.object(config, "AUDIOCPP_INSTRUCTIONS", + "A calm elderly storyteller"): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}) + self.assertEqual(client.instructions, "A calm elderly storyteller") + + def test_explicit_instructions_override_config_default(self): + with patch.object(config, "AUDIOCPP_INSTRUCTIONS", "from config"): + client = self._client(models={"data": [ + {"id": config.AUDIOCPP_MODEL_ID, "family": "qwen3_tts", + "task": "vdes"}]}, + instructions="from flag") + self.assertEqual(client.instructions, "from flag") + + class AudioCppFamilyDetectionTests(unittest.TestCase): """Family auto-detection and per-family adaptations.""" @@ -768,7 +906,8 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): @staticmethod def _make_client(preset_mode=False, voice="Vivian", language="English", seed=-1, - chunk_text=True, family="qwen3_tts"): + chunk_text=True, family="qwen3_tts", task="tts", + instructions=None, request_options=None): client = AudioCppTTSClient.__new__(AudioCppTTSClient) client.api_url = "http://127.0.0.1:8080" client.model_id = config.AUDIOCPP_MODEL_ID @@ -778,8 +917,18 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): client._seed = seed client.chunk_text = chunk_text client.family = family + client.task = task client.profile = tts.AUDIOCPP_FAMILY_PROFILES.get( family, tts.AUDIOCPP_DEFAULT_FAMILY_PROFILE) + client.instructions = instructions or "" + client.request_options = dict(request_options or {}) + client.design_mode = task == tts.AUDIOCPP_TASK_VDES + # Mirrors the connect-time rule: an instruction-defined voice on a + # family without built-in speakers (design mode takes precedence). + client.instruction_voice = ( + not preset_mode and not client.design_mode + and not client.profile.builtin_speakers + and bool(client.instructions)) return client @staticmethod @@ -862,6 +1011,81 @@ class AudioCppTTSClientRequestTests(unittest.TestCase): payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) self.assertEqual(payload["instructions"], config.INSTRUCT) + def test_explicit_instructions_replace_config_instruct(self): + # --instructions overrides the INSTRUCT default in speaker mode. + client = self._make_client(preset_mode=False, + instructions="Read whisper quiet.") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["instructions"], "Read whisper quiet.") + + def test_preset_mode_sends_instructions_alongside_voice(self): + # Clone + style control: both the server-side voice and the + # instruction reach the model. + client = self._make_client(preset_mode=True, voice="narrator", + instructions="Calm and steady.") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["voice"], "narrator") + self.assertEqual(payload["instructions"], "Calm and steady.") + + def test_design_mode_payload_omits_voice_and_sends_instructions(self): + client = self._make_client(task="vdes", + instructions="A warm adult narrator") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("voice", payload) + self.assertEqual(payload["instructions"], "A warm adult narrator") + + def test_design_mode_language_follows_family_profile(self): + # The VoiceDesign package is family qwen3_tts, whose language field + # takes Qwen display names like the other variants. + client = self._make_client(task="vdes", language="Japanese", + instructions="A warm adult narrator") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["language"], "Japanese") + + def test_instruction_voice_payload_omits_voice(self): + # Instruction-defined voice on a family without built-in speakers: + # no speaker name is invented, the instruction carries the voice. + client = self._make_client(family="omnivoice", + instructions="female, young adult") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("voice", payload) + self.assertNotIn("language", payload) # generic profile: omitted + self.assertEqual(payload["instructions"], "female, young adult") + + def test_request_options_forwarded_in_payload(self): + client = self._make_client(preset_mode=True, voice="narrator", + request_options={"emotion": "neutral", + "speed": "1.1"}) + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertEqual(payload["options"], {"emotion": "neutral", + "speed": "1.1"}) + + def test_empty_request_options_omit_options_field(self): + client = self._make_client(preset_mode=True, voice="narrator") + with patch("converter.tts.urllib.request.urlopen", + return_value=self._post_response(self._wav_bytes())) as mock_urlopen: + client._request_wav("Hello.") + payload = json.loads(mock_urlopen.call_args[0][0].data.decode("utf-8")) + self.assertNotIn("options", payload) + def test_generic_family_omits_language_and_instructions(self): # Clone-only families (higgs_audio_tts, voxcpm2, ...) detect the # language themselves and take no style instruction. @@ -1112,7 +1336,9 @@ class BackendWiringTests(unittest.TestCase): backend=tts.BACKEND_AUDIOCPP, voice="narrator", language="ja") mock_audiocpp.assert_called_once_with(voice="narrator", language="Japanese", - chunk_text=False, model_id=None) + chunk_text=False, model_id=None, + instructions=None, + request_options={}) mock_faster.assert_not_called() mock_qwen.assert_not_called() @@ -1121,7 +1347,9 @@ class BackendWiringTests(unittest.TestCase): AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, backend=tts.BACKEND_AUDIOCPP) mock_audiocpp.assert_called_once_with(voice=None, language=config.LANGUAGE, - chunk_text=False, model_id=None) + chunk_text=False, model_id=None, + instructions=None, + request_options={}) def test_audiocpp_backend_chunk_flag_forces_client_chunking(self): with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: @@ -1130,7 +1358,9 @@ class BackendWiringTests(unittest.TestCase): voice="narrator", chunk=True) mock_audiocpp.assert_called_once_with(voice="narrator", language=config.LANGUAGE, - chunk_text=True, model_id=None) + chunk_text=True, model_id=None, + instructions=None, + request_options={}) self.assertTrue(converter.client_chunks) def test_audiocpp_backend_model_id_is_wired_through(self): @@ -1140,7 +1370,21 @@ class BackendWiringTests(unittest.TestCase): model_id="higgs") mock_audiocpp.assert_called_once_with( voice="narrator", language=config.LANGUAGE, - chunk_text=False, model_id="higgs") + chunk_text=False, model_id="higgs", instructions=None, + request_options={}) + + def test_audiocpp_backend_instructions_and_options_are_wired_through(self): + with patch("converter.converter.AudioCppTTSClient") as mock_audiocpp: + AudiobookConverter(voice_mode=tts.VOICE_MODE_CUSTOM, + backend=tts.BACKEND_AUDIOCPP, + instructions="A warm adult narrator", + request_options={"emotion": "neutral", + "speed": "1.1"}) + mock_audiocpp.assert_called_once_with( + voice=None, language=config.LANGUAGE, + chunk_text=False, model_id=None, + instructions="A warm adult narrator", + request_options={"emotion": "neutral", "speed": "1.1"}) def test_qwen_backend_uses_qwen_client(self): with patch("converter.converter.FasterTTSClient") as mock_faster, \ |