diff options
Diffstat (limited to 'app/tests/test_tts_sglomni.py')
| -rw-r--r-- | app/tests/test_tts_sglomni.py | 99 |
1 files changed, 99 insertions, 0 deletions
diff --git a/app/tests/test_tts_sglomni.py b/app/tests/test_tts_sglomni.py index 0e3a7de..c4cdfaf 100644 --- a/app/tests/test_tts_sglomni.py +++ b/app/tests/test_tts_sglomni.py @@ -146,6 +146,29 @@ class ConnectHealthTests(unittest.TestCase): self.assertIn("not reachable", str(ctx.exception)) self.assertIn("sgl-omni", str(ctx.exception)) + def test_booting_503_tells_the_user_to_wait(self): + # A booting sgl-omni answers /health with 503 + an "unhealthy" + # body (urlopen surfaces that as an HTTPError before any JSON + # could be inspected) — the message must say wait, not start. + with patch("converter.clients.sglomni.urllib.request.urlopen", + side_effect=_http_error(503, '{"status": "unhealthy"}')): + with self.assertRaises(RuntimeError) as ctx: + SgOmniTTSClient(_DUMMY_CHUNKS, model="higgs_audio_v3_tts") + message = str(ctx.exception) + self.assertIn("not healthy yet", message) + self.assertIn("HTTP 503", message) + self.assertIn("booting", message) + self.assertNotIn("not reachable", message) + + def test_other_health_errors_name_the_code(self): + with patch("converter.clients.sglomni.urllib.request.urlopen", + side_effect=_http_error(404, "<html>nope</html>")): + with self.assertRaises(RuntimeError) as ctx: + SgOmniTTSClient(_DUMMY_CHUNKS, model="higgs_audio_v3_tts") + message = str(ctx.exception) + self.assertIn("HTTP 404", message) + self.assertIn("Is this an sgl-omni server?", message) + def test_booting_server_raises(self): with self.assertRaises(RuntimeError) as ctx: self._connect([{"status": "unhealthy"}]) @@ -192,6 +215,7 @@ class PayloadTests(unittest.TestCase): client.language = "English" client._seed = None client._kv_fit = None + client._ref_audio_cached = None return client def test_speaker_payload_sends_the_preset_name(self): @@ -237,6 +261,20 @@ class PayloadTests(unittest.TestCase): b"abc") self.assertNotIn("ref_text", payload) + def test_reference_audio_is_encoded_once_per_run(self): + # The clip cannot change mid-run: the data URL (or resolved path) + # is computed on the first sub-request and reused verbatim. + from converter.clients.sglomni import _data_url as real_data_url + client = self._make_client("higgs_audio_v3_tts", + ref_audio=str(self.ref)) + client.api_url = "http://10.20.30.40:8100" + with patch("converter.clients.sglomni._data_url", + wraps=real_data_url) as encode: + first = client._request_payload("Hello.") + second = client._request_payload("Hello again.") + self.assertEqual(encode.call_count, 1) + self.assertEqual(first["ref_audio"], second["ref_audio"]) + def test_clone_without_reference_sends_no_reference_fields(self): client = self._make_client("higgs_audio_v3_tts") payload = client._request_payload("Hello.") @@ -423,6 +461,33 @@ class KvAdmissionTests(unittest.TestCase): self.assertEqual(second["max_new_tokens"], 2531) +class ErrorClassificationTests(unittest.TestCase): + """HTTP status → retry decision: every 4xx envelope is deterministic.""" + + def _request_error(self, status, detail): + client = SgOmniTTSClient.__new__(SgOmniTTSClient) + return client._request_error(status, detail) + + def test_every_4xx_envelope_is_non_retryable(self): + # Including types outside the OpenAI-style names: the identical + # request fails identically on every attempt. + exception = self._request_error( + 401, json.dumps({"error": {"message": "bad key", + "type": "AuthenticationError"}})) + self.assertIsInstance(exception, NonRetryableTTSError) + self.assertIn("bad key", str(exception)) + + def test_non_json_4xx_bodies_are_non_retryable(self): + exception = self._request_error(400, "plain text refusal") + self.assertIsInstance(exception, NonRetryableTTSError) + self.assertIn("plain text refusal", str(exception)) + + def test_5xx_stays_retryable(self): + exception = self._request_error(500, "CUDA out of memory") + self.assertNotIsInstance(exception, NonRetryableTTSError) + self.assertIn("CUDA out of memory", str(exception)) + + class GenerateChunkTests(unittest.TestCase): """Chunk generation: WAV output, sub-chunking, bookkeeping.""" @@ -511,6 +576,40 @@ class GenerateChunkTests(unittest.TestCase): self.assertIsNone(client.generate_chunk("Hello.", 1)) self.assertEqual(mock_wav.call_count, 1) + def test_non_retryable_errors_propagate(self): + # Deterministic server errors must reach the retry loop directly + # (which skips its remaining attempts and re-raises with the + # actionable message), not come back as a generic failed attempt. + client = self._make_client() + with patch.object(client, "_request_wav", + side_effect=NonRetryableTTSError( + "unknown voice")): + with self.assertRaises(NonRetryableTTSError): + client.generate_chunk("Hello.", 1) + + def test_retry_loop_skips_remaining_attempts(self): + client = self._make_client() + with patch.object(client, "_request_wav", + side_effect=NonRetryableTTSError( + "unknown voice")) as mock_wav: + with self.assertRaises(NonRetryableTTSError): + client.process_chunk_with_retry(1, "Hello.") + self.assertEqual(mock_wav.call_count, 1) + + def test_a_non_wav_200_body_fails_the_request(self): + # A JSON error body served with HTTP 200 must not be written as + # chunk bytes (it would only fail later, confusingly, in the + # concat step). + client = self._make_client() + response = MagicMock() + response.__enter__.return_value = response + response.read.return_value = b'{"error": {"message": "nope"}}' + with patch("converter.clients.sglomni.urllib.request.urlopen", + return_value=response): + with self.assertRaises(RuntimeError) as ctx: + client._request_wav("Hello.") + self.assertIn("not a WAV file", str(ctx.exception)) + def test_stale_chunk_files_are_removed(self): stale = Path(self._tmp.name) / "chunk_0001.mp3" stale.write_bytes(b"old") |