import pytest from fastapi.testclient import TestClient from unittest.mock import AsyncMock, MagicMock from app.routers.openai_compatible import OpenAISpeechRequest import os @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)) # 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 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": "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"fake audio data" mock_engine.synthesize.assert_called_once_with( text="Hello OpenAI", speaker="speaker1", model=None, # Explicitly no model_id in this case fmt="wav" ) def test_openai_speech_endpoint_engine_not_found(app_client): payload = { "model": "non-existent-engine", "input": "Test", "voice": "default" } response = app_client.post("/v1/audio/speech", json=payload) assert response.status_code == 404 assert "not found" in response.json()["detail"] 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": "speaker1", "response_format": "mp3" } response = app_client.post("/v1/audio/speech", json=payload) assert response.status_code == 200 assert response.headers["content-type"] == "audio/mpeg" mock_engine.synthesize.assert_called_once() args, kwargs = mock_engine.synthesize.call_args 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")