"""Custom backend selection keeps endpoint, model, and credentials together.""" import asyncio import os import tempfile import unittest from pathlib import Path from unittest.mock import AsyncMock, Mock, patch from fastapi import HTTPException os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="custom-tts-tests-")) from core.config import _preview_backend_base_url from core import tts_helpers from routes import docker, stt, tts class CustomTtsTests(unittest.TestCase): def settings(self): return { "tts_url": "http://default/v1", "tts_api_key": "default-key", "tts_backend": "openai", "llm_url": "http://gateway/v1", "llm_api_key": "gateway-key", "custom_engine_cards": [ {"id": "first", "name": "Not a model", "role": "tts", "url": "http://gateway/v1", "params": {"speed": 0.8}}, {"id": "chosen", "name": "Also not a model", "role": "tts", "url": "http://gateway/v1", "params": {"speed": 1.2}}, ], "engine_api_keys": {"first": "first-key", "chosen": "chosen-key"}, "engine_models": {"first": "first-model", "chosen": "real-model"}, } def test_custom_preview_posts_selected_model_key_and_params(self): settings = self.settings() response = Mock(content=b"audio", headers={"content-type": "audio/wav"}, status_code=200) with patch("core.config._load_settings", return_value=settings), \ patch("core.config.request_with_retries", return_value=response) as post: result = tts_helpers._preview_request_audio("hello", "alloy", settings, backend="custom:chosen") self.assertEqual(result, (b"audio", "audio/wav")) self.assertEqual(post.call_args.args, ("POST", "http://gateway/v1/audio/speech")) payload = post.call_args.kwargs["json"] self.assertEqual((payload["model"], payload["voice"], payload["input"], payload["speed"]), ("real-model", "alloy", "hello", 1.2)) self.assertEqual(post.call_args.kwargs["headers"], {"Authorization": "Bearer chosen-key"}) def test_url_priority_and_model_fallback_never_use_card_name(self): settings = self.settings() settings["engine_local_urls"] = {"dc-chosen": "http://override/v1", "chosen": "http://bare"} self.assertEqual(_preview_backend_base_url(settings, "custom:chosen"), "http://override/v1") settings["engine_models"].pop("chosen") endpoint, payload, headers = tts_helpers._tts_request_config("hello", "voice", settings, backend_override="custom:chosen") self.assertEqual(endpoint, "http://override/v1/audio/speech") self.assertEqual(payload["model"], "tts-1") self.assertEqual(headers, {"Authorization": "Bearer chosen-key"}) self.assertEqual(_preview_backend_base_url(settings, "custom:unknown"), "") def test_keyless_gateway_probe_activation_and_inference_agree(self): settings = self.settings() settings["engine_api_keys"].pop("chosen") for name in ("chosen",): self.assertEqual(docker._engine_api_key(settings, name, "tts", url="http://gateway/"), "gateway-key") self.assertEqual(docker._engine_api_key(settings, name, "tts", url="http://unrelated"), "sk-dummy-key") self.assertEqual(docker._engine_api_key({"llm_url": "http://gateway/v1", "llm_api_key": "gateway-key"}, "", "tts", url="http://gateway"), "gateway-key") with patch("routes.docker._load_settings", return_value=settings), patch("routes.docker._save_settings") as save: request = Mock(json=AsyncMock(return_value={"url": "http://gateway/v1", "role": "tts", "setting_key": "tts_url", "model": "real-model"})) asyncio.run(docker.use_as_engine("chosen", request, "")) activated = save.call_args.args[0] _, _, headers = tts_helpers._tts_request_config("hello", "voice", activated, backend_override="custom:chosen") self.assertEqual(headers, {"Authorization": "Bearer gateway-key"}) self.assertEqual(docker._engine_api_key(settings, "chosen", "tts", "typed", "http://gateway"), "typed") def test_unswitched_cards_list_with_card_url_and_skip_invalid_urls(self): settings = self.settings() settings["custom_engine_cards"][1]["label"] = "My Speech Engine" settings["custom_engine_cards"].append({"id": "bad", "role": "tts", "url": ""}) with patch("routes.tts._load_settings", return_value=settings), \ patch("routes.tts._fetch_backend_voices", return_value=["alloy"]), \ patch("routes.tts._backend_health_detail", return_value={"ok": True}): result = asyncio.run(tts.tts_backends()) custom = {item["id"]: item for item in result["backends"] if item["id"].startswith("custom:")} self.assertEqual(set(custom), {"custom:first", "custom:chosen"}) self.assertEqual(custom["custom:chosen"]["url"], "http://gateway/v1") self.assertEqual(custom["custom:chosen"]["label"], "real-model (Custom)") self.assertEqual(custom["custom:chosen"]["title"], "My Speech Engine") self.assertEqual(custom["custom:first"]["title"], "Not a model") native = {item["id"]: item for item in result["backends"] if not item["id"].startswith("custom:")} self.assertEqual(native["voice_clone"]["title"], "Qwen3 TTS · Voice Clone") self.assertEqual(native["voice_design"]["title"], "Qwen3 TTS · Voice Design") self.assertEqual(native["customvoice"]["title"], "Qwen3 TTS · Custom Voice") self.assertEqual(native["streaming"]["title"], "Qwen3 TTS · Streaming") self.assertEqual(native["fishspeech"]["title"], "Fish-Speech API") self.assertEqual(settings["engine_models"]["chosen"], "real-model") def test_model_discovery_does_not_request_voice_ids(self): response = Mock(status_code=200, json=Mock(return_value={"data": [{"id": "real-model"}]})) with patch("routes.docker._load_settings", return_value=self.settings()), \ patch("routes.docker.requests.get", return_value=response) as get: result = asyncio.run(docker.engine_models("http://gateway/v1", "tts", "chosen", "")) self.assertEqual(result["models"], ["real-model"]) self.assertEqual(get.call_args.args[0], "http://gateway/v1/models") self.assertEqual(get.call_args.kwargs["headers"]["Authorization"], "Bearer chosen-key") def test_speech_proxy_forwards_custom_backend(self): async def bounded(fn, *args): return fn(*args) async def generate(text, _language, _settings, fn): return await fn(text) request = Mock(json=AsyncMock(return_value={"input": "hello", "voice": "alloy", "app": "test"})) with patch("routes.tts._load_settings", return_value=self.settings()), \ patch("routes.tts._resolve_tts_route", return_value=("alloy", {"backend": "custom:chosen"})), \ patch("core.presets._resolve_virtual_voice", return_value=None), \ patch("routes.tts._resolve_virtual_voice", return_value=None), \ patch("routes.tts._bounded_tts_call", side_effect=bounded), \ patch("routes.tts._generate_with_tags", side_effect=generate), \ patch("routes.tts._preview_request_audio", return_value=(b"audio", "audio/wav")) as preview, \ patch("routes.tts._apply_route_sounds", return_value=(b"audio", "audio/wav", [])), \ patch("routes.tts._prepare_proxy_audio", return_value=(b"audio", "audio/wav", "wav", None, False)), \ patch("routes.tts._routing_log_request"): response = asyncio.run(tts.openai_speech_proxy(request)) self.assertEqual(response.status_code, 200) self.assertEqual(preview.call_args.args[-1], "custom:chosen") def test_custom_stt_params_keep_selected_identity_on_shared_gateway(self): settings = self.settings() for card in settings["custom_engine_cards"]: card["role"] = "stt" card["params"] = {"language": "de" if card["id"] == "first" else "en"} response = Mock(ok=True, status_code=200, text="transcribed") with tempfile.NamedTemporaryFile(suffix=".wav") as audio, \ patch("routes.stt.requests.post", return_value=response) as post: stt._transcribe_audio(Path(audio.name), settings, "custom:chosen") self.assertEqual(post.call_args.kwargs["data"]["language"], "en") self.assertEqual(post.call_args.kwargs["headers"]["Authorization"], "Bearer chosen-key") def test_legacy_card_name_maps_remain_usable_before_migration(self): settings = self.settings() name = settings["custom_engine_cards"][1]["name"] settings["engine_api_keys"] = {name: "legacy-key", "unrelated": "wrong-key"} settings["engine_models"] = {name: "legacy-model"} settings["engine_local_urls"] = {"dc-" + name: "http://legacy/v1"} endpoint, payload, headers = tts_helpers._tts_request_config("hello", "voice", settings, backend_override="custom:chosen") self.assertEqual(endpoint, "http://legacy/v1/audio/speech") self.assertEqual(payload["model"], "legacy-model") self.assertEqual(payload["speed"], 1.2) self.assertEqual(headers, {"Authorization": "Bearer legacy-key"}) for role in ("tts", "stt", "ocr", "llm"): self.assertEqual(docker._engine_api_key(settings, "chosen", role, url="http://legacy/v1"), "legacy-key") self.assertEqual(docker._engine_api_key(settings, "unknown", role, url="http://unknown"), "sk-dummy-key") settings["engine_api_keys"]["chosen"] = "id-key" settings["engine_models"]["chosen"] = "id-model" _, payload, headers = tts_helpers._tts_request_config("hello", "voice", settings, backend_override="custom:chosen") self.assertEqual(payload["model"], "id-model") self.assertEqual(headers, {"Authorization": "Bearer id-key"}) settings["engine_models"]["chosen"] = "" _, payload, _ = tts_helpers._tts_request_config("hello", "voice", settings, backend_override="custom:chosen") self.assertEqual(payload["model"], "tts-1") def test_disconnect_persists_without_erasing_config_and_reconnect_restores(self): settings = self.settings() original_keys = dict(settings["engine_api_keys"]) original_models = dict(settings["engine_models"]) pairs = (("tts-faster-qwen3-tts-voiceclone", "voice_clone"), ("fish-speech-api", "fishspeech"), ("chosen", "custom:chosen")) with patch("routes.docker._load_settings", side_effect=lambda: settings), \ patch("routes.docker._save_settings") as save: for card, backend in pairs: request = Mock(json=AsyncMock(return_value={"connected": False})) self.assertEqual(asyncio.run(docker.engine_connection(card, request)), {"ok": True, "connected": False}) settings = save.call_args.args[0] self.assertEqual(settings["engine_api_keys"], original_keys) self.assertEqual(settings["engine_models"], original_models) self.assertEqual(settings["tts_url"], "http://default/v1") with patch("routes.tts._load_settings", side_effect=lambda: settings), \ patch("routes.tts._fetch_backend_voices", return_value=["alloy"]) as voices, \ patch("routes.tts._backend_health_detail", return_value={"ok": True}) as health: listed = {item["id"]: item for item in asyncio.run(tts.tts_backends())["backends"]} self.assertEqual(health.call_count, 10) for card, backend in pairs: self.assertTrue(listed[backend]["disabled"]) self.assertFalse(listed[backend]["available"]) self.assertNotIn(backend, [call.args[1] for call in voices.call_args_list]) with patch("routes.docker._load_settings", side_effect=lambda: settings), \ patch("routes.docker._save_settings") as save: for card, backend in pairs: asyncio.run(docker.engine_connection(card, Mock(json=AsyncMock(return_value={"connected": True})))) settings = save.call_args.args[0] listed = {item["id"]: item for item in asyncio.run(tts.tts_backends())["backends"]} for card, backend in pairs: self.assertFalse(listed[backend]["disabled"]) self.assertTrue(listed[backend]["available"]) def test_connection_rejects_unknown_identity_and_invalid_boolean(self): with patch("routes.docker._load_settings", return_value=self.settings()), \ patch("routes.docker._save_settings") as save: for body in ([], {}, {"connected": "false"}, {"connected": 1}): with self.assertRaises(HTTPException) as caught: asyncio.run(docker.engine_connection("chosen", Mock(json=AsyncMock(return_value=body)))) self.assertEqual(caught.exception.status_code, 400) with self.assertRaises(HTTPException) as caught: asyncio.run(docker.engine_connection("unknown", Mock(json=AsyncMock(return_value={"connected": False})))) self.assertEqual(caught.exception.status_code, 404) save.assert_not_called() def test_use_as_reconnects_without_changing_other_disabled_cards(self): settings = self.settings() settings["engine_disabled"] = {"chosen": True, "first": True} request = Mock(json=AsyncMock(return_value={"role": "tts", "setting_key": "tts_url", "url": "http://gateway/v1", "model": "real-model"})) with patch("routes.docker._load_settings", return_value=settings), patch("routes.docker._save_settings") as save: asyncio.run(docker.use_as_engine("chosen", request, "")) self.assertEqual(save.call_args.args[0]["engine_disabled"], {"first": True}) if __name__ == "__main__": unittest.main()