xiaoxia-saas/tests/unit/test_voice_clone_preview.py

"""音色克隆试听接口单元测试 (#1195).

直接测试路由函数逻辑，mock 所有依赖。
"""

from __future__ import annotations

import time
from datetime import datetime, timezone
from unittest.mock import MagicMock

import pytest
from fastapi import HTTPException

from packages.application.cosyvoice_service import CosyVoiceError, SynthesizeResult
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus


def _make_profile(**kwargs) -> VoiceCloneProfile:
    defaults = {
        "id": "clone_001",
        "user_id": "user_001",
        "name": "测试音色",
        "description": "",
        "source_audio_url": "",
        "voice_id": "clone_voice_001",
        "voice_model": "cosyvoice-v3",
        "language": "zh-CN",
        "gender": "female",
        "status": VoiceCloneStatus.READY,
        "error_message": "",
        "retry_count": 0,
        "max_retries": 3,
        "metadata": {},
        "created_at": datetime.now(timezone.utc),
        "updated_at": datetime.now(timezone.utc),
    }
    defaults.update(kwargs)
    return VoiceCloneProfile(**defaults)


def _make_auth_user(user_id: str = "user_001") -> MagicMock:
    user = MagicMock()
    user.id = user_id
    auth = MagicMock()
    auth.user = user
    return auth


class TestVoiceClonePreview:
    """克隆音色试听接口测试。"""

    def _call_preview(self, profile, cosyvoice_mock, text="", user_id="user_001"):
        """调用路由函数，模拟 FastAPI 注入依赖。"""
        from app.api.routes.voice_clones import get_voice_clone_preview

        repo = MagicMock()
        if profile is None:
            from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError

            repo.get.return_value = None
            use_case_instance = MagicMock()
            use_case_instance.execute.side_effect = VoiceCloneNotFoundError("not found")
        else:
            repo.get.return_value = profile
            use_case_instance = MagicMock()
            use_case_instance.execute.return_value = profile

        # 替换 use case 构造
        from packages.application.voice_clone.use_cases import GetVoiceCloneUseCase

        original_init = GetVoiceCloneUseCase.__init__
        original_execute = GetVoiceCloneUseCase.execute

        def mock_init(self, repository):
            self.repository = repository

        def mock_execute(self, clone_id, uid):
            if profile is None:
                from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError

                raise VoiceCloneNotFoundError("not found")
            if profile.user_id != uid:
                from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError

                raise VoiceCloneNotFoundError("not found")
            return profile

        GetVoiceCloneUseCase.__init__ = mock_init
        GetVoiceCloneUseCase.execute = mock_execute

        try:
            result = get_voice_clone_preview(
                clone_id=profile.id if profile else "nonexistent",
                text=text,
                authenticated_user=_make_auth_user(user_id),
                repository=repo,
                cosyvoice=cosyvoice_mock,
            )
            return result
        finally:
            GetVoiceCloneUseCase.__init__ = original_init
            GetVoiceCloneUseCase.execute = original_execute

    def test_preview_success_ready_clone(self) -> None:
        """就绪的克隆音色可以正常试听。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile()
        cosyvoice = MagicMock()
        cosyvoice.synthesize_speech.return_value = SynthesizeResult(
            audio_url="https://oss.example.com/preview/clone_001.mp3",
            duration=3.5,
            file_size=56000,
            request_id="req_001",
        )

        result = self._call_preview(profile, cosyvoice)

        assert result.clone_id == "clone_001"
        assert result.voice_id == "clone_voice_001"
        assert result.audio_url == "https://oss.example.com/preview/clone_001.mp3"
        assert result.duration == 3.5
        assert result.file_size == 56000
        assert "克隆音色" in result.text
        cosyvoice.synthesize_speech.assert_called_once()
        call_kwargs = cosyvoice.synthesize_speech.call_args
        assert call_kwargs.kwargs["voice_id"] == "clone_voice_001"
        assert call_kwargs.kwargs["format"] == "mp3"

    def test_preview_custom_text(self) -> None:
        """自定义试听文本。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile()
        cosyvoice = MagicMock()
        cosyvoice.synthesize_speech.return_value = SynthesizeResult(
            audio_url="https://oss.example.com/preview/custom.mp3",
            duration=2.0,
            file_size=32000,
        )

        custom_text = "大家好，这是自定义试听文本。"
        result = self._call_preview(profile, cosyvoice, text=custom_text)

        assert result.text == custom_text
        call_kwargs = cosyvoice.synthesize_speech.call_args
        assert call_kwargs.kwargs["text"] == custom_text

    def test_preview_not_found(self) -> None:
        """克隆音色不存在时返回 404。"""
        cosyvoice = MagicMock()

        with pytest.raises(HTTPException) as exc_info:
            self._call_preview(None, cosyvoice)

        assert exc_info.value.status_code == 404
        cosyvoice.synthesize_speech.assert_not_called()

    def test_preview_not_ready_pending(self) -> None:
        """pending 状态的克隆音色不能试听。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile(status=VoiceCloneStatus.PENDING, voice_id="")
        cosyvoice = MagicMock()

        with pytest.raises(HTTPException) as exc_info:
            self._call_preview(profile, cosyvoice)

        assert exc_info.value.status_code == 400
        assert "not ready" in exc_info.value.detail.lower() or "未就绪" in exc_info.value.detail
        cosyvoice.synthesize_speech.assert_not_called()

    def test_preview_not_ready_processing(self) -> None:
        """processing 状态的克隆音色不能试听。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
        cosyvoice = MagicMock()

        with pytest.raises(HTTPException) as exc_info:
            self._call_preview(profile, cosyvoice)

        assert exc_info.value.status_code == 400
        cosyvoice.synthesize_speech.assert_not_called()

    def test_preview_not_ready_failed(self) -> None:
        """failed 状态的克隆音色不能试听。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile(status=VoiceCloneStatus.FAILED, error_message="超时")
        cosyvoice = MagicMock()

        with pytest.raises(HTTPException) as exc_info:
            self._call_preview(profile, cosyvoice)

        assert exc_info.value.status_code == 400
        cosyvoice.synthesize_speech.assert_not_called()

    def test_preview_tts_failure(self) -> None:
        """TTS 合成失败返回 502。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile()
        cosyvoice = MagicMock()
        cosyvoice.synthesize_speech.side_effect = CosyVoiceError("API 调用失败")

        with pytest.raises(HTTPException) as exc_info:
            self._call_preview(profile, cosyvoice)

        assert exc_info.value.status_code == 502
        assert "TTS" in exc_info.value.detail

    def test_preview_cache_default_text(self) -> None:
        """默认试听文本使用缓存。"""
        from app.api.routes.voice_clones import CLONE_PREVIEW_CACHE_TTL, _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile()
        cosyvoice = MagicMock()
        cosyvoice.synthesize_speech.return_value = SynthesizeResult(
            audio_url="https://oss.example.com/preview/cached.mp3",
            duration=3.0,
            file_size=48000,
        )

        # 第一次调用：合成
        result1 = self._call_preview(profile, cosyvoice)
        assert cosyvoice.synthesize_speech.call_count == 1

        # 第二次调用：走缓存，不重复合成
        result2 = self._call_preview(profile, cosyvoice)
        assert cosyvoice.synthesize_speech.call_count == 1
        assert result2.audio_url == result1.audio_url
        assert result2.duration == result1.duration

    def test_preview_custom_text_no_cache(self) -> None:
        """自定义文本不使用缓存。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile()
        cosyvoice = MagicMock()
        cosyvoice.synthesize_speech.side_effect = [
            SynthesizeResult(audio_url="https://example.com/1.mp3", duration=2.0, file_size=32000),
            SynthesizeResult(audio_url="https://example.com/2.mp3", duration=2.5, file_size=40000),
        ]

        result1 = self._call_preview(profile, cosyvoice, text="自定义文本一")
        result2 = self._call_preview(profile, cosyvoice, text="自定义文本二")

        assert cosyvoice.synthesize_speech.call_count == 2
        assert result1.audio_url != result2.audio_url

    def test_preview_wrong_user(self) -> None:
        """非所有者不能访问他人克隆音色的试听。"""
        from app.api.routes.voice_clones import _clone_preview_cache

        _clone_preview_cache.clear()

        profile = _make_profile(user_id="user_001")
        cosyvoice = MagicMock()

        with pytest.raises(HTTPException) as exc_info:
            self._call_preview(profile, cosyvoice, user_id="user_002")

        assert exc_info.value.status_code == 404
        cosyvoice.synthesize_speech.assert_not_called()