xiaoxia-saas/tests/unit/test_voices_clone_from_profile.py

"""统一音色列表接口单元测试 (#1196).

验证 /voices?type=clone 从 voice_clone_profile 读取克隆音色。
"""

from __future__ import annotations

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

import pytest

from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus


def _make_clone_profile(**kwargs) -> VoiceCloneProfile:
    defaults = {
        "id": "clone_001",
        "user_id": "user_001",
        "name": "我的克隆音色",
        "description": "测试用克隆音色",
        "source_audio_url": "https://example.com/source.mp3",
        "voice_id": "cosyvoice_clone_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(2026, 7, 30, 10, 0, 0, tzinfo=timezone.utc),
        "updated_at": datetime(2026, 7, 30, 10, 5, 0, tzinfo=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 TestVoicesCloneFromProfile:
    """测试 /voices?type=clone 从克隆音色档案读取。"""

    def _call_list_voices(
        self,
        type_filter=None,
        status_filter=None,
        skip=0,
        limit=50,
        clone_profiles=None,
        clone_count=0,
        user_id="user_001",
    ):
        """调用 list_voices_unified 路由函数。"""
        from app.api.routes.voices import list_voices_unified

        # mock 依赖
        voice_repo = MagicMock()
        clone_profile_repo = MagicMock()
        clone_profile_repo.list_by_user.return_value = clone_profiles or []
        clone_profile_repo.count_by_user.return_value = clone_count

        sign_url = MagicMock(side_effect=lambda u: u + "?signed=1")

        result = list_voices_unified(
            type=type_filter,
            status_filter=status_filter,
            skip=skip,
            limit=limit,
            authenticated_user=_make_auth_user(user_id),
            voice_repository=voice_repo,
            clone_profile_repository=clone_profile_repo,
            sign_url=sign_url,
        )
        return result, clone_profile_repo, voice_repo

    def test_type_clone_returns_ready_clones(self) -> None:
        """type=clone 返回 ready 状态的克隆音色。"""
        profiles = [
            _make_clone_profile(id="c1", name="音色1", voice_id="v1"),
            _make_clone_profile(id="c2", name="音色2", voice_id="v2"),
        ]

        result, clone_repo, voice_repo = self._call_list_voices(
            type_filter="clone",
            clone_profiles=profiles,
            clone_count=2,
        )

        assert result.total == 2
        assert result.clone_count == 2
        assert result.preset_count == 0
        assert len(result.items) == 2

        # 验证字段
        item = result.items[0]
        assert item.id == "c1"
        assert item.type == "clone"
        assert item.name == "音色1"
        assert item.voice_id == "v1"
        assert item.voice_clone_profile_id == "c1"
        assert item.status == "ready"
        assert item.gender == "female"
        assert item.language == "zh-CN"
        assert item.user_id == "user_001"

        # 调用了 clone_profile_repository
        clone_repo.list_by_user.assert_called_once()
        call_kwargs = clone_repo.list_by_user.call_args
        assert call_kwargs.kwargs.get("status") == "ready"

        # 没有调用 voice_repository（克隆音色不从配音库取）
        voice_repo.execute.assert_not_called()

    def test_type_clone_default_status_ready(self) -> None:
        """不传 status 时默认只返回 ready 状态。"""
        result, clone_repo, _ = self._call_list_voices(
            type_filter="clone",
            clone_profiles=[],
            clone_count=0,
        )

        clone_repo.list_by_user.assert_called_once()
        call_kwargs = clone_repo.list_by_user.call_args
        assert call_kwargs.kwargs.get("status") == "ready"

    def test_type_clone_status_filter(self) -> None:
        """status 参数可以过滤特定状态。"""
        result, clone_repo, _ = self._call_list_voices(
            type_filter="clone",
            status_filter="processing",
            clone_profiles=[],
            clone_count=0,
        )

        clone_repo.list_by_user.assert_called_once()
        call_kwargs = clone_repo.list_by_user.call_args
        assert call_kwargs.kwargs.get("status") == "processing"

    def test_type_clone_status_all(self) -> None:
        """status=all 返回所有状态的克隆音色。"""
        result, clone_repo, _ = self._call_list_voices(
            type_filter="clone",
            status_filter="all",
            clone_profiles=[],
            clone_count=0,
        )

        clone_repo.list_by_user.assert_called_once()
        call_kwargs = clone_repo.list_by_user.call_args
        assert call_kwargs.kwargs.get("status") is None

    def test_type_preset_no_clone_data(self) -> None:
        """type=preset 时不查询克隆音色。"""
        result, clone_repo, _ = self._call_list_voices(
            type_filter="preset",
            clone_profiles=[],
            clone_count=0,
        )

        assert result.preset_count > 0
        assert result.clone_count == 0
        clone_repo.list_by_user.assert_not_called()

    def test_type_none_returns_both(self) -> None:
        """不传 type 时返回预置 + 克隆。"""
        profiles = [_make_clone_profile(id="c1", name="克隆音色")]

        result, clone_repo, _ = self._call_list_voices(
            type_filter=None,
            clone_profiles=profiles,
            clone_count=1,
        )

        assert result.preset_count > 0
        assert result.clone_count == 1
        assert result.total == result.preset_count + 1
        clone_repo.list_by_user.assert_called_once()

    def test_pagination(self) -> None:
        """分页参数正确传递。"""
        result, clone_repo, _ = self._call_list_voices(
            type_filter="clone",
            skip=10,
            limit=20,
            clone_profiles=[],
            clone_count=50,
        )

        clone_repo.list_by_user.assert_called_once()
        call_kwargs = clone_repo.list_by_user.call_args
        assert call_kwargs.kwargs.get("offset") == 10
        assert call_kwargs.kwargs.get("limit") == 20

    def test_empty_clone_list(self) -> None:
        """用户没有克隆音色时返回空列表。"""
        result, clone_repo, _ = self._call_list_voices(
            type_filter="clone",
            clone_profiles=[],
            clone_count=0,
        )

        assert result.total == 0
        assert result.clone_count == 0
        assert len(result.items) == 0

    def test_clone_item_has_voice_id(self) -> None:
        """克隆音色条目包含 voice_id，前端可用于 TTS 合成。"""
        profile = _make_clone_profile(voice_id="my_clone_voice_123")
        result, _, _ = self._call_list_voices(
            type_filter="clone",
            clone_profiles=[profile],
            clone_count=1,
        )

        assert result.items[0].voice_id == "my_clone_voice_123"
        assert result.items[0].voice_provider == "cosyvoice-v3"

    def test_clone_item_no_audio_url(self) -> None:
        """克隆音色没有预合成 audio_url（需通过 preview 接口试听）。"""
        profile = _make_clone_profile()
        result, _, _ = self._call_list_voices(
            type_filter="clone",
            clone_profiles=[profile],
            clone_count=1,
        )

        item = result.items[0]
        assert item.audio_url == ""
        assert item.preview_url == ""
        assert item.duration == 0
        assert item.file_size == 0

    def test_clone_item_has_profile_id(self) -> None:
        """克隆音色条目携带 voice_clone_profile_id，前端可用于调用 preview 接口。"""
        profile = _make_clone_profile(id="profile_abc_123")
        result, _, _ = self._call_list_voices(
            type_filter="clone",
            clone_profiles=[profile],
            clone_count=1,
        )

        assert result.items[0].voice_clone_profile_id == "profile_abc_123"

    def test_multiple_statuses(self) -> None:
        """不同状态的克隆音色返回正确的 status 字段。"""
        for status in [
            VoiceCloneStatus.PENDING,
            VoiceCloneStatus.PROCESSING,
            VoiceCloneStatus.READY,
            VoiceCloneStatus.FAILED,
        ]:
            profile = _make_clone_profile(id=f"c_{status.value}", status=status)
            result, _, _ = self._call_list_voices(
                type_filter="clone",
                status_filter=status.value,
                clone_profiles=[profile],
                clone_count=1,
            )
            assert result.items[0].status == status.value