111 lines
5.4 KiB
Python
111 lines
5.4 KiB
Python
"""TTS reachability probing must work for OpenAI-style card URLs that end in /v1."""
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="tts-health-tests-"))
|
|
|
|
from routes import tts
|
|
|
|
|
|
class TtsHealthProbeTests(unittest.TestCase):
|
|
def test_v1_card_url_is_probed_from_the_server_root(self):
|
|
with patch("routes.tts.probe_provider", return_value={"ok": True}) as probe:
|
|
tts._backend_health_detail("http://gateway:4000/v1/", "key")
|
|
base, checks = probe.call_args.args[:2]
|
|
self.assertEqual(base, "http://gateway:4000")
|
|
self.assertIn(("/health/liveliness", None), checks)
|
|
self.assertEqual(probe.call_args.kwargs["headers"], {"Authorization": "Bearer key"})
|
|
|
|
def test_plain_host_url_is_unchanged(self):
|
|
with patch("routes.tts.probe_provider", return_value={"ok": False}) as probe:
|
|
tts._backend_health_detail("http://host:8005")
|
|
self.assertEqual(probe.call_args.args[0], "http://host:8005")
|
|
|
|
|
|
|
|
class PublishUnhiddenVoiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from pathlib import Path
|
|
self.dir = Path(tempfile.mkdtemp(prefix="publish-voice-"))
|
|
self.voices = self.dir / "voices.json"
|
|
self.settings = {"voices_scan_dir": str(self.dir), "tts_config_dir": str(self.dir)}
|
|
self.audio = self.dir / "DE_F_New.wav"
|
|
self.audio.write_bytes(b"RIFF")
|
|
|
|
def run_publish(self):
|
|
return tts.publish_voice_clone_if_missing(self.audio, "DE", self.settings)
|
|
|
|
def test_unknown_voice_with_transcript_is_published(self):
|
|
import json
|
|
self.voices.write_text(json.dumps({"other": {"ref_text": "x"}}))
|
|
self.audio.with_suffix(".reference.txt").write_text("Hallo Welt")
|
|
with patch("routes.tts.time.sleep"):
|
|
self.assertTrue(self.run_publish())
|
|
entry = json.loads(self.voices.read_text())["DE_F_New"]
|
|
self.assertEqual((entry["ref_text"], entry["language"]), ("Hallo Welt", "German"))
|
|
|
|
def test_known_voice_keeps_its_cached_embedding(self):
|
|
import json
|
|
self.voices.write_text(json.dumps({"DE_F_New": {"ref_text": "a", "speaker_embeddings": "/x.pt"}}))
|
|
self.audio.with_suffix(".reference.txt").write_text("Hallo")
|
|
self.assertFalse(self.run_publish())
|
|
self.assertIn("speaker_embeddings", json.loads(self.voices.read_text())["DE_F_New"])
|
|
|
|
def test_no_transcript_or_no_voices_file_is_a_no_op(self):
|
|
self.assertFalse(self.run_publish()) # no voices.json
|
|
self.voices.write_text("{}")
|
|
self.assertFalse(self.run_publish()) # no transcript
|
|
self.assertEqual(self.voices.read_text(), "{}")
|
|
|
|
|
|
class GatewayUpstreamProbeTests(unittest.TestCase):
|
|
URL = "http://gateway:4000/v1"
|
|
|
|
def test_running_upstream_is_available(self):
|
|
with patch("routes.tts._gateway_upstream", return_value="http://host:8023/v1"), \
|
|
patch("routes.tts.probe_provider", return_value={"ok": True}) as probe:
|
|
self.assertTrue(tts._backend_health_detail(self.URL, "k", "m")["ok"])
|
|
self.assertEqual(probe.call_count, 1)
|
|
self.assertEqual(probe.call_args.args[0], "http://host:8023")
|
|
|
|
def test_stopped_upstream_is_not_rescued_by_a_live_gateway(self):
|
|
refused = {"ok": False, "error": "Connection refused"}
|
|
with patch("routes.tts._gateway_upstream", return_value="http://host:8020/v1"), \
|
|
patch("routes.tts.probe_provider", return_value=refused) as probe:
|
|
self.assertFalse(tts._backend_health_detail(self.URL, "k", "m")["ok"])
|
|
self.assertEqual(probe.call_count, 1) # the gateway's /health/liveliness is not consulted
|
|
|
|
def test_unroutable_upstream_falls_back_to_the_gateway(self):
|
|
results = [{"ok": False, "error": "Timeout (2 s)"}, {"ok": True, "endpoint": "/health/liveliness"}]
|
|
with patch("routes.tts._gateway_upstream", return_value="http://internal:8020/v1"), \
|
|
patch("routes.tts.probe_provider", side_effect=results) as probe:
|
|
self.assertTrue(tts._backend_health_detail(self.URL, "k", "m")["ok"])
|
|
self.assertEqual(probe.call_args_list[1].args[0], "http://gateway:4000")
|
|
|
|
def test_upstream_lookup_needs_key_and_model_and_reads_model_info(self):
|
|
tts._GATEWAY_UPSTREAMS.clear()
|
|
self.assertEqual(tts._gateway_upstream("http://gw", "", "m"), "")
|
|
info = {"data": [{"model_name": "m", "litellm_params": {"api_base": "http://h:1/v1"}}, {"model_name": "x"}]}
|
|
resp = type("R", (), {"status_code": 200, "json": lambda self: info})()
|
|
with patch("routes.tts.requests.get", return_value=resp) as get:
|
|
self.assertEqual(tts._gateway_upstream("http://gw", "k", "m"), "http://h:1/v1")
|
|
self.assertEqual(tts._gateway_upstream("http://gw", "k", "other"), "")
|
|
self.assertEqual(get.call_count, 1) # second lookup served from the cache
|
|
|
|
|
|
class CustomAvailabilityTests(unittest.TestCase):
|
|
def test_custom_engine_availability_follows_health_not_the_library_voice_list(self):
|
|
library = ["EN_F_a", "DE_M_b"]
|
|
self.assertFalse(tts._backend_available("custom:abc", library, False))
|
|
self.assertTrue(tts._backend_available("custom:abc", [], True))
|
|
|
|
def test_builtin_backends_keep_their_rules(self):
|
|
self.assertTrue(tts._backend_available("voice_clone", ["x"], False))
|
|
self.assertFalse(tts._backend_available("kokoro", ["x"], False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|