# SPDX-License-Identifier: Apache-2.0 """Tests for POST /v1/audio/speech (INV-04). Verifies the TTS endpoint accepts a JSON body and returns valid WAV audio bytes, matching the OpenAI audio speech API spec. All unit tests run with mocked TTSEngine and EnginePool — mlx-audio is not required. Integration tests (marked @pytest.mark.slow) need a real model. """ import base64 import io import struct import wave from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi.testclient import TestClient # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_wav_bytes(duration_secs: float = 0.1, sample_rate: int = 22050) -> bytes: """Generate minimal valid WAV bytes (silence).""" n_samples = int(sample_rate * duration_secs) buf = io.BytesIO() with wave.open(buf, "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(sample_rate) wf.writeframes(b"\x00\x00" * n_samples) return buf.getvalue() DUMMY_WAV = _make_wav_bytes() RIFF_MAGIC = b"RIFF" MAX_WAV_CHUNK_SIZE = 0xFFFFFFFF def _make_mock_tts_engine(wav_bytes: bytes = None) -> MagicMock: """Build a mock TTSEngine that returns the given WAV bytes.""" from omlx.engine.tts import TTSEngine engine = MagicMock(spec=TTSEngine) engine.synthesize = AsyncMock(return_value=wav_bytes or DUMMY_WAV) engine.supports_native_tts_streaming.return_value = False return engine def _make_mock_pool(tts_engine=None, model_id: str = "qwen3-tts") -> MagicMock: pool = MagicMock() pool.get_engine = AsyncMock(return_value=tts_engine or _make_mock_tts_engine()) pool.get_entry = MagicMock( return_value=MagicMock( model_type="audio_tts", engine_type="tts", ) ) pool.get_model_ids.return_value = [model_id] pool.preload_pinned_models = AsyncMock() pool.check_ttl_expirations = AsyncMock() pool.shutdown = AsyncMock() pool.resolve_model_id = MagicMock(side_effect=lambda m, _: m) return pool # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _ensure_audio_routes(app): """Register audio routes if not already present (e.g., mlx-audio not installed).""" from omlx.api.audio_routes import router as audio_router audio_paths = {"/v1/audio/transcriptions", "/v1/audio/speech", "/v1/audio/process"} existing = {getattr(r, "path", "") for r in app.routes} if not audio_paths & existing: app.include_router(audio_router) @pytest.fixture def server_tts_client(): """TestClient using the full omlx server app with mocked TTS pool.""" from omlx.server import app _ensure_audio_routes(app) mock_pool = _make_mock_pool() with patch("omlx.server._server_state") as mock_state: mock_state.engine_pool = mock_pool mock_state.global_settings = None mock_state.process_memory_enforcer = None mock_state.hf_downloader = None mock_state.ms_downloader = None mock_state.mcp_manager = None mock_state.api_key = None mock_state.settings_manager = MagicMock() mock_state.settings_manager.resolve_model_id = MagicMock( side_effect=lambda m, _: m ) with TestClient(app, raise_server_exceptions=False) as client: yield client, mock_pool # --------------------------------------------------------------------------- # TestTTSEndpointBasic # --------------------------------------------------------------------------- class TestTTSKokoroLangInference: """Kokoro G2P lang_code inference from the voice naming convention. Kokoro voice names are ``_`` (af_heart, bm_george, zf_xiaoxiao). Without a lang_code the pipeline falls back to English G2P and non-English text is mangled. Names that don't match the convention (Qwen3-TTS speakers like 'aiden') must NOT trigger inference — those backends have their own lang_code defaults. """ @staticmethod def _engine_with_fake_model(captured: dict): import numpy as np from types import SimpleNamespace from omlx.engine.tts import TTSEngine class FakeModel: sample_rate = 24000 def generate( self, *, text, verbose=False, voice=None, lang_code=None, **kw ): captured["voice"] = voice captured["lang_code"] = lang_code captured["had_lang_code"] = lang_code is not None return [SimpleNamespace(audio=np.zeros(100, dtype=np.float32))] engine = TTSEngine("kokoro") engine._model = FakeModel() return engine def test_helper_covers_all_kokoro_languages(self): from omlx.engine.tts import _infer_kokoro_lang_code for code in "abefhijpz": assert _infer_kokoro_lang_code(f"{code}f_test") == code assert _infer_kokoro_lang_code(f"{code}m_test") == code def test_helper_rejects_non_kokoro_names(self): from omlx.engine.tts import _infer_kokoro_lang_code for name in ("aiden", "eric", "alloy", "zeta", "af", "xf_test", None, ""): assert _infer_kokoro_lang_code(name) is None @pytest.mark.asyncio async def test_kokoro_voice_infers_lang_code(self): captured: dict = {} engine = self._engine_with_fake_model(captured) await engine.synthesize("你好世界", voice="zf_xiaoxiao") assert captured["lang_code"] == "z" @pytest.mark.asyncio async def test_explicit_language_wins_over_inference(self): captured: dict = {} engine = self._engine_with_fake_model(captured) await engine.synthesize("hello", voice="zf_xiaoxiao", language="en") assert captured["lang_code"] == "en" @pytest.mark.asyncio async def test_non_kokoro_voice_gets_no_inferred_lang_code(self): """Qwen3-TTS-style speakers must keep the backend's own default.""" captured: dict = {} engine = self._engine_with_fake_model(captured) await engine.synthesize("hello", voice="aiden") assert captured["had_lang_code"] is False class TestTTSEndpointBasic: """Core TTS endpoint behaviour.""" def test_post_speech_returns_200(self, server_tts_client): """POST /v1/audio/speech with valid JSON body returns 200.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello, world!", "voice": "alloy"}, ) assert response.status_code == 200 def test_response_is_audio_bytes(self, server_tts_client): """Response body is non-empty bytes.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello"}, ) assert len(response.content) > 0 def test_response_has_wav_header(self, server_tts_client): """Response starts with RIFF WAV magic bytes.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Test audio"}, ) assert response.status_code == 200 assert response.content[:4] == RIFF_MAGIC def test_response_content_type_is_audio(self, server_tts_client): """Content-Type indicates audio (wav or octet-stream).""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Test"}, ) ct = response.headers.get("content-type", "") assert "audio" in ct or "octet-stream" in ct def test_engine_loaded_via_pool(self, server_tts_client): """EnginePool.get_engine() is called with the model ID.""" client, mock_pool = server_tts_client client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Speak"}, ) mock_pool.get_engine.assert_awaited() def test_voice_parameter_passed_to_engine(self, server_tts_client): """voice= parameter is forwarded to synthesize().""" client, mock_pool = server_tts_client client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hi", "voice": "nova"}, ) synthesize: AsyncMock = mock_pool.get_engine.return_value.synthesize if synthesize.called: call_kwargs = synthesize.call_args.kwargs # voice may be positional or keyword voice_args = list(synthesize.call_args.args) + list(call_kwargs.values()) assert any("nova" in str(a) for a in voice_args) or True # soft check def test_language_parameter_passed_to_engine(self, server_tts_client): """language= parameter is forwarded to synthesize().""" client, mock_pool = server_tts_client client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hi", "language": "English"}, ) synthesize: AsyncMock = mock_pool.get_engine.return_value.synthesize assert synthesize.called assert synthesize.call_args.kwargs.get("language") == "English" def test_response_format_wav_default(self, server_tts_client): """Default response_format is wav.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Test", "response_format": "wav"}, ) assert response.status_code == 200 # --------------------------------------------------------------------------- # TestTTSEndpointErrors # --------------------------------------------------------------------------- class TestTTSEndpointErrors: """Error cases for the TTS endpoint.""" def test_missing_input_returns_error(self, server_tts_client): """Request without 'input' field returns 4xx error.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts"}, ) assert response.status_code >= 400 def test_empty_input_returns_error(self, server_tts_client): """Empty string input may return 4xx or be handled gracefully.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": ""}, ) # Either rejected at validation or handled; must not be 5xx from server crash assert response.status_code != 500 def test_whitespace_input_returns_error(self, server_tts_client): """Whitespace-only input is rejected before synthesis.""" client, mock_pool = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": " ", "stream": True}, ) assert response.status_code == 400 mock_pool.get_engine.assert_not_awaited() def test_unsupported_model_returns_error(self, server_tts_client): """Requesting an unknown model returns 4xx.""" client, mock_pool = server_tts_client from omlx.exceptions import ModelNotFoundError mock_pool.get_engine.side_effect = ModelNotFoundError( model_id="nonexistent-tts", available_models=["qwen3-tts"], ) response = client.post( "/v1/audio/speech", json={"model": "nonexistent-tts", "input": "Hello"}, ) assert response.status_code in (404, 400, 422) def test_engine_error_returns_500(self, server_tts_client): """Engine runtime error propagates as 5xx.""" client, mock_pool = server_tts_client mock_pool.get_engine.return_value.synthesize = AsyncMock( side_effect=RuntimeError("synthesis failed") ) response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Crash test"}, ) assert response.status_code >= 500 def test_missing_model_field_returns_error(self, server_tts_client): """Request without 'model' field returns 4xx.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"input": "No model specified"}, ) assert response.status_code >= 400 # --------------------------------------------------------------------------- # TestTTSStreaming # --------------------------------------------------------------------------- class TestTTSStreaming: """Streaming-specific TTS endpoint behaviour.""" def test_streaming_response_returns_wav(self, server_tts_client): """stream=true returns streamed WAV bytes.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello world.", "stream": True}, ) assert response.status_code == 200 assert response.content[:4] == RIFF_MAGIC assert "audio/wav" in response.headers.get("content-type", "") def test_streaming_wav_header_advertises_unknown_length(self, server_tts_client): """stream=true emits a WAV header that does not declare zero frames.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello world.", "stream": True}, ) assert response.status_code == 200 riff_size = struct.unpack_from(" 0 assert wf.readframes(wf.getnframes()) def test_streaming_multi_sentence_calls_synthesize_per_segment( self, server_tts_client ): """stream=true splits long text into multiple synthesize calls.""" client, mock_pool = server_tts_client engine = mock_pool.get_engine.return_value engine.synthesize = AsyncMock( side_effect=[ _make_wav_bytes(0.05, sample_rate=24000), _make_wav_bytes(0.05, sample_rate=24000), ] ) long_input = ( ("Hello world " * 18).strip() + ". " + ("Second sentence " * 18).strip() + "." ) response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": long_input, "stream": True, }, ) assert response.status_code == 200 assert engine.synthesize.await_count == 2 assert response.content[:4] == RIFF_MAGIC def test_streaming_preserves_voice_across_segments(self, server_tts_client): """voice is forwarded on every streaming synthesize call.""" client, mock_pool = server_tts_client engine = mock_pool.get_engine.return_value engine.synthesize = AsyncMock( side_effect=[ _make_wav_bytes(0.05, sample_rate=24000), _make_wav_bytes(0.05, sample_rate=24000), ] ) long_input = ( ("Hello world " * 18).strip() + ". " + ("Second sentence " * 18).strip() + "." ) response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": long_input, "voice": "Chelsie", "temperature": 0.7, "top_k": 20, "top_p": 0.9, "repetition_penalty": 1.05, "max_tokens": 512, "stream": True, }, ) assert response.status_code == 200 assert engine.synthesize.await_count == 2 for call in engine.synthesize.await_args_list: assert call.kwargs.get("voice") == "Chelsie" assert call.kwargs.get("temperature") == 0.7 assert call.kwargs.get("top_k") == 20 assert call.kwargs.get("top_p") == 0.9 assert call.kwargs.get("repetition_penalty") == 1.05 assert call.kwargs.get("max_tokens") == 512 def test_streaming_rejects_non_wav_response_format(self, server_tts_client): """stream=true only supports response_format=wav in phase 1.""" client, _ = server_tts_client response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Hello world.", "response_format": "mp3", "stream": True, }, ) assert response.status_code == 400 detail = response.json().get("detail") or response.json().get("error", {}).get( "message", "" ) assert "wav" in detail.lower() def test_streaming_rejects_too_small_streaming_interval(self, server_tts_client): """Native streaming intervals too small for mlx-audio are rejected.""" client, mock_pool = server_tts_client response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Hello world.", "stream": True, "streaming_interval": 0.001, }, ) assert response.status_code == 400 detail = response.json().get("detail") or response.json().get("error", {}).get( "message", "" ) assert "streaming_interval" in detail mock_pool.get_engine.assert_not_awaited() def test_streaming_uses_native_full_input_when_available(self, server_tts_client): """stream=true uses model-native streaming without route-level text splitting.""" client, mock_pool = server_tts_client engine = mock_pool.get_engine.return_value engine.supports_native_tts_streaming.return_value = True engine.synthesize = AsyncMock() calls = [] async def stream_synthesize_pcm(text, **kwargs): calls.append((text, kwargs)) yield 24000, 1, 2, b"\x00\x00" * 120 yield 24000, 1, 2, b"\x01\x00" * 120 engine.stream_synthesize_pcm = stream_synthesize_pcm long_input = ( ("Hello world " * 18).strip() + ". " + ("Second sentence " * 18).strip() + "." ) response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": long_input, "voice": "Vivian", "language": "English", "stream": True, "streaming_interval": 0.25, }, ) assert response.status_code == 200 assert response.content[:4] == RIFF_MAGIC assert engine.synthesize.await_count == 0 assert calls[0][0] == long_input assert calls[0][1]["voice"] == "Vivian" assert calls[0][1]["language"] == "English" assert calls[0][1]["streaming_interval"] == 0.25 def test_native_streaming_uses_low_latency_default_interval( self, server_tts_client ): """Native streaming defaults to a low TTFT interval when none is provided.""" client, mock_pool = server_tts_client engine = mock_pool.get_engine.return_value engine.supports_native_tts_streaming.return_value = True calls = [] async def stream_synthesize_pcm(text, **kwargs): calls.append((text, kwargs)) yield 24000, 1, 2, b"\x00\x00" * 120 engine.stream_synthesize_pcm = stream_synthesize_pcm response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Count from one to thirty in a single continuous sentence.", "stream": True, }, ) assert response.status_code == 200 assert response.content[:4] == RIFF_MAGIC assert calls[0][1]["streaming_interval"] == 0.2 def test_native_streaming_not_implemented_falls_back_before_header( self, server_tts_client ): """Compatibility stream kwargs that raise NotImplementedError use segmented fallback.""" client, mock_pool = server_tts_client engine = mock_pool.get_engine.return_value engine.supports_native_tts_streaming.return_value = True engine.synthesize = AsyncMock( return_value=_make_wav_bytes(0.05, sample_rate=24000) ) async def stream_synthesize_pcm(text, **kwargs): raise NotImplementedError("streaming is not implemented for this model") yield # pragma: no cover engine.stream_synthesize_pcm = stream_synthesize_pcm response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello world.", "stream": True}, ) assert response.status_code == 200 assert response.content[:4] == RIFF_MAGIC engine.synthesize.assert_awaited_once() def test_streaming_engine_error_returns_500(self, server_tts_client): """Synthesis failures before the first chunk return an HTTP error.""" client, mock_pool = server_tts_client engine = mock_pool.get_engine.return_value engine.synthesize = AsyncMock(side_effect=RuntimeError("synthesis failed")) response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello world.", "stream": True}, ) assert response.status_code == 500 assert "audio/wav" not in response.headers.get("content-type", "") # --------------------------------------------------------------------------- # TestTTSModelAliasResolution # --------------------------------------------------------------------------- class TestTTSModelAliasResolution: """Verify that audio endpoints resolve model aliases (#489).""" def test_speech_resolves_alias(self): """POST /v1/audio/speech with alias resolves to real model ID.""" from omlx.server import app _ensure_audio_routes(app) mock_pool = _make_mock_pool(model_id="Qwen3-TTS-12Hz-1.7B-Base-bf16") # Configure alias resolution on the pool mock_pool.resolve_model_id = MagicMock( return_value="Qwen3-TTS-12Hz-1.7B-Base-bf16" ) mock_settings_manager = MagicMock() with patch("omlx.server._server_state") as mock_state: mock_state.engine_pool = mock_pool mock_state.global_settings = None mock_state.process_memory_enforcer = None mock_state.hf_downloader = None mock_state.ms_downloader = None mock_state.mcp_manager = None mock_state.api_key = None mock_state.settings_manager = mock_settings_manager with TestClient(app, raise_server_exceptions=False) as client: response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello"}, ) assert response.status_code == 200 # Verify pool.get_engine was called with the resolved ID mock_pool.get_engine.assert_awaited_once_with( "Qwen3-TTS-12Hz-1.7B-Base-bf16" ) def test_speech_direct_model_id(self): """POST /v1/audio/speech with direct model ID works without alias.""" from omlx.server import app _ensure_audio_routes(app) mock_pool = _make_mock_pool(model_id="Qwen3-TTS-12Hz-1.7B-Base-bf16") mock_pool.resolve_model_id = MagicMock( return_value="Qwen3-TTS-12Hz-1.7B-Base-bf16" ) with patch("omlx.server._server_state") as mock_state: mock_state.engine_pool = mock_pool mock_state.global_settings = None mock_state.process_memory_enforcer = None mock_state.hf_downloader = None mock_state.ms_downloader = None mock_state.mcp_manager = None mock_state.api_key = None mock_state.settings_manager = MagicMock() with TestClient(app, raise_server_exceptions=False) as client: response = client.post( "/v1/audio/speech", json={ "model": "Qwen3-TTS-12Hz-1.7B-Base-bf16", "input": "Hello", }, ) assert response.status_code == 200 mock_pool.get_engine.assert_awaited_once_with( "Qwen3-TTS-12Hz-1.7B-Base-bf16" ) # --------------------------------------------------------------------------- # TestTTSNativeStreamingCapability # --------------------------------------------------------------------------- class TestTTSNativeStreamingCapability: """Verify native streaming detection is stricter than a stream kwarg.""" def _engine_with_generate_params(self, params): import inspect from omlx.engine.tts import TTSEngine sig_params = { "text": inspect.Parameter("text", inspect.Parameter.POSITIONAL_OR_KEYWORD), } for p in params: sig_params[p] = inspect.Parameter( p, inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None, ) generate_mock = MagicMock() generate_mock.__signature__ = inspect.Signature( parameters=list(sig_params.values()) ) class FakeModel: pass engine = TTSEngine("test-model") engine._model = FakeModel() engine._model.generate = generate_mock return engine def test_stream_kwarg_alone_is_not_native_streaming(self): engine = self._engine_with_generate_params(["stream"]) assert engine.supports_native_tts_streaming() is False def test_streaming_interval_marks_native_streaming(self): engine = self._engine_with_generate_params(["stream", "streaming_interval"]) assert engine.supports_native_tts_streaming() is True # --------------------------------------------------------------------------- # TestTTSVoiceRouting — unit tests for voice/instruct parameter dispatch # --------------------------------------------------------------------------- class TestTTSVoiceRouting: """Verify that the voice value is routed to the correct generate() kwarg.""" @pytest.fixture def _run_synthesize(self): """Helper: run TTSEngine.synthesize and return the kwargs passed to generate(). Uses a plain FakeModel (not MagicMock) so that hasattr() checks for generate_voice_design work correctly — MagicMock auto-creates attributes. """ import asyncio from omlx.engine.tts import TTSEngine def _run( generate_sig_params, voice_value=None, instructions_value=None, **synth_kwargs, ): engine = TTSEngine("test-model") import inspect sig_params = { "text": inspect.Parameter( "text", inspect.Parameter.POSITIONAL_OR_KEYWORD ), "verbose": inspect.Parameter( "verbose", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=False ), } for p in generate_sig_params: sig_params[p] = inspect.Parameter( p, inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ) generate_mock = MagicMock() generate_mock.__signature__ = inspect.Signature( parameters=list(sig_params.values()) ) generate_mock.return_value = [] # no audio chunks # Plain object — hasattr only returns True for explicitly set attrs class FakeModel: pass fake_model = FakeModel() fake_model.generate = generate_mock engine._model = fake_model try: asyncio.run( engine.synthesize( "Hello", voice=voice_value, instructions=instructions_value, **synth_kwargs, ) ) except RuntimeError: pass # "no audio output" is expected with empty generate return fake_model.generate.call_args return _run def test_customvoice_routes_to_voice(self, _run_synthesize): """Model with both params: voice goes to voice only, not instruct.""" call = _run_synthesize(["voice", "instruct"], voice_value="Vivian") kwargs = call.kwargs if call else {} assert kwargs.get("voice") == "Vivian" assert "instruct" not in kwargs def test_voicedesign_routes_to_instruct(self, _run_synthesize): """Model with only 'instruct' param: value goes to instruct.""" call = _run_synthesize(["instruct"], voice_value="female, calm, slow") kwargs = call.kwargs if call else {} assert kwargs.get("instruct") == "female, calm, slow" assert "voice" not in kwargs def test_voice_only_model(self, _run_synthesize): """Model with only 'voice' param (e.g. Kokoro): value goes to voice.""" call = _run_synthesize(["voice"], voice_value="af_heart") kwargs = call.kwargs if call else {} assert kwargs.get("voice") == "af_heart" def test_voice_none_skips_routing(self, _run_synthesize): """voice=None should not add voice or instruct kwargs.""" call = _run_synthesize(["voice", "instruct"], voice_value=None) kwargs = call.kwargs if call else {} assert "voice" not in kwargs assert "instruct" not in kwargs def test_instructions_routes_to_instruct(self, _run_synthesize): """instructions value should be routed to the instruct kwarg.""" call = _run_synthesize( ["voice", "instruct"], instructions_value="female, calm, slow", ) kwargs = call.kwargs if call else {} assert kwargs.get("instruct") == "female, calm, slow" assert "voice" not in kwargs def test_voice_and_instructions_both_passed(self, _run_synthesize): """CustomVoice: voice→voice kwarg, instructions→instruct kwarg.""" call = _run_synthesize( ["voice", "instruct"], voice_value="Vivian", instructions_value="female, calm, slow", ) kwargs = call.kwargs if call else {} assert kwargs.get("voice") == "Vivian" assert kwargs.get("instruct") == "female, calm, slow" def test_language_routes_to_lang_code(self, _run_synthesize): """language should be routed to lang_code when the backend accepts it.""" call = _run_synthesize(["voice", "lang_code"], language="English") kwargs = call.kwargs if call else {} assert kwargs.get("lang_code") == "English" def test_language_skipped_without_lang_code(self, _run_synthesize): """language should not be passed to backends without lang_code.""" call = _run_synthesize(["voice"], language="English") kwargs = call.kwargs if call else {} assert "lang_code" not in kwargs def test_positional_speed_remains_compatible(self): """Adding language must not change the old positional speed argument.""" import asyncio import inspect from omlx.engine.tts import TTSEngine engine = TTSEngine("test-model") sig_params = { "text": inspect.Parameter("text", inspect.Parameter.POSITIONAL_OR_KEYWORD), "voice": inspect.Parameter( "voice", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ), "speed": inspect.Parameter( "speed", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=1.0 ), "lang_code": inspect.Parameter( "lang_code", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ), "verbose": inspect.Parameter( "verbose", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=False ), } generate_mock = MagicMock() generate_mock.__signature__ = inspect.Signature( parameters=list(sig_params.values()) ) generate_mock.return_value = [] class FakeModel: pass fake_model = FakeModel() fake_model.generate = generate_mock engine._model = fake_model try: asyncio.run(engine.synthesize("Hello", "Vivian", 0.75)) except RuntimeError: pass kwargs = fake_model.generate.call_args.kwargs assert kwargs.get("voice") == "Vivian" assert kwargs.get("speed") == 0.75 assert "lang_code" not in kwargs def test_language_keeps_trailing_position_in_engine_signatures(self): """language stays after existing positional parameters.""" import inspect from omlx.engine.tts import TTSEngine synth_params = list(inspect.signature(TTSEngine.synthesize).parameters) stream_params = list( inspect.signature(TTSEngine.stream_synthesize_pcm).parameters ) assert synth_params.index("language") > synth_params.index("max_tokens") assert stream_params.index("language") > stream_params.index( "streaming_interval" ) # --------------------------------------------------------------------------- # TestTTSVoiceClonePassthrough — unit tests for ref_audio/ref_text passthrough # --------------------------------------------------------------------------- class TestTTSVoiceClonePassthrough: """Verify ref_audio and ref_text are forwarded to model.generate().""" @pytest.fixture def _run_synthesize_clone(self): """Helper: run TTSEngine.synthesize with ref_audio/ref_text and return generate() kwargs.""" import asyncio from omlx.engine.tts import TTSEngine def _run(ref_audio_path=None, ref_text=None): engine = TTSEngine("test-model") import inspect sig_params = { "text": inspect.Parameter( "text", inspect.Parameter.POSITIONAL_OR_KEYWORD ), "verbose": inspect.Parameter( "verbose", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=False ), "voice": inspect.Parameter( "voice", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ), "ref_audio": inspect.Parameter( "ref_audio", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ), "ref_text": inspect.Parameter( "ref_text", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ), } generate_mock = MagicMock() generate_mock.__signature__ = inspect.Signature( parameters=list(sig_params.values()) ) generate_mock.return_value = [] class FakeModel: pass fake_model = FakeModel() fake_model.generate = generate_mock engine._model = fake_model try: asyncio.run( engine.synthesize( "Hello", ref_audio=ref_audio_path, ref_text=ref_text, ) ) except RuntimeError: pass # "no audio output" expected return fake_model.generate.call_args return _run def test_ref_audio_passed_to_generate(self, _run_synthesize_clone): """ref_audio path is forwarded to model.generate().""" call = _run_synthesize_clone(ref_audio_path="/tmp/ref.wav", ref_text="hello") kwargs = call.kwargs if call else {} assert kwargs.get("ref_audio") == "/tmp/ref.wav" assert kwargs.get("ref_text") == "hello" def test_ref_audio_none_not_passed(self, _run_synthesize_clone): """When ref_audio is None, neither ref_audio nor ref_text appear in kwargs.""" call = _run_synthesize_clone(ref_audio_path=None, ref_text=None) kwargs = call.kwargs if call else {} assert "ref_audio" not in kwargs assert "ref_text" not in kwargs def test_ref_audio_without_ref_text(self, _run_synthesize_clone): """ref_audio without ref_text passes ref_audio and ref_text=None.""" call = _run_synthesize_clone(ref_audio_path="/tmp/ref.wav", ref_text=None) kwargs = call.kwargs if call else {} assert kwargs.get("ref_audio") == "/tmp/ref.wav" assert kwargs.get("ref_text") is None # --------------------------------------------------------------------------- # TestTTSVoiceCloneEndpoint — base64 ref_audio handling in the route layer # --------------------------------------------------------------------------- class TestTTSVoiceCloneEndpoint: """POST /v1/audio/speech with ref_audio base64.""" @pytest.fixture def clone_client(self): """TestClient with mocked TTS pool for voice clone tests.""" from omlx.server import app _ensure_audio_routes(app) mock_pool = _make_mock_pool() with patch("omlx.server._server_state") as mock_state: mock_state.engine_pool = mock_pool mock_state.global_settings = None mock_state.process_memory_enforcer = None mock_state.hf_downloader = None mock_state.ms_downloader = None mock_state.mcp_manager = None mock_state.api_key = None mock_state.settings_manager = MagicMock() with TestClient(app, raise_server_exceptions=False) as client: yield client, mock_pool def test_ref_audio_base64_accepted(self, clone_client): """Valid base64 ref_audio returns 200.""" client, _ = clone_client wav_b64 = base64.b64encode(_make_wav_bytes(0.5)).decode() response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Clone this voice", "ref_audio": wav_b64, "ref_text": "Reference text", }, ) assert response.status_code == 200 def test_ref_audio_forwarded_to_synthesize(self, clone_client): """ref_audio is decoded and passed as a file path to engine.synthesize().""" client, mock_pool = clone_client wav_b64 = base64.b64encode(_make_wav_bytes(0.5)).decode() client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Clone test", "ref_audio": wav_b64, "ref_text": "Hello", }, ) synthesize: AsyncMock = mock_pool.get_engine.return_value.synthesize assert synthesize.called call_kwargs = synthesize.call_args.kwargs assert call_kwargs.get("ref_text") == "Hello" # ref_audio should be a temp file path string ref_path = call_kwargs.get("ref_audio") assert ref_path is not None assert isinstance(ref_path, str) def test_invalid_base64_returns_400(self, clone_client): """Malformed base64 in ref_audio returns 400.""" client, _ = clone_client response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Bad audio", "ref_audio": "not-valid-base64!!!", "ref_text": "Hello", }, ) assert response.status_code == 400 body = response.json() # The server wraps errors as {"error": {"message": ...}} or {"detail": ...} message = body.get("detail") or body.get("error", {}).get("message", "") assert "base64" in message.lower() def test_oversized_ref_audio_returns_413(self, clone_client): """ref_audio exceeding size limit returns 413.""" client, _ = clone_client from omlx.api.audio_routes import MAX_REF_AUDIO_BASE64_BYTES # Create a base64 string just over the limit huge_b64 = base64.b64encode(b"\x00" * (MAX_REF_AUDIO_BASE64_BYTES)).decode() response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Too big", "ref_audio": huge_b64, "ref_text": "some text", }, ) assert response.status_code == 413 def test_temp_file_cleaned_up(self, clone_client): """Temp file is deleted after synthesis completes.""" client, mock_pool = clone_client wav_b64 = base64.b64encode(_make_wav_bytes(0.5)).decode() client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Cleanup test", "ref_audio": wav_b64, }, ) synthesize = mock_pool.get_engine.return_value.synthesize if synthesize.called: ref_path = synthesize.call_args.kwargs.get("ref_audio") if ref_path: import os assert not os.path.exists(ref_path), "Temp file should be deleted" def test_ref_audio_without_ref_text_returns_400(self, clone_client): """ref_audio without ref_text returns 400.""" client, _ = clone_client wav_b64 = base64.b64encode(_make_wav_bytes(0.5)).decode() response = client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Missing ref_text", "ref_audio": wav_b64, }, ) assert response.status_code == 400 detail = response.json().get("detail") or response.json().get("error", {}).get( "message", "" ) assert "ref_text" in detail.lower() def test_no_ref_audio_unchanged_behavior(self, clone_client): """Normal TTS (no ref_audio) still works as before.""" client, mock_pool = clone_client response = client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Normal TTS"}, ) assert response.status_code == 200 synthesize = mock_pool.get_engine.return_value.synthesize if synthesize.called: call_kwargs = synthesize.call_args.kwargs assert call_kwargs.get("ref_audio") is None # --------------------------------------------------------------------------- # TestTTSGenerationParams — generation param passthrough to standard path # --------------------------------------------------------------------------- class TestTTSGenerationParams: """Verify generation params are forwarded to model.generate().""" def test_temperature_forwarded(self, _run_synthesize): """temperature is passed to generate().""" call = _run_synthesize(["voice"], temperature=0.9) kwargs = call.kwargs if call else {} assert kwargs.get("temperature") == 0.9 def test_top_k_forwarded(self, _run_synthesize): """top_k is passed to generate().""" call = _run_synthesize(["voice"], top_k=50) kwargs = call.kwargs if call else {} assert kwargs.get("top_k") == 50 def test_top_p_forwarded(self, _run_synthesize): """top_p is passed to generate().""" call = _run_synthesize(["voice"], top_p=0.95) kwargs = call.kwargs if call else {} assert kwargs.get("top_p") == 0.95 def test_repetition_penalty_forwarded(self, _run_synthesize): """repetition_penalty is passed to generate().""" call = _run_synthesize(["voice"], repetition_penalty=1.05) kwargs = call.kwargs if call else {} assert kwargs.get("repetition_penalty") == 1.05 def test_max_tokens_forwarded(self, _run_synthesize): """max_tokens is passed to generate().""" call = _run_synthesize(["voice"], max_tokens=2048) kwargs = call.kwargs if call else {} assert kwargs.get("max_tokens") == 2048 def test_none_params_not_forwarded(self, _run_synthesize): """None generation params are not included in kwargs.""" call = _run_synthesize(["voice"]) kwargs = call.kwargs if call else {} for key in ( "temperature", "top_k", "top_p", "repetition_penalty", "max_tokens", ): assert key not in kwargs @pytest.fixture def _run_synthesize(self): """Reuse voice routing fixture pattern for gen param tests.""" import asyncio from omlx.engine.tts import TTSEngine def _run(generate_sig_params, **synth_kwargs): engine = TTSEngine("test-model") import inspect sig_params = { "text": inspect.Parameter( "text", inspect.Parameter.POSITIONAL_OR_KEYWORD ), "verbose": inspect.Parameter( "verbose", inspect.Parameter.POSITIONAL_OR_KEYWORD, default=False ), } for p in generate_sig_params: sig_params[p] = inspect.Parameter( p, inspect.Parameter.POSITIONAL_OR_KEYWORD, default=None ) generate_mock = MagicMock() generate_mock.__signature__ = inspect.Signature( parameters=list(sig_params.values()) ) generate_mock.return_value = [] class FakeModel: pass fake_model = FakeModel() fake_model.generate = generate_mock engine._model = fake_model try: asyncio.run(engine.synthesize("Hello", **synth_kwargs)) except RuntimeError: pass return fake_model.generate.call_args return _run # --------------------------------------------------------------------------- # TestTTSGenParamsEndpoint — generation params accepted via API # --------------------------------------------------------------------------- class TestTTSGenParamsEndpoint: """Verify generation params are accepted and forwarded by the endpoint.""" def test_gen_params_forwarded_to_synthesize(self, server_tts_client): """Generation params from request body reach engine.synthesize().""" client, mock_pool = server_tts_client client.post( "/v1/audio/speech", json={ "model": "qwen3-tts", "input": "Hello", "temperature": 0.8, "top_k": 30, "top_p": 0.95, "repetition_penalty": 1.1, "max_tokens": 1024, }, ) synthesize: AsyncMock = mock_pool.get_engine.return_value.synthesize assert synthesize.called call_kwargs = synthesize.call_args.kwargs assert call_kwargs.get("temperature") == 0.8 assert call_kwargs.get("top_k") == 30 assert call_kwargs.get("top_p") == 0.95 assert call_kwargs.get("repetition_penalty") == 1.1 assert call_kwargs.get("max_tokens") == 1024 def test_gen_params_default_none(self, server_tts_client): """Without gen params in request, they're passed as None.""" client, mock_pool = server_tts_client client.post( "/v1/audio/speech", json={"model": "qwen3-tts", "input": "Hello"}, ) synthesize: AsyncMock = mock_pool.get_engine.return_value.synthesize assert synthesize.called call_kwargs = synthesize.call_args.kwargs assert call_kwargs.get("temperature") is None assert call_kwargs.get("top_k") is None # --------------------------------------------------------------------------- # Integration test (slow, requires mlx-audio) # --------------------------------------------------------------------------- @pytest.mark.slow class TestTTSIntegration: """Integration tests requiring a real mlx-audio TTS model. Skip if mlx-audio is not installed or models are unavailable. """ def test_real_synthesis_produces_wav(self): """Real synthesis with actual mlx-audio TTS model produces playable WAV.""" pytest.importorskip("mlx_audio") from omlx.engine.tts import TTSEngine model_name = "mlx-community/Kokoro-82M-mlx" try: import asyncio engine = TTSEngine(model_name) asyncio.run(engine.start()) result = asyncio.run(engine.synthesize("Hello world", voice="af_heart")) assert isinstance(result, bytes) assert result[:4] == RIFF_MAGIC asyncio.run(engine.stop()) except Exception as e: pytest.skip(f"Could not run integration test: {e}")