vllm-omni/tests/entrypoints/openai_api/test_serving_speech.py

# tests/entrypoints/openai/test_serving_speech.py
import asyncio
import logging
import os
import struct
from inspect import Signature, signature
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import numpy as np
import pytest
import torch
from fastapi import FastAPI, HTTPException, Request, UploadFile
from fastapi.params import File, Form
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from pydantic import ValidationError
from pytest_mock import MockerFixture
from vllm.entrypoints.openai.engine.protocol import ErrorInfo, ErrorResponse

from vllm_omni.entrypoints.openai import api_server as api_server_module
from vllm_omni.entrypoints.openai.audio_utils_mixin import AudioMixin
from vllm_omni.entrypoints.openai.protocol.audio import (
    BatchSpeechRequest,
    CreateAudio,
    OpenAICreateSpeechRequest,
    SpeechBatchItem,
)
from vllm_omni.entrypoints.openai.serving_speech import (
    OmniOpenAIServingSpeech,
    _create_wav_header,
)
from vllm_omni.outputs import OmniRequestOutput

pytestmark = [pytest.mark.core_model, pytest.mark.cpu]

logger = logging.getLogger(__name__)


class TestAudioMixin:
    @pytest.fixture
    def audio_mixin(self):
        return AudioMixin()

    def test_stereo_to_mono_conversion(self, audio_mixin, mocker: MockerFixture):
        stereo_tensor = np.random.rand(24000, 2).astype(np.float32)
        audio_obj = CreateAudio(audio_tensor=stereo_tensor)

        mock_speed = mocker.patch.object(
            audio_mixin, "_apply_speed_adjustment", side_effect=lambda tensor, speed, sr: (tensor, sr)
        )
        mocker.patch("soundfile.write")

        audio_mixin.create_audio(audio_obj)

        # Check that the tensor passed to speed adjustment is mono
        mock_speed.assert_called_once()
        adjusted_tensor = mock_speed.call_args[0][0]
        assert len(adjusted_tensor) == 24000

    def test_speed_adjustment(self, audio_mixin, mocker: MockerFixture):
        mock_time_stretch = mocker.patch("librosa.effects.time_stretch")
        mock_time_stretch.return_value = np.zeros(12000)
        audio_tensor = np.random.rand(24000).astype(np.float32)

        adjusted_audio, _ = audio_mixin._apply_speed_adjustment(audio_tensor, speed=2.0, sample_rate=24000)

        mock_time_stretch.assert_called_with(y=audio_tensor, rate=2.0)
        assert adjusted_audio.shape == (12000,)

    def test_unsupported_format_fallback(self, audio_mixin, caplog, mocker: MockerFixture):
        mock_write = mocker.patch("soundfile.write")
        audio_tensor = np.random.rand(24000).astype(np.float32)
        # Use a format that is not in the list of supported formats
        audio_obj = CreateAudio(audio_tensor=audio_tensor, response_format="vorbis")

        audio_mixin.create_audio(audio_obj)

        # Should fall back to 'wav'
        mock_write.assert_called_once()
        write_kwargs = mock_write.call_args.kwargs
        assert write_kwargs["format"] == "WAV"

    def test_mono_audio_preservation(self, audio_mixin, mocker: MockerFixture):
        """Test that mono (1D) audio tensors are processed correctly and passed to writer."""
        mono_tensor = np.random.rand(24000).astype(np.float32)
        audio_obj = CreateAudio(audio_tensor=mono_tensor)

        mock_write = mocker.patch("soundfile.write")
        audio_mixin.create_audio(audio_obj)

        mock_write.assert_called_once()
        # Verify the tensor passed to soundfile.write is the exact 1D tensor
        output_tensor = mock_write.call_args[0][1]
        assert output_tensor.ndim == 1
        assert output_tensor.shape == (24000,)
        assert np.array_equal(output_tensor, mono_tensor)

    def test_stereo_audio_preservation(self, audio_mixin, mocker: MockerFixture):
        """Test that stereo (2D) audio tensors are processed correctly and preserved."""
        stereo_tensor = np.random.rand(24000, 2).astype(np.float32)
        audio_obj = CreateAudio(audio_tensor=stereo_tensor)

        mock_write = mocker.patch("soundfile.write")
        audio_mixin.create_audio(audio_obj)

        mock_write.assert_called_once()
        # Verify the tensor passed to soundfile.write is the exact 2D tensor
        output_tensor = mock_write.call_args[0][1]
        assert output_tensor.ndim == 2
        assert output_tensor.shape == (24000, 2)
        assert np.array_equal(output_tensor, stereo_tensor)

    def test_speed_adjustment_bypass(self, audio_mixin, mocker: MockerFixture):
        """Test that speed=1.0 bypasses the expensive librosa time stretching."""
        audio_tensor = np.random.rand(24000).astype(np.float32)

        mock_time_stretch = mocker.patch("librosa.effects.time_stretch")
        # speed=1.0 should return immediately without calling librosa
        result, _ = audio_mixin._apply_speed_adjustment(audio_tensor, speed=1.0, sample_rate=24000)

        mock_time_stretch.assert_not_called()
        assert np.array_equal(result, audio_tensor)

    def test_speed_adjustment_stereo_handling(self, audio_mixin, mocker: MockerFixture):
        """Test that speed adjustment is attempted on stereo inputs."""
        mock_time_stretch = mocker.patch("librosa.effects.time_stretch")
        stereo_tensor = np.random.rand(24000, 2).astype(np.float32)
        # Mock return value representing a sped-up version (half length)
        mock_time_stretch.return_value = np.zeros((12000, 2), dtype=np.float32)

        result, _ = audio_mixin._apply_speed_adjustment(stereo_tensor, speed=2.0, sample_rate=24000)

        mock_time_stretch.assert_called_once()
        # Ensure the stereo tensor was passed to librosa
        call_args = mock_time_stretch.call_args
        assert np.array_equal(call_args.kwargs["y"], stereo_tensor)
        assert call_args.kwargs["rate"] == 2.0
        assert result.shape == (12000, 2)


# Helper to create mock model output for endpoint tests
def create_mock_audio_output_for_test(
    request_id: str = "speech-mock-123",
) -> OmniRequestOutput:
    class MockCompletionOutput:
        def __init__(self, index: int = 0):
            self.index = index
            self.text = ""
            self.token_ids = []
            self.finish_reason = "stop"
            self.stop_reason = None
            self.logprobs = None

    class MockRequestOutput:
        def __init__(self, request_id: str, audio_tensor: torch.Tensor):
            self.request_id = request_id
            self.outputs = [MockCompletionOutput(index=0)]
            self.multimodal_output = {"audio": audio_tensor}
            self.finished = True
            self.prompt_token_ids = None
            self.encoder_prompt_token_ids = None
            self.num_cached_tokens = None
            self.prompt_logprobs = None
            self.kv_transfer_params = None

    num_samples = 24000
    audio_tensor = torch.sin(torch.linspace(0, 440 * 2 * torch.pi, num_samples))
    mock_request_output = MockRequestOutput(request_id=request_id, audio_tensor=audio_tensor)

    return OmniRequestOutput(
        stage_id=0,
        final_output_type="audio",
        request_output=mock_request_output,
    )


@pytest.fixture
def test_app(mocker: MockerFixture):
    # Mock the engine client
    mock_engine_client = mocker.MagicMock()
    mock_engine_client.errored = False

    async def mock_generate_fn(*args, **kwargs):
        yield create_mock_audio_output_for_test(request_id=kwargs.get("request_id"))

    mock_engine_client.generate = mocker.MagicMock(side_effect=mock_generate_fn)
    mock_engine_client.default_sampling_params_list = [{}]
    mock_engine_client.tts_batch_max_items = 32

    # Mock models to have an is_base_model method
    mock_models = mocker.MagicMock()
    mock_models.is_base_model.return_value = True

    mock_request_logger = mocker.MagicMock()

    speech_server = OmniOpenAIServingSpeech(
        engine_client=mock_engine_client,
        models=mock_models,
        request_logger=mock_request_logger,
    )

    # Skip TTS validation in tests (mock doesn't set up supported_speakers)
    speech_server._validate_tts_request = mocker.MagicMock(return_value=None)

    # Patch the signature of create_speech to remove 'raw_request' for FastAPI route introspection
    original_create_speech = speech_server.create_speech
    _ = mocker.MagicMock(side_effect=original_create_speech)

    sig = signature(original_create_speech)

    new_parameters = [param for name, param in sig.parameters.items() if name != "raw_request"]

    new_sig = Signature(parameters=new_parameters, return_annotation=sig.return_annotation)

    async def awaitable_patched_create_speech(*args, **kwargs):
        return await original_create_speech(*args, **kwargs)

    awaitable_patched_create_speech.__signature__ = new_sig
    speech_server.create_speech = awaitable_patched_create_speech

    app = FastAPI()
    app.add_api_route("/v1/audio/speech", speech_server.create_speech, methods=["POST"], response_model=None)

    # Add list_voices endpoint
    async def list_voices():
        speakers = sorted(speech_server.supported_speakers) if speech_server.supported_speakers else []
        uploaded_voices = []
        if hasattr(speech_server, "uploaded_speakers"):
            for voice_name, info in speech_server.uploaded_speakers.items():
                uploaded_voices.append(
                    {
                        "name": info.get("name", voice_name),
                        "consent": info.get("consent", ""),
                        "created_at": info.get("created_at", 0),
                        "file_size": info.get("file_size", 0),
                        "mime_type": info.get("mime_type", ""),
                        "embedding_source": info.get("embedding_source", "audio"),
                        "embedding_dim": info.get("embedding_dim"),
                    }
                )
        return {"voices": speakers, "uploaded_voices": uploaded_voices}

    app.add_api_route("/v1/audio/voices", list_voices, methods=["GET"])
    app.add_api_route("/v1/audio/speech/batch", speech_server.create_speech_batch, methods=["POST"])

    # Add upload_voice endpoint
    async def upload_voice(
        audio_sample: UploadFile | None = File(None),
        speaker_embedding: str | None = Form(None),
        consent: str = Form(...),
        name: str = Form(...),
        ref_text: str = Form(None),
    ):
        try:
            if speaker_embedding is not None and audio_sample is not None:
                raise ValueError("'audio_sample' and 'speaker_embedding' are mutually exclusive")
            if speaker_embedding is not None:
                result = await speech_server.upload_voice_embedding(speaker_embedding, consent, name)
            elif audio_sample is not None:
                result = await speech_server.upload_voice(audio_sample, consent, name, ref_text=ref_text)
            else:
                raise ValueError("Either 'audio_sample' or 'speaker_embedding' must be provided")
            return {"success": True, "voice": result}
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))
        except Exception as e:
            logger.exception(f"Failed to upload voice: {e}")
            raise HTTPException(status_code=500, detail=f"Failed to upload voice: {str(e)}")

    app.add_api_route("/v1/audio/voices", upload_voice, methods=["POST"])

    # Add delete_voice endpoint
    async def delete_voice(name: str):
        try:
            success = await speech_server.delete_voice(name)
            if not success:
                raise HTTPException(status_code=404, detail=f"Voice '{name}' not found")
            return {"success": True, "message": f"Voice '{name}' deleted successfully"}
        except HTTPException:
            raise
        except ValueError as e:
            raise HTTPException(status_code=400, detail=str(e))
        except Exception as e:
            logger.exception(f"Failed to delete voice '{name}': {e}")
            raise HTTPException(status_code=500, detail=f"Failed to delete voice: {str(e)}")

    app.add_api_route("/v1/audio/voices/{name}", delete_voice, methods=["DELETE"])

    return app


@pytest.fixture
def client(test_app):
    return TestClient(test_app)


class TestSpeechAPI:
    def test_create_speech_success(self, client):
        payload = {
            "input": "Hello world",
            "model": "tts-model",
            "voice": "alloy",
            "response_format": "wav",
        }
        response = client.post("/v1/audio/speech", json=payload)
        assert response.status_code == 200
        assert response.headers["content-type"] == "audio/wav"
        assert len(response.content) > 0

    def test_create_speech_mp3_format(self, client):
        payload = {
            "input": "Hello world",
            "model": "tts-model",
            "voice": "alloy",
            "response_format": "mp3",
        }
        response = client.post("/v1/audio/speech", json=payload)
        assert response.status_code == 200
        assert response.headers["content-type"] == "audio/mpeg"
        assert len(response.content) > 0

    def test_create_speech_invalid_format(self, client):
        payload = {
            "input": "Hello world",
            "model": "tts-model",
            "voice": "alloy",
            "response_format": "invalid_format",
        }
        response = client.post("/v1/audio/speech", json=payload)
        assert response.status_code == 422  # Unprocessable Entity

    def test_speed_parameter_is_used(self, test_app, mocker: MockerFixture):
        mock_create_audio = mocker.patch(
            "vllm_omni.entrypoints.openai.serving_speech.OmniOpenAIServingSpeech.create_audio"
        )
        client = TestClient(test_app)

        mock_audio_response = mocker.MagicMock()
        mock_audio_response.audio_data = b"dummy_audio"
        mock_audio_response.media_type = "audio/wav"
        mock_create_audio.return_value = mock_audio_response

        payload = {
            "input": "This should be fast.",
            "model": "tts-model",
            "voice": "alloy",
            "response_format": "wav",
            "speed": 2.5,
        }
        client.post("/v1/audio/speech", json=payload)

        mock_create_audio.assert_called_once()
        call_args = mock_create_audio.call_args[0]
        audio_obj = call_args[0]
        assert isinstance(audio_obj, CreateAudio)
        assert audio_obj.speed == 2.5

    def test_list_voices_endpoint(self, client):
        response = client.get("/v1/audio/voices")
        assert response.status_code == 200
        assert "voices" in response.json()

    def test_upload_voice_success(self, client, tmp_path):
        """Test successful voice upload without ref_text."""
        audio_content = b"fake audio content" * 1000
        files = {"audio_sample": ("test.wav", audio_content, "audio/wav")}
        data = {"consent": "user_consent_123", "name": "test_voice"}

        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 200
        result = response.json()
        assert result["success"] is True
        voice_info = result["voice"]
        assert voice_info["name"] == "test_voice"
        assert voice_info["consent"] == "user_consent_123"
        assert voice_info["mime_type"] == "audio/wav"
        assert voice_info["file_size"] == len(audio_content)
        response = client.delete("/v1/audio/voices/test_voice")

    def test_upload_voice_with_ref_text(self, client, tmp_path):
        """Test voice upload with ref_text enables in-context cloning."""
        audio_content = b"fake audio content" * 1000
        files = {"audio_sample": ("test.wav", audio_content, "audio/wav")}
        data = {"consent": "c1", "name": "test_voice_rt", "ref_text": "Hello world transcript"}

        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 200
        result = response.json()
        assert result["success"] is True
        assert result["voice"]["name"] == "test_voice_rt"
        assert result["voice"].get("ref_text") == "Hello world transcript"
        response = client.delete("/v1/audio/voices/test_voice_rt")

    def test_upload_voice_file_too_large(self, client):
        """Test voice upload with file exceeding size limit."""
        # Create a file larger than 10MB
        audio_content = b"x" * (11 * 1024 * 1024)  # 11MB
        files = {
            "audio_sample": ("test.wav", audio_content, "audio/wav"),
        }
        data = {
            "consent": "user_consent_123",
            "name": "test_voice",
        }

        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 400
        result = response.json()
        assert "detail" in result
        assert "10MB" in result["detail"]

    def test_upload_voice_invalid_mime_type(self, client):
        """Test voice upload with invalid MIME type."""
        audio_content = b"fake audio content"
        files = {
            "audio_sample": ("test.txt", audio_content, "text/plain"),
        }
        data = {
            "consent": "user_consent_123",
            "name": "test_voice",
        }

        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 400
        result = response.json()
        assert "detail" in result
        assert "MIME type" in result["detail"]

    def test_upload_voice_name_collision(self, client):
        """Test voice upload with duplicate name."""
        # First upload
        audio_content = b"fake audio content"
        files = {
            "audio_sample": ("test.wav", audio_content, "audio/wav"),
        }
        data = {
            "consent": "user_consent_123",
            "name": "test_voice",
        }

        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 200

        # Second upload with same name
        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 400
        result = response.json()
        assert "detail" in result
        assert "already exists" in result["detail"]
        response = client.delete("/v1/audio/voices/test_voice")

    def test_upload_voice_missing_parameters(self, client):
        """Test voice upload with missing required parameters."""
        audio_content = b"fake audio content"
        files = {
            "audio_sample": ("test.wav", audio_content, "audio/wav"),
        }

        # Missing consent
        data = {"name": "test_voice5"}
        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 422  # Validation error

        # Missing name
        data = {"consent": "user_consent_123"}
        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 422  # Validation error

        # Missing both audio_sample and speaker_embedding
        data = {
            "consent": "user_consent_123",
            "name": "test_voice6",
        }
        response = client.post("/v1/audio/voices", data=data)
        assert response.status_code == 400

    def test_delete_voice_success(self, client):
        """Test successful voice deletion."""
        # First upload a voice
        audio_content = b"fake audio content"
        files = {
            "audio_sample": ("test.wav", audio_content, "audio/wav"),
        }
        data = {
            "consent": "user_consent_123",
            "name": "test_voice7",
        }

        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 200

        # Then delete it
        response = client.delete("/v1/audio/voices/test_voice7")
        assert response.status_code == 200
        result = response.json()
        assert result["success"] is True
        assert "deleted successfully" in result["message"]

        # Verify it's gone by trying to delete again
        response = client.delete("/v1/audio/voices/test_voice7")
        assert response.status_code == 404
        result = response.json()
        assert "not found" in result["detail"]

    def test_delete_voice_not_found(self, client):
        """Test deleting a non-existent voice."""
        response = client.delete("/v1/audio/voices/nonexistent")
        assert response.status_code == 404
        result = response.json()
        assert "not found" in result["detail"]

    # ── speaker_embedding upload via voices endpoint ──

    def test_upload_voice_embedding_success(self, client):
        """Upload a voice via speaker_embedding JSON."""
        import json

        emb = [0.1] * 1024
        data = {
            "speaker_embedding": json.dumps(emb),
            "consent": "consent_emb_1",
            "name": "emb_voice",
        }
        response = client.post("/v1/audio/voices", data=data)
        assert response.status_code == 200, f"Upload failed: {response.text}"
        result = response.json()
        assert result["success"] is True
        voice = result["voice"]
        assert voice["name"] == "emb_voice"
        assert voice["embedding_source"] == "direct"
        assert voice["embedding_dim"] == 1024
        # Clean up
        client.delete("/v1/audio/voices/emb_voice")

    def test_upload_voice_embedding_appears_in_listing(self, client):
        """Embedding-uploaded voice appears in list with correct source."""
        import json

        emb = [0.2] * 2048
        data = {
            "speaker_embedding": json.dumps(emb),
            "consent": "consent_list",
            "name": "listed_emb_voice",
        }
        response = client.post("/v1/audio/voices", data=data)
        assert response.status_code == 200

        listing = client.get("/v1/audio/voices").json()
        uploaded = {v["name"]: v for v in listing["uploaded_voices"]}
        assert "listed_emb_voice" in uploaded
        assert uploaded["listed_emb_voice"]["embedding_source"] == "direct"
        assert uploaded["listed_emb_voice"]["embedding_dim"] == 2048
        # Clean up
        client.delete("/v1/audio/voices/listed_emb_voice")

    def test_upload_voice_embedding_and_audio_mutually_exclusive(self, client):
        """Providing both audio_sample and speaker_embedding returns 400."""
        import json

        emb = [0.1] * 1024
        files = {"audio_sample": ("test.wav", b"fake", "audio/wav")}
        data = {
            "speaker_embedding": json.dumps(emb),
            "consent": "consent_mx",
            "name": "mx_voice",
        }
        response = client.post("/v1/audio/voices", files=files, data=data)
        assert response.status_code == 400
        assert "mutually exclusive" in response.json()["detail"]

    def test_upload_voice_embedding_invalid_json(self, client):
        """Invalid JSON in speaker_embedding returns 400."""
        data = {
            "speaker_embedding": "not valid json [[[",
            "consent": "consent_bad",
            "name": "bad_json_voice",
        }
        response = client.post("/v1/audio/voices", data=data)
        assert response.status_code == 400
        assert "JSON" in response.json()["detail"]

    def test_upload_voice_embedding_nan_rejected(self, client):
        """NaN values in speaker_embedding return 400."""
        import json

        data = {
            "speaker_embedding": json.dumps([0.1] * 1023 + [float("nan")]),
            "consent": "consent_nan",
            "name": "nan_voice",
        }
        response = client.post("/v1/audio/voices", data=data)
        assert response.status_code == 400
        assert "finite" in response.json()["detail"]


class TestTTSMethods:
    """Unit tests for TTS validation and parameter building."""

    @pytest.fixture
    def speech_server(self, mocker: MockerFixture):
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        mock_engine_client.stage_configs = []
        mock_engine_client.tts_max_instructions_length = None
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True
        return OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )

    def test_is_tts_detection_no_stage(self, speech_server):
        """Test TTS model detection when no TTS stage exists."""
        # Fixture creates server with stage_configs = [] -> _is_tts should be False
        assert speech_server._is_tts is False
        assert speech_server._tts_stage is None

    def test_is_tts_detection_with_tts_stage(self, mocker: MockerFixture):
        """Test TTS model detection when TTS stage exists."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        mock_engine_client.tts_max_instructions_length = None

        # Create a TTS stage
        mock_stage = mocker.MagicMock()
        mock_stage.engine_args.model_stage = "qwen3_tts"
        mock_stage.tts_args = {}
        mock_engine_client.stage_configs = [mock_stage]

        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )

        assert server._is_tts is True
        assert server._tts_stage is mock_stage

    def test_estimate_prompt_len_fallback(self, speech_server):
        """Test prompt length estimation falls back to 2048 when model is unavailable."""
        tts_params = {"text": ["Hello"], "task_type": ["CustomVoice"]}
        result = speech_server._estimate_prompt_len(tts_params)
        # Without a real model, it should fall back to 2048.
        assert result == 2048

    def test_validate_tts_request_basic(self, speech_server):
        """Test basic validation cases."""
        # Empty input
        req = OpenAICreateSpeechRequest(input="")
        assert speech_server._validate_tts_request(req) == "Input text cannot be empty"

        # Invalid language
        req = OpenAICreateSpeechRequest(input="Hello", language="InvalidLang")
        assert "Invalid language" in speech_server._validate_tts_request(req)

        # CustomVoice on model with no speakers -> rejected
        req = OpenAICreateSpeechRequest(input="Hello", voice="Invalid")
        assert "does not support CustomVoice" in speech_server._validate_tts_request(req)

        # CustomVoice without voice on model with no speakers -> also rejected
        req = OpenAICreateSpeechRequest(input="Hello")
        assert "does not support CustomVoice" in speech_server._validate_tts_request(req)

    def test_validate_tts_request_task_types(self, speech_server):
        """Test task-specific validation."""
        # Base task requires ref_audio
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base")
        assert "ref_audio" in speech_server._validate_tts_request(req)

        # VoiceDesign requires instructions
        req = OpenAICreateSpeechRequest(input="Hello", task_type="VoiceDesign")
        assert "instructions" in speech_server._validate_tts_request(req)

        # ref_text without task_type auto-infers Base, then fails on missing ref_audio
        req = OpenAICreateSpeechRequest(input="Hello", ref_text="text")
        assert "ref_audio" in speech_server._validate_tts_request(req)

    def test_validate_tts_request_auto_infer_base(self, speech_server):
        """Test auto-inference of Base task when ref_audio/ref_text is provided."""
        # ref_audio without task_type -> infers Base, requires non-empty ref_text
        req = OpenAICreateSpeechRequest(input="Hello", ref_audio="data:audio/wav;base64,abc")
        result = speech_server._validate_tts_request(req)
        assert "ref_text" in result
        assert req.task_type == "Base"

        # ref_text without task_type -> infers Base, requires ref_audio
        req = OpenAICreateSpeechRequest(input="Hello", ref_text="some text")
        result = speech_server._validate_tts_request(req)
        assert "ref_audio" in result
        assert req.task_type == "Base"

    def test_validate_tts_request_base_empty_ref_text(self, speech_server):
        """Empty ref_text on Base task returns 400 instead of crashing engine."""
        req = OpenAICreateSpeechRequest(
            input="Hello", task_type="Base", ref_audio="data:audio/wav;base64,abc", ref_text=""
        )
        result = speech_server._validate_tts_request(req)
        assert "non-empty 'ref_text'" in result

        # x_vector_only_mode bypasses ref_text requirement
        req = OpenAICreateSpeechRequest(
            input="Hello", task_type="Base", ref_audio="data:audio/wav;base64,abc", ref_text="", x_vector_only_mode=True
        )
        assert speech_server._validate_tts_request(req) is None

    def test_validate_tts_request_customvoice_no_speakers(self, speech_server):
        """CustomVoice on a model with no speakers returns 400 instead of crashing engine."""
        req = OpenAICreateSpeechRequest(input="Hello", task_type="CustomVoice")
        result = speech_server._validate_tts_request(req)
        assert "does not support CustomVoice" in result

    # ── speaker_embedding validation ──

    def test_speaker_embedding_valid_base_task(self, speech_server):
        """speaker_embedding with Base task, x_vector_only_mode, and no ref_audio is accepted."""
        emb = [0.1] * 1024
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb, x_vector_only_mode=True)
        assert speech_server._validate_tts_request(req) is None

    def test_speaker_embedding_auto_sets_x_vector_only_mode(self, speech_server):
        """speaker_embedding auto-implies x_vector_only_mode, so validation passes."""
        emb = [0.1] * 1024
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb)
        result = speech_server._validate_tts_request(req)
        assert result is None
        assert req.x_vector_only_mode is True

    def test_speaker_embedding_wrong_task_type(self, speech_server):
        """speaker_embedding is only valid for Base task."""
        emb = [0.1] * 1024
        req = OpenAICreateSpeechRequest(
            input="Hello", task_type="VoiceDesign", speaker_embedding=emb, instructions="warm"
        )
        result = speech_server._validate_tts_request(req)
        assert "only valid for Base task" in result

    def test_speaker_embedding_mutually_exclusive_with_ref_audio(self, speech_server):
        """speaker_embedding and ref_audio cannot both be provided (pydantic validation)."""
        emb = [0.1] * 1024
        with pytest.raises(ValidationError, match="mutually exclusive"):
            OpenAICreateSpeechRequest(
                input="Hello", task_type="Base", speaker_embedding=emb, ref_audio="data:audio/wav;base64,abc"
            )

    def test_speaker_embedding_nan_rejected(self, speech_server):
        """NaN values in speaker_embedding are rejected at parse level."""
        with pytest.raises(ValidationError, match="finite"):
            OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=[0.1] * 1023 + [float("nan")])

    def test_speaker_embedding_inf_rejected(self, speech_server):
        """Inf values in speaker_embedding are rejected at parse level."""
        with pytest.raises(ValidationError, match="finite"):
            OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=[float("inf")] + [0.1] * 1023)

    def test_speaker_embedding_empty_list_rejected(self, speech_server):
        """Empty speaker_embedding list is rejected."""
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=[])
        result = speech_server._validate_tts_request(req)
        assert "non-empty" in result

    def test_speaker_embedding_wrong_dims_accepted(self, speech_server):
        """Non-standard dimensions pass validation (warning only, not an error)."""
        emb = [0.1] * 512  # not 1024 or 2048
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb, x_vector_only_mode=True)
        result = speech_server._validate_tts_request(req)
        assert result is None

    def test_speaker_embedding_2048_dims_accepted(self, speech_server):
        """2048-dim embedding (1.7B model) is accepted without warning."""
        emb = [0.1] * 2048
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb, x_vector_only_mode=True)
        assert speech_server._validate_tts_request(req) is None

    def test_base_task_requires_ref_audio_or_speaker_embedding(self, speech_server):
        """Base task without ref_audio or speaker_embedding is rejected."""
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base")
        result = speech_server._validate_tts_request(req)
        assert "ref_audio" in result and "speaker_embedding" in result

    # ── speaker_embedding in _build_tts_params ──

    def test_build_tts_params_with_speaker_embedding(self, speech_server):
        """speaker_embedding produces voice_clone_prompt and x_vector_only_mode."""
        emb = [0.1] * 1024
        req = OpenAICreateSpeechRequest(input="Hello", task_type="Base", speaker_embedding=emb)
        params = speech_server._build_tts_params(req)

        assert "voice_clone_prompt" in params
        vcp = params["voice_clone_prompt"][0]
        assert "ref_spk_embedding" in vcp
        # Stored as plain list (not tensor) so it survives msgspec IPC serialization
        assert isinstance(vcp["ref_spk_embedding"], list)
        assert len(vcp["ref_spk_embedding"]) == 1024
        assert params["x_vector_only_mode"] == [True]

    def test_build_tts_params_without_speaker_embedding(self, speech_server):
        """Without speaker_embedding, voice_clone_prompt is not set."""
        req = OpenAICreateSpeechRequest(input="Hello", voice="Ryan", language="English")
        params = speech_server._build_tts_params(req)
        assert "voice_clone_prompt" not in params

    def test_build_tts_params(self, speech_server):
        """Test TTS parameter building."""
        req = OpenAICreateSpeechRequest(input="Hello", voice="Ryan", language="English")
        params = speech_server._build_tts_params(req)

        assert params["text"] == ["Hello"]
        assert params["speaker"] == ["Ryan"]
        assert params["language"] == ["English"]
        assert params["task_type"] == ["CustomVoice"]

    def test_load_supported_speakers(self, mocker: MockerFixture):
        """Test _load_supported_speakers."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        mock_engine_client.stage_configs = []

        # Mock talker_config with mixed-case speaker names
        mock_talker_config = mocker.MagicMock()
        mock_talker_config.spk_id = {"Ryan": 0, "Vivian": 1, "Aiden": 2}
        mock_engine_client.model_config.hf_config.talker_config = mock_talker_config

        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )

        # Verify speakers are normalized to lowercase
        assert server.supported_speakers == {"ryan", "vivian", "aiden"}

    def test_build_tts_params_with_uploaded_voice(self, speech_server):
        """Test _build_tts_params auto-sets ref_audio for uploaded voices (x_vector only)."""
        speech_server.uploaded_speakers = {
            "custom_voice": {
                "name": "custom_voice",
                "file_path": "/tmp/voice_samples/custom_voice_consent_123.wav",
                "mime_type": "audio/wav",
                "ref_text": None,
            }
        }
        speech_server.supported_speakers = {"ryan", "vivian", "custom_voice"}

        with patch.object(speech_server, "_get_uploaded_audio_data") as mock_get_audio:
            mock_get_audio.return_value = "data:audio/wav;base64,ZmFrZWF1ZGlv"
            req = OpenAICreateSpeechRequest(input="Hello", voice="custom_voice")
            params = speech_server._build_tts_params(req)

            assert params["ref_audio"] == ["data:audio/wav;base64,ZmFrZWF1ZGlv"]
            assert params["x_vector_only_mode"] == [True]
            assert params["task_type"] == ["Base"]
            assert "ref_text" not in params

    def test_build_tts_params_with_uploaded_voice_ref_text(self, speech_server):
        """Test _build_tts_params enables in-context cloning when ref_text is stored."""
        speech_server.uploaded_speakers = {
            "custom_voice": {
                "name": "custom_voice",
                "file_path": "/tmp/voice_samples/custom_voice_consent_123.wav",
                "mime_type": "audio/wav",
                "ref_text": "Hello world transcript",
            }
        }
        speech_server.supported_speakers = {"ryan", "vivian", "custom_voice"}

        with patch.object(speech_server, "_get_uploaded_audio_data") as mock_get_audio:
            mock_get_audio.return_value = "data:audio/wav;base64,ZmFrZWF1ZGlv"
            req = OpenAICreateSpeechRequest(input="Hello", voice="custom_voice")
            params = speech_server._build_tts_params(req)

            assert params["ref_audio"] == ["data:audio/wav;base64,ZmFrZWF1ZGlv"]
            assert params["x_vector_only_mode"] == [False]
            assert params["task_type"] == ["Base"]
            assert params["ref_text"] == ["Hello world transcript"]

    def test_build_tts_params_without_uploaded_voice(self, speech_server):
        """Test _build_tts_params does not auto-set ref_audio for non-uploaded voices."""
        # No uploaded speakers
        speech_server.uploaded_speakers = {}
        speech_server.supported_speakers = {"ryan", "vivian"}

        req = OpenAICreateSpeechRequest(input="Hello", voice="ryan", task_type="Base")

        params = speech_server._build_tts_params(req)

        # Verify ref_audio was NOT auto-set
        assert "ref_audio" not in params
        assert "x_vector_only_mode" not in params

    def test_build_tts_params_with_explicit_ref_audio(self, speech_server):
        """Test _build_tts_params uses explicit ref_audio even for uploaded voices."""
        # Mock an uploaded speaker
        speech_server.uploaded_speakers = {
            "custom_voice": {
                "name": "custom_voice",
                "file_path": "/tmp/voice_samples/custom_voice_consent_123.wav",
                "mime_type": "audio/wav",
            }
        }
        speech_server.supported_speakers = {"ryan", "vivian", "custom_voice"}

        req = OpenAICreateSpeechRequest(
            input="Hello", voice="custom_voice", task_type="Base", ref_audio="data:audio/wav;base64,ZXhwbGljaXQ="
        )

        params = speech_server._build_tts_params(req)

        # _build_tts_params should NOT auto-set ref_audio when explicit ref_audio
        # is provided (request.ref_audio is not None skips the auto-set branch).
        # The explicit ref_audio is resolved later in create_speech() via
        # _resolve_ref_audio(), not in _build_tts_params().
        assert "ref_audio" not in params
        # x_vector_only_mode should not be set when explicit ref_audio is provided
        assert "x_vector_only_mode" not in params

    def test_get_uploaded_audio_data(self, speech_server):
        """Test _get_uploaded_audio_data function."""
        # Mock file operations
        with (
            patch("builtins.open", create=True) as mock_open,
            patch("base64.b64encode") as mock_b64encode,
            patch("pathlib.Path.exists") as mock_exists,
        ):
            mock_exists.return_value = True
            mock_b64encode.return_value = b"ZmFrZWF1ZGlv"

            # Setup mock file
            mock_file = MagicMock()
            mock_file.read.return_value = b"fakeaudio"
            mock_open.return_value.__enter__.return_value = mock_file

            # Setup uploaded speaker
            speech_server.uploaded_speakers = {
                "test_voice": {"name": "test_voice", "file_path": "/tmp/test.wav", "mime_type": "audio/wav"}
            }
            result = speech_server._get_uploaded_audio_data("test_voice")

            assert result == "data:audio/wav;base64,ZmFrZWF1ZGlv"
            mock_open.assert_called_once_with(Path("/tmp/test.wav"), "rb")
            mock_b64encode.assert_called_once_with(b"fakeaudio")

    def test_get_uploaded_audio_data_missing_file(self, speech_server):
        """Test _get_uploaded_audio_data when file is missing."""
        with patch("pathlib.Path.exists") as mock_exists:
            mock_exists.return_value = False

            # Setup uploaded speaker
            speech_server.uploaded_speakers = {
                "test_voice": {"name": "test_voice", "file_path": "/tmp/test.wav", "mime_type": "audio/wav"}
            }

            result = speech_server._get_uploaded_audio_data("test_voice")

            assert result is None

    def test_get_uploaded_audio_data_voice_not_found(self, speech_server):
        """Test _get_uploaded_audio_data when voice is not in uploaded_speakers."""
        speech_server.uploaded_speakers = {}

        result = speech_server._get_uploaded_audio_data("nonexistent")

        assert result is None

    def test_max_instructions_length_default(self, speech_server):
        """Test default max instructions length (500) when no config provided."""
        # Fixture creates server with no CLI override and no TTS stage
        assert speech_server._max_instructions_length == 500

    def test_max_instructions_length_cli_override(self, mocker: MockerFixture):
        """Test CLI override (stored in engine_client) takes highest priority."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        mock_engine_client.stage_configs = []
        # CLI override is stored in engine_client
        mock_engine_client.tts_max_instructions_length = 1000
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )
        # Value is cached during __init__
        assert server._max_instructions_length == 1000

    def test_max_instructions_length_stage_config(self, mocker: MockerFixture):
        """Test stage config value is used when no CLI override."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        mock_engine_client.tts_max_instructions_length = None  # No CLI override
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        # Mock stage with tts_args
        mock_stage = mocker.MagicMock()
        mock_stage.engine_args.model_stage = "qwen3_tts"
        mock_stage.tts_args = {"max_instructions_length": 750}
        mock_engine_client.stage_configs = [mock_stage]

        server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )
        # Value is cached during __init__
        assert server._max_instructions_length == 750

    def test_max_instructions_length_cli_overrides_stage_config(self, mocker: MockerFixture):
        """Test CLI override (in engine_client) takes precedence over stage config."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        # CLI override stored in engine_client
        mock_engine_client.tts_max_instructions_length = 2000
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        # Mock stage with tts_args
        mock_stage = mocker.MagicMock()
        mock_stage.engine_args.model_stage = "qwen3_tts"
        mock_stage.tts_args = {"max_instructions_length": 750}
        mock_engine_client.stage_configs = [mock_stage]

        server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )
        # CLI value (2000) should override stage config (750)
        assert server._max_instructions_length == 2000

    def test_validate_instructions_length_uses_cached_value(self, mocker: MockerFixture):
        """Test instructions length validation uses cached _max_instructions_length."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False
        mock_engine_client.stage_configs = []
        # CLI override with max length of 10 characters
        mock_engine_client.tts_max_instructions_length = 10
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )

        # Verify cached value
        assert server._max_instructions_length == 10

        # Instructions within limit should pass
        req = OpenAICreateSpeechRequest(
            input="Hello",
            task_type="VoiceDesign",
            instructions="short",
        )
        assert server._validate_tts_request(req) is None

        # Instructions exceeding limit should fail
        req = OpenAICreateSpeechRequest(
            input="Hello",
            task_type="VoiceDesign",
            instructions="this is too long",
        )
        error = server._validate_tts_request(req)
        assert error is not None
        assert "max 10 characters" in error


class TestFileValidationFunctions:
    """Unit tests for file validation helper functions."""

    def test_sanitize_filename(self):
        """Test _sanitize_filename function."""
        from vllm_omni.entrypoints.openai.serving_speech import _sanitize_filename

        # Test normal filenames
        assert _sanitize_filename("test.wav") == "test.wav"
        assert _sanitize_filename("test-file.mp3") == "test-file.mp3"
        assert _sanitize_filename("test_file.flac") == "test_file.flac"

        # Test path traversal attempts
        assert _sanitize_filename("../../../etc/passwd") == "passwd"
        assert _sanitize_filename("/absolute/path/file.wav") == "file.wav"

        # Test special characters
        assert _sanitize_filename("file with spaces.wav") == "file_with_spaces.wav"
        assert _sanitize_filename("file&with&special&chars.wav") == "file_with_special_chars.wav"
        assert _sanitize_filename("file@with#special$chars%.wav") == "file_with_special_chars_.wav"

        # Test empty filename
        assert _sanitize_filename("") == "file"

        # Test very long filename
        long_name = "a" * 300
        sanitized = _sanitize_filename(long_name)
        assert len(sanitized) == 255
        assert sanitized.startswith("a")

    def test_validate_path_within_directory(self, tmp_path):
        """Test _validate_path_within_directory function."""
        from vllm_omni.entrypoints.openai.serving_speech import _validate_path_within_directory

        # Create test directory structure
        base_dir = tmp_path / "uploads"
        base_dir.mkdir()

        # Valid paths within directory
        valid_file = base_dir / "test.wav"
        valid_subdir_file = base_dir / "subdir" / "test.wav"
        valid_subdir_file.parent.mkdir()

        assert _validate_path_within_directory(valid_file, base_dir) is True
        assert _validate_path_within_directory(valid_subdir_file, base_dir) is True

        # Invalid paths outside directory
        outside_file = tmp_path / "outside.wav"
        assert _validate_path_within_directory(outside_file, base_dir) is False

        # Test with symlink (should fail)
        if hasattr(os, "symlink"):
            link_target = tmp_path / "target.wav"
            link_target.touch()
            symlink = base_dir / "link.wav"
            os.symlink(link_target, symlink)
            # Symlinks to outside should be rejected
            assert _validate_path_within_directory(symlink, base_dir) is False

        # Test with non-existent file (should still validate path)
        non_existent = base_dir / "nonexistent.wav"
        assert _validate_path_within_directory(non_existent, base_dir) is True


class TestStreamingProtocolValidation:
    """Unit tests for the stream field validators in OpenAICreateSpeechRequest."""

    def test_stream_validation_errors(self):
        """stream=True requires response_format not in ('pcm', 'wav') and speed=1.0."""
        with pytest.raises(ValidationError, match="requires response_format='pcm' or 'wav'"):
            OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="mp3")
        with pytest.raises(ValidationError, match="Speed adjustment is not supported"):
            OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="pcm", speed=2.0)

    def test_stream_valid(self):
        """stream=True + response_format in ('pcm', 'wav') + speed=1.0 is accepted."""
        req = OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="pcm")
        assert req.stream is True

        req = OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="wav")
        assert req.stream is True

    def test_sse_stream_format_is_blocked(self):
        """stream_format='sse' is blocked."""
        with pytest.raises(ValidationError, match="sse"):
            OpenAICreateSpeechRequest(input="Hello", stream_format="sse")


class TestStreamingResponse:
    """Integration tests for the streaming audio response path."""

    @pytest.fixture
    def streaming_app(self, mocker: MockerFixture):
        """Test app whose mock engine yields one intermediate chunk then a final chunk."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False

        def _make_output(finished: bool) -> OmniRequestOutput:
            chunk = torch.sin(torch.linspace(0, 440 * 2 * torch.pi, 24000))

            class MockCompletionOutput:
                def __init__(self, index: int = 0):
                    self.index = index
                    self.text = ""
                    self.token_ids = []
                    self.finish_reason = "stop"
                    self.stop_reason = None
                    self.logprobs = None

            class MockRequestOutput:
                def __init__(self, audio_tensor: torch.Tensor):
                    self.request_id = "speech-stream-test"
                    self.outputs = [MockCompletionOutput(index=0)]
                    self.multimodal_output = {"audio": audio_tensor}
                    self.finished = finished
                    self.prompt_token_ids = None
                    self.encoder_prompt_token_ids = None
                    self.num_cached_tokens = None
                    self.prompt_logprobs = None
                    self.kv_transfer_params = None

            return OmniRequestOutput(
                stage_id=0,
                final_output_type="audio",
                request_output=MockRequestOutput(audio_tensor=chunk),
                finished=finished,
            )

        async def mock_generate_streaming(*args, **kwargs):
            yield _make_output(finished=False)
            yield _make_output(finished=True)

        mock_engine_client.generate = mocker.MagicMock(side_effect=mock_generate_streaming)
        mock_engine_client.default_sampling_params_list = [{}]
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        speech_server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )

        original_create_speech = speech_server.create_speech
        sig = signature(original_create_speech)
        new_parameters = [p for name, p in sig.parameters.items() if name != "raw_request"]
        new_sig = Signature(parameters=new_parameters, return_annotation=sig.return_annotation)

        async def awaitable_create_speech(*args, **kwargs):
            return await original_create_speech(*args, **kwargs)

        awaitable_create_speech.__signature__ = new_sig
        speech_server.create_speech = awaitable_create_speech

        app = FastAPI()
        app.add_api_route("/v1/audio/speech", speech_server.create_speech, methods=["POST"], response_model=None)
        return app

    def test_streaming(self, streaming_app):
        """stream=True must return audio/pcm with non-empty body."""
        client = TestClient(streaming_app)
        response = client.post("/v1/audio/speech", json={"input": "Hello", "stream": True, "response_format": "pcm"})
        assert response.status_code == 200
        assert "audio/pcm" in response.headers["content-type"]
        assert len(response.content) > 0

    def test_non_streaming_unchanged(self, streaming_app):
        """Non-streaming path must still return audio/wav."""
        client = TestClient(streaming_app)
        response = client.post("/v1/audio/speech", json={"input": "Hello", "response_format": "wav"})
        assert response.status_code == 200
        assert "audio/wav" in response.headers["content-type"]


class TestSpeechBatchAPI:
    """Tests for the /v1/audio/speech/batch endpoint."""

    def test_batch_success(self, client):
        """Batch with two items should return two successful results with base64 audio."""
        payload = {
            "items": [
                {"input": "Hello world"},
                {"input": "Goodbye world"},
            ],
            "response_format": "wav",
        }
        response = client.post("/v1/audio/speech/batch", json=payload)
        assert response.status_code == 200
        body = response.json()
        assert body["total"] == 2
        assert body["succeeded"] == 2
        assert body["failed"] == 0
        assert all(r["status"] == "success" for r in body["results"])
        assert all(r["audio_data"] is not None for r in body["results"])
        # Verify audio_data is valid base64
        import base64

        for r in body["results"]:
            decoded = base64.b64decode(r["audio_data"])
            assert len(decoded) > 0

    def test_batch_single_item(self, client):
        """Batch with a single item should work."""
        payload = {"items": [{"input": "Solo"}]}
        response = client.post("/v1/audio/speech/batch", json=payload)
        assert response.status_code == 200
        body = response.json()
        assert body["total"] == 1
        assert body["succeeded"] == 1

    def test_batch_empty_items_rejected(self, client):
        """Empty items list should be rejected by Pydantic validation."""
        response = client.post("/v1/audio/speech/batch", json={"items": []})
        assert response.status_code == 422

    def test_batch_too_many_items(self, client):
        """Exceeding the batch max items limit (default 32) should be rejected."""
        payload = {"items": [{"input": f"text {i}"} for i in range(33)]}
        with pytest.raises(ValueError, match="exceeding the maximum"):
            client.post("/v1/audio/speech/batch", json=payload)

    def test_batch_max_items_allowed(self, client):
        """Exactly 32 items should be accepted."""
        payload = {"items": [{"input": f"text {i}"} for i in range(32)]}
        response = client.post("/v1/audio/speech/batch", json=payload)
        assert response.status_code == 200
        body = response.json()
        assert body["total"] == 32
        assert body["succeeded"] == 32

    def test_batch_results_have_correct_indices(self, client):
        """Each result should have an index matching its position."""
        payload = {"items": [{"input": f"text {i}"} for i in range(3)]}
        response = client.post("/v1/audio/speech/batch", json=payload)
        body = response.json()
        indices = [r["index"] for r in body["results"]]
        assert indices == [0, 1, 2]

    def test_batch_response_has_id(self, client):
        """Batch response should have a unique id starting with 'speech-batch-'."""
        payload = {"items": [{"input": "Hello"}]}
        response = client.post("/v1/audio/speech/batch", json=payload)
        body = response.json()
        assert body["id"].startswith("speech-batch-")


class TestMergeBatchItem:
    """Tests for the _merge_batch_item static method."""

    def test_item_override_wins(self):
        """Per-item voice should override batch-level voice."""
        batch = BatchSpeechRequest(
            items=[SpeechBatchItem(input="hi", voice="Ryan")],
            voice="Vivian",
        )
        merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0])
        assert merged.voice == "Ryan"

    def test_batch_default_used(self):
        """Batch-level voice should be used when item doesn't specify one."""
        batch = BatchSpeechRequest(
            items=[SpeechBatchItem(input="hi")],
            voice="Vivian",
        )
        merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0])
        assert merged.voice == "Vivian"

    def test_response_format_override(self):
        """Per-item response_format should override batch default."""
        batch = BatchSpeechRequest(
            items=[SpeechBatchItem(input="hi", response_format="mp3")],
            response_format="wav",
        )
        merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0])
        assert merged.response_format == "mp3"

    def test_stream_always_false(self):
        """Merged requests should always have stream=False."""
        batch = BatchSpeechRequest(items=[SpeechBatchItem(input="hi")])
        merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0])
        assert merged.stream is False

    def test_all_fields_merge(self):
        """All overridable fields should merge correctly."""
        batch = BatchSpeechRequest(
            items=[
                SpeechBatchItem(
                    input="hello",
                    voice="Ryan",
                    language="English",
                    speed=1.5,
                    task_type="CustomVoice",
                    max_new_tokens=512,
                )
            ],
            voice="Vivian",
            language="Chinese",
            speed=1.0,
        )
        merged = OmniOpenAIServingSpeech._merge_batch_item(batch, batch.items[0])
        assert merged.voice == "Ryan"
        assert merged.language == "English"
        assert merged.speed == 1.5
        assert merged.task_type == "CustomVoice"
        assert merged.max_new_tokens == 512


class TestAsyncOmniSupportedTasks:
    """Test that AsyncOmni reports correct supported tasks based on output modalities."""

    @pytest.mark.asyncio
    async def test_tts_only_no_generate_task(self):
        """TTS-only models (audio output, no text) should not include 'generate'."""
        from types import SimpleNamespace

        from vllm_omni.entrypoints.async_omni import AsyncOmni

        omni = AsyncOmni.__new__(AsyncOmni)
        omni.engine = SimpleNamespace(supported_tasks=("speech",))
        tasks = await omni.get_supported_tasks()
        assert "generate" not in tasks
        assert "speech" in tasks

    @pytest.mark.asyncio
    async def test_omni_model_includes_generate(self):
        """Models with text output (e.g. Qwen3-Omni) should include 'generate'."""
        from types import SimpleNamespace

        from vllm_omni.entrypoints.async_omni import AsyncOmni

        omni = AsyncOmni.__new__(AsyncOmni)
        omni.engine = SimpleNamespace(supported_tasks=("generate", "speech"))
        tasks = await omni.get_supported_tasks()
        assert "generate" in tasks


def test_api_server_create_speech_wraps_error_response_status():
    handler = MagicMock()
    handler.create_speech = AsyncMock(
        return_value=ErrorResponse(
            error=ErrorInfo(message="bad request", type="BadRequestError", param=None, code=400),
        )
    )

    app = FastAPI()
    app.state.openai_serving_speech = handler
    scope = {
        "type": "http",
        "app": app,
        "method": "POST",
        "path": "/v1/audio/speech",
        "headers": [],
        "query_string": b"",
        "client": ("127.0.0.1", 12345),
        "server": ("testserver", 80),
        "scheme": "http",
    }
    raw_request = Request(scope)
    request = OpenAICreateSpeechRequest(input="Hello")

    response = asyncio.run(api_server_module.create_speech(request, raw_request))

    assert isinstance(response, JSONResponse)
    assert response.status_code == 400


class TestWAVHeaderGeneration:
    """Unit tests for WAV header generation with placeholder values."""

    def test_wav_header_basic_structure(self):
        """Test basic WAV header structure with default parameters."""
        header = _create_wav_header(sample_rate=24000, num_channels=1, bits_per_sample=16)

        # Verify header length (should be 44 bytes)
        assert len(header) == 44, f"Expected 44 bytes, got {len(header)}"

        # Parse and verify header structure
        (
            chunk_id,
            chunk_size,
            format_type,
            subchunk1_id,
            subchunk1_size,
            audio_format,
            num_channels,
            sample_rate,
            byte_rate,
            block_align,
            bits_per_sample,
            subchunk2_id,
            subchunk2_size,
        ) = struct.unpack("<4sI4s4sIHHIIHH4sI", header)

        # Verify RIFF header
        assert chunk_id == b"RIFF", f"Expected RIFF, got {chunk_id}"
        assert chunk_size == 0xFFFFFFFF, f"Expected placeholder 0xFFFFFFFF, got {chunk_size:#x}"
        assert format_type == b"WAVE", f"Expected WAVE, got {format_type}"

        # Verify fmt chunk
        assert subchunk1_id == b"fmt ", f"Expected 'fmt ', got {subchunk1_id}"
        assert subchunk1_size == 16, f"Expected 16, got {subchunk1_size}"
        assert audio_format == 1, f"Expected PCM (1), got {audio_format}"
        assert num_channels == 1, f"Expected 1 channel, got {num_channels}"
        assert sample_rate == 24000, f"Expected 24000 Hz, got {sample_rate}"
        assert byte_rate == 48000, f"Expected 48000 byte/s, got {byte_rate}"
        assert block_align == 2, f"Expected 2 bytes block align, got {block_align}"
        assert bits_per_sample == 16, f"Expected 16 bits, got {bits_per_sample}"

        # Verify data chunk
        assert subchunk2_id == b"data", f"Expected 'data', got {subchunk2_id}"
        assert subchunk2_size == 0xFFFFFFFF, f"Expected placeholder 0xFFFFFFFF, got {subchunk2_size:#x}"

    def test_wav_header_different_sample_rates(self):
        """Test WAV header with different sample rates."""
        test_cases = [
            (16000, 1, 16),
            (22050, 1, 16),
            (24000, 1, 16),
            (44100, 1, 16),
            (48000, 1, 16),
        ]

        for sample_rate, num_channels, bits_per_sample in test_cases:
            header = _create_wav_header(sample_rate, num_channels, bits_per_sample)
            assert len(header) == 44, f"Header length mismatch for {sample_rate} Hz"

            # Parse sample rate from header
            parsed_sample_rate = struct.unpack("<I", header[24:28])[0]
            assert parsed_sample_rate == sample_rate, (
                f"Sample rate mismatch: expected {sample_rate}, got {parsed_sample_rate}"
            )

    def test_wav_header_stereo(self):
        """Test WAV header with stereo audio."""
        header = _create_wav_header(sample_rate=44100, num_channels=2, bits_per_sample=16)

        # Parse header
        parsed = struct.unpack("<4sI4s4sIHHIIHH4sI", header)
        num_channels = parsed[6]
        byte_rate = parsed[8]
        block_align = parsed[9]

        assert num_channels == 2, f"Expected 2 channels, got {num_channels}"
        assert byte_rate == 44100 * 2 * 16 // 8, "Byte rate mismatch"
        assert block_align == 2 * 16 // 8, "Block align mismatch"

    def test_wav_header_placeholder_values(self):
        """Test that placeholder values are correctly set to 0xFFFFFFFF."""
        header = _create_wav_header(sample_rate=24000)

        # Extract size fields
        chunk_size = struct.unpack("<I", header[4:8])[0]
        subchunk2_size = struct.unpack("<I", header[40:44])[0]

        assert chunk_size == 0xFFFFFFFF, "ChunkSize should be 0xFFFFFFFF for streaming"
        assert subchunk2_size == 0xFFFFFFFF, "Subchunk2Size should be 0xFFFFFFFF for streaming"


class TestWAVStreaming:
    """Integration tests for WAV format streaming."""

    @pytest.fixture
    def wav_streaming_app(self, mocker: MockerFixture):
        """Test app configured for WAV streaming."""
        mock_engine_client = mocker.MagicMock()
        mock_engine_client.errored = False

        def _make_output(finished: bool) -> OmniRequestOutput:
            chunk = torch.sin(torch.linspace(0, 440 * 2 * torch.pi, 24000))

            class MockCompletionOutput:
                def __init__(self, index: int = 0):
                    self.index = index
                    self.text = ""
                    self.token_ids = []
                    self.finish_reason = "stop"
                    self.stop_reason = None
                    self.logprobs = None

            class MockRequestOutput:
                def __init__(self, audio_tensor: torch.Tensor):
                    self.request_id = "speech-wav-stream-test"
                    self.outputs = [MockCompletionOutput(index=0)]
                    self.multimodal_output = {"audio": audio_tensor, "sr": 24000}
                    self.finished = finished
                    self.prompt_token_ids = None
                    self.encoder_prompt_token_ids = None
                    self.num_cached_tokens = None
                    self.prompt_logprobs = None
                    self.kv_transfer_params = None

            return OmniRequestOutput(
                stage_id=0,
                final_output_type="audio",
                request_output=MockRequestOutput(audio_tensor=chunk),
                finished=finished,
            )

        async def mock_generate_streaming(*args, **kwargs):
            yield _make_output(finished=False)
            yield _make_output(finished=True)

        mock_engine_client.generate = mocker.MagicMock(side_effect=mock_generate_streaming)
        mock_engine_client.default_sampling_params_list = [{}]
        mock_models = mocker.MagicMock()
        mock_models.is_base_model.return_value = True

        speech_server = OmniOpenAIServingSpeech(
            engine_client=mock_engine_client,
            models=mock_models,
            request_logger=mocker.MagicMock(),
        )

        original_create_speech = speech_server.create_speech
        sig = signature(original_create_speech)
        new_parameters = [p for name, p in sig.parameters.items() if name != "raw_request"]
        new_sig = Signature(parameters=new_parameters, return_annotation=sig.return_annotation)

        async def awaitable_create_speech(*args, **kwargs):
            return await original_create_speech(*args, **kwargs)

        awaitable_create_speech.__signature__ = new_sig
        speech_server.create_speech = awaitable_create_speech

        app = FastAPI()
        app.add_api_route("/v1/audio/speech", speech_server.create_speech, methods=["POST"], response_model=None)
        return app

    def test_wav_streaming_success(self, wav_streaming_app):
        """Test WAV format streaming returns correct content type and includes WAV header."""
        client = TestClient(wav_streaming_app)
        response = client.post("/v1/audio/speech", json={"input": "Hello", "stream": True, "response_format": "wav"})

        assert response.status_code == 200
        assert "audio/wav" in response.headers["content-type"]
        assert len(response.content) > 44  # Should have WAV header + audio data

        # Verify WAV header is present
        header = response.content[:44]
        chunk_id = header[0:4]
        format_type = header[8:12]
        assert chunk_id == b"RIFF", "Should start with RIFF"
        assert format_type == b"WAVE", "Should contain WAVE format"

    def test_streaming_unsupported_format_rejected(self, wav_streaming_app):
        """Test that unsupported formats are rejected for streaming."""
        client = TestClient(wav_streaming_app)

        unsupported_formats = ["mp3"]
        for fmt in unsupported_formats:
            response = client.post("/v1/audio/speech", json={"input": "Hello", "stream": True, "response_format": fmt})
            assert response.status_code == 422