xiaoxia-saas/tests/unit/test_tts_preview.py

"""Tests for POST /tts/preview endpoint."""

from __future__ import annotations

from dataclasses import dataclass
from unittest.mock import MagicMock, patch

import pytest
from pydantic import ValidationError


@dataclass
class FakeSynthesizeResult:
    audio_url: str
    duration: float = 0.0
    file_size: int = 0
    request_id: str = ""


class TestTTSPreviewEndpoint:
    """Integration-style tests for the /tts/preview route."""

    def _make_client(self, app):
        from fastapi.testclient import TestClient

        return TestClient(app)

    def test_schema_preview_request_validation(self):
        """TTSPreviewRequest rejects text > 200 chars and empty voice_id."""
        from app.schemas.tts import TTSPreviewRequest

        # Valid
        req = TTSPreviewRequest(text="hello", voice_id="v1")
        assert req.text == "hello"
        assert req.voice_id == "v1"
        assert req.speed == 1.0

        # Empty voice_id rejected
        with pytest.raises(ValidationError):
            TTSPreviewRequest(text="hello", voice_id="")

        # Text > 200 chars rejected
        with pytest.raises(ValidationError):
            TTSPreviewRequest(text="a" * 201, voice_id="v1")

    def test_schema_preview_response(self):
        """TTSPreviewResponse serialization."""
        from app.schemas.tts import TTSPreviewResponse

        resp = TTSPreviewResponse(audio_url="https://example.com/audio.mp3")
        assert resp.audio_url == "https://example.com/audio.mp3"
        assert resp.duration is None

        resp2 = TTSPreviewResponse(audio_url="https://x.com/a.mp3", duration=3.5)
        assert resp2.duration == 3.5

    def test_preview_success(self):
        """Successful preview returns audio_url."""
        from app.schemas.tts import TTSPreviewRequest
        from fastapi import FastAPI

        app = FastAPI()

        # We need to register the route with proper dependencies
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        # Override dependencies
        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
            audio_url="https://cosyvoice.example.com/audio.mp3",
            duration=2.5,
        )
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        from app.dependencies import get_voice_clone_profile_repository

        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None  # no profile found = preset voice
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "你好世界", "voice_id": "longxiaochun"},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["audio_url"] == "https://cosyvoice.example.com/audio.mp3"
        assert data["duration"] == 2.5

        mock_service.synthesize_speech.assert_called_once_with(
            text="你好世界",
            voice_id="longxiaochun",
            speed=1.0,
            emotion="",
            language="zh-CN",
        )

    def test_preview_with_speed(self):
        """Custom speed is passed through to CosyVoice."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
            audio_url="https://x.com/a.mp3",
            duration=0.0,
        )
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        from app.dependencies import get_voice_clone_profile_repository

        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None  # no profile found = preset voice
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "测试", "voice_id": "v1", "speed": 1.5},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["audio_url"] == "https://x.com/a.mp3"
        assert data["duration"] is None  # 0.0 -> None

        mock_service.synthesize_speech.assert_called_once_with(
            text="测试",
            voice_id="v1",
            speed=1.5,
            emotion="",
            language="zh-CN",
        )

    def test_preview_cosyvoice_error_returns_502(self):
        """CosyVoice failure returns 502."""
        from fastapi import FastAPI

        from packages.application.cosyvoice_service import CosyVoiceError

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        mock_service.synthesize_speech.side_effect = CosyVoiceError("API timeout")
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        from app.dependencies import get_voice_clone_profile_repository

        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None  # no profile found = preset voice
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "测试", "voice_id": "v1"},
        )
        assert resp.status_code == 502
        assert "TTS 合成失败" in resp.json()["detail"]

    def test_preview_value_error_returns_400(self):
        """Invalid params return 400."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        mock_service.synthesize_speech.side_effect = ValueError("text 不能为空")
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        from app.dependencies import get_voice_clone_profile_repository

        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None  # no profile found = preset voice
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "测试", "voice_id": "v1"},
        )
        assert resp.status_code == 400
        assert "text 不能为空" in resp.json()["detail"]

    def test_preview_text_too_long_returns_422(self):
        """Text > 200 chars is rejected by Pydantic validation."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        from app.dependencies import get_voice_clone_profile_repository

        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None  # no profile found = preset voice
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "a" * 201, "voice_id": "v1"},
        )
        assert resp.status_code == 422

    def test_preview_empty_voice_id_returns_422(self):
        """Empty voice_id is rejected by Pydantic validation."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        from app.dependencies import get_voice_clone_profile_repository

        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None  # no profile found = preset voice
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "hello", "voice_id": ""},
        )
        assert resp.status_code == 422

    def test_preview_clone_voice_resolves_to_cosyvoice_id(self):
        """Clone voice UUID is resolved to CosyVoice voice_id."""
        from dataclasses import dataclass

        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
            audio_url="https://x.com/cloned.mp3",
            duration=1.8,
        )
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        # Mock voice clone profile with voice_id
        mock_profile = MagicMock()
        mock_profile.user_id = "user-1"
        mock_profile.voice_id = "cosyvoice_actual_voice_123"
        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = mock_profile
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        # Frontend sends the profile UUID as voice_id
        resp = client.post(
            "/tts/preview",
            json={"text": "克隆音色测试", "voice_id": "abc123-uuid-of-profile"},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert data["audio_url"] == "https://x.com/cloned.mp3"

        # Verify CosyVoice was called with the resolved voice_id, not the UUID
        mock_service.synthesize_speech.assert_called_once_with(
            text="克隆音色测试",
            voice_id="cosyvoice_actual_voice_123",
            speed=1.0,
            emotion="",
            language="zh-CN",
        )
        # Verify repo was queried with the UUID
        mock_clone_repo.get.assert_called_once_with("abc123-uuid-of-profile")

    def test_preview_clone_voice_incomplete_returns_400(self):
        """Clone profile with empty voice_id returns 400."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_voice_clone_profile_repository

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        # Mock voice clone profile with empty voice_id (clone not finished)
        mock_profile = MagicMock()
        mock_profile.user_id = "user-1"
        mock_profile.voice_id = ""
        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = mock_profile
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "测试未完成克隆", "voice_id": "abc123-uuid"},
        )
        assert resp.status_code == 400
        assert "音色克隆尚未完成" in resp.json()["detail"]

    def test_preview_preset_voice_passthrough(self):
        """Preset voice ID (not a profile UUID) passes through unchanged."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        mock_service = MagicMock()
        mock_service.synthesize_speech.return_value = FakeSynthesizeResult(
            audio_url="https://x.com/preset.mp3",
            duration=2.0,
        )
        app.dependency_overrides[get_cosyvoice_service] = lambda: mock_service

        # Mock repo returns None (preset voice, not a clone profile)
        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = None
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "预设音色测试", "voice_id": "longxiaoxia_v3"},
        )
        assert resp.status_code == 200

        # Verify CosyVoice was called with the original preset voice_id
        mock_service.synthesize_speech.assert_called_once_with(
            text="预设音色测试",
            voice_id="longxiaoxia_v3",
            speed=1.0,
            emotion="",
            language="zh-CN",
        )

    def test_preview_clone_voice_wrong_user_returns_403(self):
        """Accessing another user's clone profile returns 403."""
        from fastapi import FastAPI

        app = FastAPI()
        from app.api.routes.tts import router

        app.include_router(router, prefix="/tts")

        from app.auth import get_current_user
        from app.dependencies import get_voice_clone_profile_repository

        fake_user = MagicMock()
        fake_user.user.id = "user-1"
        app.dependency_overrides[get_current_user] = lambda: fake_user

        # Mock profile belonging to a different user
        mock_profile = MagicMock()
        mock_profile.user_id = "user-2"
        mock_profile.voice_id = "cosyvoice_voice_xyz"
        mock_clone_repo = MagicMock()
        mock_clone_repo.get.return_value = mock_profile
        app.dependency_overrides[get_voice_clone_profile_repository] = lambda: mock_clone_repo

        client = self._make_client(app)
        resp = client.post(
            "/tts/preview",
            json={"text": "越权测试", "voice_id": "other-user-profile-uuid"},
        )
        assert resp.status_code == 403
        assert "无权访问该音色" in resp.json()["detail"]