"""A voice belongs to a model, and a model is added with its voices. Disposable SQLite. The rules worth pinning: an Orpheus id is routed through the LiteLLM gateway (it used to fall through to the OpenAI path for not starting with "local-"), a bare model id becomes one row per voice, a voice from another family is refused with the list that would work, and learners are offered every active voice — not only the locally served ones. """ import unittest from unittest.mock import Mock, patch import test_quiz_builder as fixtures # noqa: F401 (sets DATABASE_URL before app imports) from fastapi import FastAPI from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.database import Base, get_db from app.models.ai_model_config import AIModelConfig from app.models.user import User from app.routers import admin, tts from app.services import ai_service, tts_voices from app.utils.auth import get_current_user, require_admin class VoiceTableTests(unittest.TestCase): def test_families_and_their_voices(self): self.assertEqual(tts_voices.family("groq-orpheus-english:hannah"), "orpheus-english") self.assertEqual(tts_voices.family("canopylabs/orpheus-arabic-saudi"), "orpheus-arabic") self.assertEqual(tts_voices.family("openrouter-fish-s2.1-pro-tts"), "fish") self.assertEqual(tts_voices.family("local-kokoro-tts:am_adam"), "kokoro") self.assertIsNone(tts_voices.family("tts-1:alloy")) self.assertEqual([v for v, _ in tts_voices.voices_for("groq-orpheus-english")], ["autumn", "diana", "hannah", "austin", "daniel", "troy"]) self.assertEqual(tts_voices.voices_for("openrouter-fish-s2.1-pro-tts"), [("alloy", "Fish Alloy")]) self.assertEqual(tts_voices.voices_for("something-new"), []) def test_litellm_routing_and_request_shape(self): self.assertTrue(tts_voices.is_litellm_tts("groq-orpheus-english:daniel")) self.assertTrue(tts_voices.is_litellm_tts("local-anything:x")) self.assertFalse(tts_voices.is_litellm_tts("tts-1:alloy")) self.assertFalse(tts_voices.is_litellm_tts("elevenlabs/rachel")) # Groq refuses mp3 for Orpheus; the gateway's own models take mp3. self.assertEqual(tts_voices.request_options("groq-orpheus-english"), {"response_format": "wav"}) self.assertEqual(tts_voices.media_type("groq-orpheus-english:troy"), "audio/wav") self.assertEqual(tts_voices.request_options("local-kokoro-tts:af_bella"), {"response_format": "mp3"}) self.assertEqual(tts_voices.media_type("local-kokoro-tts"), "audio/mpeg") def test_accepts_refuses_only_another_familys_voice(self): self.assertTrue(tts_voices.accepts("groq-orpheus-english:hannah")) self.assertFalse(tts_voices.accepts("groq-orpheus-english:am_adam")) self.assertFalse(tts_voices.accepts("local-kokoro-tts:hannah")) self.assertTrue(tts_voices.accepts("groq-orpheus-english")) # gets its first voice self.assertTrue(tts_voices.accepts("local-newmodel-tts:whatever")) # no list: nothing to refuse def test_expand_makes_one_row_per_voice(self): rows = tts_voices.expand("groq-orpheus-arabic-saudi") self.assertEqual(len(rows), 6) self.assertEqual(rows[0], ("groq-orpheus-arabic-saudi:abdullah", "Orpheus Abdullah")) # An id that already names a voice, or a model we have no list for, is one row as given. self.assertEqual(tts_voices.expand("local-kokoro-tts:am_adam", "Kokoro Adam"), [("local-kokoro-tts:am_adam", "Kokoro Adam")]) self.assertEqual(tts_voices.expand("local-newmodel-tts"), [("local-newmodel-tts", "local-newmodel-tts")]) class GenerateAudioRoutingTests(unittest.TestCase): def _post(self, model_id): calls = [] def fake_post(url, **kw): calls.append((url, kw.get("json"))) resp = Mock(); resp.content = b"audio"; resp.raise_for_status = Mock() return resp with patch.object(ai_service.settings, "LITELLM_API_BASE", "http://gateway:4000/v1"), \ patch.object(ai_service.settings, "LITELLM_API_KEY", "k"), \ patch.object(ai_service.httpx, "post", side_effect=fake_post): out = ai_service.generate_tts_audio("Stridor.", model_id=model_id) return out, calls def test_orpheus_goes_to_the_gateway_as_wav(self): out, calls = self._post("groq-orpheus-english:hannah") self.assertEqual(out, b"audio") url, body = calls[0] self.assertEqual(url, "http://gateway:4000/v1/audio/speech") self.assertEqual(body, {"model": "groq-orpheus-english", "input": "Stridor.", "voice": "hannah", "response_format": "wav"}) def test_bare_model_gets_its_first_voice(self): _, calls = self._post("groq-orpheus-english") self.assertEqual(calls[0][1]["voice"], "autumn") _, calls = self._post("local-kokoro-tts") self.assertEqual(calls[0][1]["voice"], "am_adam") self.assertEqual(calls[0][1]["response_format"], "mp3") class AdminAddsAModelWithItsVoicesTests(unittest.TestCase): def setUp(self): self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool) Base.metadata.create_all(self.engine) self.db = Session(self.engine) self.admin = User(id=1, name="Admin", email="admin@example.test", hashed_password="x", role="admin") self.db.add(self.admin); self.db.commit() app = FastAPI() app.include_router(admin.router, prefix="/admin") app.include_router(tts.router, prefix="/tts") app.dependency_overrides[get_db] = lambda: self.db app.dependency_overrides[require_admin] = lambda: self.admin app.dependency_overrides[get_current_user] = lambda: self.admin self.client = TestClient(app) def tearDown(self): self.client.close() def _add(self, model_id, **extra): return self.client.post("/admin/models", json={"name": model_id, "model_id": model_id, "task": "tts", **extra}) def _rows(self): return [(m.model_id, m.name, m.is_default) for m in self.db.query(AIModelConfig).filter(AIModelConfig.task == "tts").order_by(AIModelConfig.id).all()] def test_a_bare_speech_model_becomes_its_voices(self): res = self._add("groq-orpheus-english") self.assertEqual(res.status_code, 200, res.text) rows = self._rows() self.assertEqual([r[0] for r in rows], [f"groq-orpheus-english:{v}" for v in ["autumn", "diana", "hannah", "austin", "daniel", "troy"]]) self.assertEqual(rows[2][1], "Orpheus Hannah") # The first voice is default only because the task had none. self.assertEqual([r[2] for r in rows], [True, False, False, False, False, False]) self.assertEqual(res.json()["model_id"], "groq-orpheus-english:autumn") def test_adding_a_second_model_keeps_the_existing_default(self): self._add("local-kokoro-tts:am_adam", is_default=True) self._add("groq-orpheus-english") rows = self._rows() self.assertEqual([r[0] for r in rows if r[2]], ["local-kokoro-tts:am_adam"]) self.assertEqual(len(rows), 7) # Again: nothing new to add. self.assertEqual(self._add("groq-orpheus-english").status_code, 409) def test_a_voice_from_another_family_is_refused_with_the_right_list(self): res = self._add("local-kokoro-tts:hannah") self.assertEqual(res.status_code, 400) self.assertIn("It accepts: am_adam, am_michael", res.json()["detail"]) self.assertEqual(self._rows(), []) def test_learners_are_offered_every_active_voice(self): self._add("local-kokoro-tts:am_adam") self._add("groq-orpheus-english") offered = self.client.get("/tts/voices").json() self.assertEqual(len(offered), 7) self.assertEqual({v["model"] for v in offered}, {"local-kokoro-tts", "groq-orpheus-english"}) self.assertEqual(offered[0], {"id": "local-kokoro-tts:am_adam", "name": "local-kokoro-tts:am_adam", "is_default": True, "model": "local-kokoro-tts"}) def test_speak_answers_with_the_media_type_the_model_returns(self): self._add("groq-orpheus-english") with patch.object(tts.ai_service, "generate_tts_audio", return_value=b"RIFF") as gen, \ patch.object(tts, "check_rate_limit", lambda **kw: None): res = self.client.post("/tts/speak", json={"text": "Stridor.", "voice": "groq-orpheus-english:troy"}) self.assertEqual(res.status_code, 200) self.assertEqual(res.headers["content-type"], "audio/wav") self.assertEqual(gen.call_args.kwargs["model_id"], "groq-orpheus-english:troy") if __name__ == "__main__": unittest.main()