From 7a2c6bb2093e67bee1af1bfed50964ea0f9755c0 Mon Sep 17 00:00:00 2001 From: stephan Date: Tue, 9 Dec 2025 12:53:07 +0100 Subject: [PATCH] test: Add comprehensive tests for OpenAI-compatible TTS endpoint - Refactor existing tests to use a shared mock engine fixture. - Add tests for unsupported response formats. - Add tests for engine synthesis failures. - Add tests for 'engine:model_id' parsing in the 'model' parameter. - Add tests for different audio response formats (opus, flac). --- tests/test_openai_api.py | 154 +++++++++++++++++++++++++++++++++------ 1 file changed, 130 insertions(+), 24 deletions(-) diff --git a/tests/test_openai_api.py b/tests/test_openai_api.py index 1da693d..02007e2 100644 --- a/tests/test_openai_api.py +++ b/tests/test_openai_api.py @@ -2,37 +2,52 @@ import pytest from fastapi.testclient import TestClient from unittest.mock import AsyncMock, MagicMock from app.routers.openai_compatible import OpenAISpeechRequest +import os -def test_openai_speech_endpoint_success(app_client, app_instance, tmp_path): - # 1. Setup Mock Engine +@pytest.fixture(autouse=True) +def mock_engine_registry(app_instance, tmp_path): + """ + Fixture to set up a mock engine and inject it into the app_instance's registry + for each test. This ensures a clean state and isolated mocks. + """ mock_engine = MagicMock() + # Mock synthesis to return a dummy file path dummy_file = tmp_path / "test_output.wav" dummy_file.write_bytes(b"fake audio data") mock_engine.synthesize = AsyncMock(return_value=str(dummy_file)) - # Inject mock engine into registry + # Add common mock methods for engine interface + mock_engine.list_models.return_value = ["model1", "model2"] + mock_engine.list_voices.return_value = ["speaker1", "speaker2"] + mock_engine.healthcheck.return_value = {"status": "ok"} + app_instance.ENGINE_REGISTRY["mock-engine"] = mock_engine - - # 2. Make Request + app_instance.ENGINE_REGISTRY["kokoro"] = mock_engine # For tts-1 mapping test + + return mock_engine + +# --- Existing Tests (Modified to use fixture) --- +def test_openai_speech_endpoint_success(app_client, mock_engine_registry, tmp_path): + # Use the mock_engine_registry fixture to get the mock_engine + mock_engine = mock_engine_registry + payload = { "model": "mock-engine", "input": "Hello OpenAI", - "voice": "default", + "voice": "speaker1", "response_format": "wav" } response = app_client.post("/v1/audio/speech", json=payload) - # 3. Assertions assert response.status_code == 200 assert response.headers["content-type"] == "audio/wav" assert response.content == b"fake audio data" - # Verify engine call mock_engine.synthesize.assert_called_once_with( text="Hello OpenAI", - speaker="default", - model=None, + speaker="speaker1", + model=None, # Explicitly no model_id in this case fmt="wav" ) @@ -46,21 +61,13 @@ def test_openai_speech_endpoint_engine_not_found(app_client): assert response.status_code == 404 assert "not found" in response.json()["detail"] -def test_openai_speech_endpoint_tts_1_mapping(app_client, app_instance, tmp_path): - # Setup Mock Engine for 'kokoro' (simulating it's active) - mock_engine = MagicMock() - dummy_file = tmp_path / "tts1_output.mp3" - dummy_file.write_bytes(b"mp3 data") - mock_engine.synthesize = AsyncMock(return_value=str(dummy_file)) - - # Inject into registry - app_instance.ENGINE_REGISTRY["kokoro"] = mock_engine - - # Request 'tts-1' -> should map to 'kokoro' +def test_openai_speech_endpoint_tts_1_mapping(app_client, mock_engine_registry): + mock_engine = mock_engine_registry # 'kokoro' engine is mapped to this mock + payload = { "model": "tts-1", "input": "Mapping test", - "voice": "alloy", + "voice": "speaker1", "response_format": "mp3" } response = app_client.post("/v1/audio/speech", json=payload) @@ -69,6 +76,105 @@ def test_openai_speech_endpoint_tts_1_mapping(app_client, app_instance, tmp_path assert response.headers["content-type"] == "audio/mpeg" mock_engine.synthesize.assert_called_once() - # Check that speaker passed through args, kwargs = mock_engine.synthesize.call_args - assert kwargs["speaker"] == "alloy" + assert kwargs["speaker"] == "speaker1" + assert kwargs["fmt"] == "mp3" + + +# --- New Test Cases --- + +def test_openai_speech_endpoint_unsupported_format(app_client, mock_engine_registry): + mock_engine = mock_engine_registry + + payload = { + "model": "mock-engine", + "input": "Test text.", + "voice": "speaker1", + "response_format": "unsupported_format" # This should be caught by pydantic's Literal + } + response = app_client.post("/v1/audio/speech", json=payload) + + assert response.status_code == 422 # Unprocessable Entity due to Pydantic validation + assert "response_format" in response.json()["detail"][0]["loc"] + +def test_openai_speech_endpoint_engine_synthesis_failure(app_client, mock_engine_registry): + mock_engine = mock_engine_registry + mock_engine.synthesize.side_effect = RuntimeError("Mock synthesis failed") + + payload = { + "model": "mock-engine", + "input": "This text will fail.", + "voice": "speaker1" + } + response = app_client.post("/v1/audio/speech", json=payload) + + assert response.status_code == 500 + assert "Mock synthesis failed" in response.json()["detail"] + +def test_openai_speech_endpoint_model_id_parsing(app_client, mock_engine_registry, tmp_path): + mock_engine = mock_engine_registry + # Ensure a different dummy file for this test's synthesis result + dummy_file = tmp_path / "model_id_output.wav" + dummy_file.write_bytes(b"model id audio data") + mock_engine.synthesize.return_value = str(dummy_file) + + # For this test, we simulate an engine that supports a model_id + app_instance = app_client.app + app_instance.ENGINE_REGISTRY["xtts"] = mock_engine # Map 'xtts' to our mock + + payload = { + "model": "xtts:multilingual", # Test parsing 'engine:model_id' + "input": "Testing specific model.", + "voice": "speaker1", + "response_format": "wav" + } + response = app_client.post("/v1/audio/speech", json=payload) + + assert response.status_code == 200 + assert response.headers["content-type"] == "audio/wav" + assert response.content == b"model id audio data" + + mock_engine.synthesize.assert_called_once_with( + text="Testing specific model.", + speaker="speaker1", + model="multilingual", # Verify model_id was passed correctly + fmt="wav" + ) + +def test_openai_speech_endpoint_different_audio_formats(app_client, mock_engine_registry, tmp_path): + mock_engine = mock_engine_registry + + # Test opus format + opus_dummy_file = tmp_path / "test_output.opus" + opus_dummy_file.write_bytes(b"opus data") + mock_engine.synthesize.return_value = str(opus_dummy_file) + + payload_opus = { + "model": "mock-engine", + "input": "Hello opus.", + "voice": "speaker1", + "response_format": "opus" + } + response_opus = app_client.post("/v1/audio/speech", json=payload_opus) + assert response_opus.status_code == 200 + assert response_opus.headers["content-type"] == "audio/opus" + assert response_opus.content == b"opus data" + mock_engine.synthesize.assert_called_with(text="Hello opus.", speaker="speaker1", model=None, fmt="opus") + + # Reset mock and test flac format + mock_engine.synthesize.reset_mock() + flac_dummy_file = tmp_path / "test_output.flac" + flac_dummy_file.write_bytes(b"flac data") + mock_engine.synthesize.return_value = str(flac_dummy_file) + + payload_flac = { + "model": "mock-engine", + "input": "Hello flac.", + "voice": "speaker1", + "response_format": "flac" + } + response_flac = app_client.post("/v1/audio/speech", json=payload_flac) + assert response_flac.status_code == 200 + assert response_flac.headers["content-type"] == "audio/flac" + assert response_flac.content == b"flac data" + mock_engine.synthesize.assert_called_with(text="Hello flac.", speaker="speaker1", model=None, fmt="flac") \ No newline at end of file