xiaoxia-saas/tests/unit/test_preset_voice_dynamic_preview.py

"""Tests for dynamic preview URL resolution in voices routes.

覆盖 _resolve_preset_preview_url 和 _resolve_all_preset_preview_urls 的逻辑：
- 缓存命中时直接返回
- 缓存失效时调用 CosyVoice 合成
- CosyVoice 失败时降级到硬编码 URL
- 并行解析多个预置音色
"""

from __future__ import annotations

import time
from unittest.mock import MagicMock, patch

import pytest

from packages.application.cosyvoice_service import CosyVoiceError, SynthesizeResult
from packages.domain.preset_voices import PRESET_VOICES

# ── fixtures ──────────────────────────────────────────────────────────


@pytest.fixture()
def mock_cosyvoice():
    """构造一个模拟的 CosyVoiceService。"""
    svc = MagicMock()
    svc.synthesize_speech.return_value = SynthesizeResult(
        audio_url="https://fresh-url.example.com/preview.mp3",
        duration=2.5,
        file_size=40000,
        request_id="test-req-id",
    )
    return svc


@pytest.fixture()
def failing_cosyvoice():
    """构造一个总是失败的 CosyVoiceService。"""
    svc = MagicMock()
    svc.synthesize_speech.side_effect = CosyVoiceError("TTS API unavailable")
    return svc


@pytest.fixture(autouse=True)
def clear_cache():
    """每个测试前清空预览缓存。"""
    from apps.api.app.api.routes import voices

    voices._preset_preview_cache.clear()
    yield
    voices._preset_preview_cache.clear()


# ── _resolve_preset_preview_url ──────────────────────────────────────


class TestResolvePresetPreviewUrl:
    """测试单个预置音色的 preview_url 解析。"""

    def test_cache_hit_returns_cached_url(self, mock_cosyvoice):
        """缓存有效时直接返回，不调用 CosyVoice。"""
        from apps.api.app.api.routes.voices import _preset_preview_cache, _resolve_preset_preview_url

        voice_id = "longxiaochun_v3"
        _preset_preview_cache[voice_id] = ("https://cached-url.example.com/audio.mp3", time.time())

        result = _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)

        assert result == "https://cached-url.example.com/audio.mp3"
        mock_cosyvoice.synthesize_speech.assert_not_called()

    def test_cache_expired_triggers_synthesis(self, mock_cosyvoice):
        """缓存过期时重新调用 CosyVoice 合成。"""
        from apps.api.app.api.routes.voices import _preset_preview_cache, _resolve_preset_preview_url

        voice_id = "longxiaochun_v3"
        # 设置一个过期的缓存（10天前）
        _preset_preview_cache[voice_id] = ("https://old-url.mp3", time.time() - 10 * 86400)

        result = _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)

        assert result == "https://fresh-url.example.com/preview.mp3"
        mock_cosyvoice.synthesize_speech.assert_called_once()
        # 新 URL 应该被缓存
        assert voice_id in _preset_preview_cache
        cached_url, _ = _preset_preview_cache[voice_id]
        assert cached_url == "https://fresh-url.example.com/preview.mp3"

    def test_cache_miss_triggers_synthesis(self, mock_cosyvoice):
        """缓存不存在时调用 CosyVoice 合成。"""
        from apps.api.app.api.routes.voices import _resolve_preset_preview_url

        voice_id = "longsanshu_v3"
        result = _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)

        assert result == "https://fresh-url.example.com/preview.mp3"
        mock_cosyvoice.synthesize_speech.assert_called_once()

    def test_synthesis_failure_returns_fallback(self, failing_cosyvoice):
        """CosyVoice 合成失败时降级返回硬编码 URL。"""
        from apps.api.app.api.routes.voices import _resolve_preset_preview_url

        voice_id = "longyue_v3"
        fallback = "https://hardcoded-fallback.example.com/audio.mp3"
        result = _resolve_preset_preview_url(voice_id, fallback, failing_cosyvoice)

        assert result == fallback
        failing_cosyvoice.synthesize_speech.assert_called_once()

    def test_nonexistent_voice_returns_fallback(self, mock_cosyvoice):
        """voice_id 不存在时直接返回 fallback。"""
        from apps.api.app.api.routes.voices import _resolve_preset_preview_url

        result = _resolve_preset_preview_url("nonexistent_voice_id", "fallback.mp3", mock_cosyvoice)

        assert result == "fallback.mp3"
        mock_cosyvoice.synthesize_speech.assert_not_called()

    def test_synthesis_call_uses_correct_params(self, mock_cosyvoice):
        """验证调用 CosyVoice 时参数正确。"""
        from apps.api.app.api.routes.voices import _resolve_preset_preview_url

        voice_id = "longshu_v3"
        _resolve_preset_preview_url(voice_id, "fallback.mp3", mock_cosyvoice)

        mock_cosyvoice.synthesize_speech.assert_called_once_with(
            text="你好，我是龙书，很高兴认识你。",
            voice_id="longshu_v3",
            format="mp3",
            speed=1.0,
        )


# ── _resolve_all_preset_preview_urls ─────────────────────────────────


class TestResolveAllPresetPreviewUrls:
    """测试批量并行解析所有预置音色的 preview_url。"""

    def test_all_voices_resolved(self, mock_cosyvoice):
        """所有 8 个预置音色都能解析出 preview_url。"""
        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls

        result = _resolve_all_preset_preview_urls(PRESET_VOICES, mock_cosyvoice)

        assert len(result) == 8
        for preset in PRESET_VOICES:
            assert preset.voice_id in result
            assert result[preset.voice_id] == "https://fresh-url.example.com/preview.mp3"

    def test_partial_failure_still_returns_all(self, mock_cosyvoice):
        """部分合成失败时仍返回所有 voice_id（成功的用新 URL，失败的用 fallback）。"""
        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls

        # 让第一个音色失败，其余成功
        call_count = 0
        original_side_effect = mock_cosyvoice.synthesize_speech.side_effect

        def side_effect_fn(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                raise CosyVoiceError("first call fails")
            return SynthesizeResult(audio_url="https://ok.mp3", duration=1.0)

        mock_cosyvoice.synthesize_speech.side_effect = side_effect_fn

        result = _resolve_all_preset_preview_urls(PRESET_VOICES, mock_cosyvoice)

        assert len(result) == 8
        # 第一个失败的音色应该返回 fallback（硬编码 URL）
        first_voice = PRESET_VOICES[0]
        assert result[first_voice.voice_id] == first_voice.preview_url

    def test_all_fail_returns_empty_strings_or_fallbacks(self, failing_cosyvoice):
        """全部合成失败时，每个 voice_id 都有 fallback URL。"""
        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls

        result = _resolve_all_preset_preview_urls(PRESET_VOICES, failing_cosyvoice)

        assert len(result) == 8
        for preset in PRESET_VOICES:
            # 失败时返回硬编码的 fallback URL
            assert result[preset.voice_id] == preset.preview_url

    def test_empty_presets_returns_empty(self, mock_cosyvoice):
        """空列表时返回空字典。"""
        from apps.api.app.api.routes.voices import _resolve_all_preset_preview_urls

        result = _resolve_all_preset_preview_urls([], mock_cosyvoice)

        assert result == {}


# ── _preset_to_unified_response with preview_url_map ─────────────────


class TestPresetToUnifiedResponseWithMap:
    """测试 _preset_to_unified_response 支持 preview_url_map 参数。"""

    def test_with_preview_url_map(self):
        """传入 preview_url_map 时优先使用 map 中的 URL。"""
        from apps.api.app.api.routes.voices import _preset_to_unified_response

        preset = PRESET_VOICES[0]
        url_map = {preset.voice_id: "https://dynamic-url.example.com/audio.mp3"}

        result = _preset_to_unified_response(preset, url_map)

        assert result.preview_url == "https://dynamic-url.example.com/audio.mp3"

    def test_without_map_uses_hardcoded(self):
        """不传 preview_url_map 时使用硬编码 URL。"""
        from apps.api.app.api.routes.voices import _preset_to_unified_response

        preset = PRESET_VOICES[0]
        result = _preset_to_unified_response(preset)

        assert result.preview_url == preset.preview_url

    def test_map_missing_voice_falls_back(self):
        """preview_url_map 中没有对应 voice_id 时使用硬编码 URL。"""
        from apps.api.app.api.routes.voices import _preset_to_unified_response

        preset = PRESET_VOICES[0]
        url_map = {"other_voice_id": "https://other.mp3"}

        result = _preset_to_unified_response(preset, url_map)

        assert result.preview_url == preset.preview_url