xiaoxia-saas/tests/unit/test_voice_clone_api.py

"""音色克隆 API 单元测试。"""

from __future__ import annotations

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

import pytest

from packages.application.voice_clone.use_cases import (
    CreateVoiceCloneUseCase,
    DeleteVoiceCloneUseCase,
    GetVoiceCloneStatusUseCase,
    GetVoiceCloneUseCase,
    ListVoiceClonesUseCase,
    RetryVoiceCloneUseCase,
    VoiceCloneNotFoundError,
    VoiceCloneNotRetryableError,
)
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus


def _make_profile(**kwargs) -> VoiceCloneProfile:
    defaults = {
        "id": "test_clone_001",
        "user_id": "user_001",
        "name": "测试音色",
        "description": "测试描述",
        "source_audio_url": "",
        "voice_id": "",
        "voice_model": "",
        "language": "zh-CN",
        "gender": "unknown",
        "status": VoiceCloneStatus.PENDING,
        "error_message": "",
        "retry_count": 0,
        "max_retries": 3,
        "metadata": {},
        "created_at": datetime.now(UTC),
        "updated_at": datetime.now(UTC),
    }
    defaults.update(kwargs)
    return VoiceCloneProfile(**defaults)


class TestCreateVoiceCloneUseCase:
    """测试创建音色克隆。"""

    def test_create_success(self) -> None:
        """正常创建。"""
        repo = MagicMock()
        repo.create.side_effect = lambda p: p

        use_case = CreateVoiceCloneUseCase(repo)
        profile = use_case.execute(
            user_id="user_001",
            name="我的音色",
            description="克隆音色",
            source_audio_url="https://example.com/audio.mp3",
        )

        assert profile.user_id == "user_001"
        assert profile.name == "我的音色"
        assert profile.status == VoiceCloneStatus.PENDING
        repo.create.assert_called_once()

    def test_create_with_metadata(self) -> None:
        """带元数据创建。"""
        repo = MagicMock()
        repo.create.side_effect = lambda p: p

        use_case = CreateVoiceCloneUseCase(repo)
        profile = use_case.execute(
            user_id="user_001",
            name="测试",
            metadata={"source": "upload"},
        )

        assert profile.metadata == {"source": "upload"}


class TestListVoiceClonesUseCase:
    """测试列出音色克隆。"""

    def test_list_empty(self) -> None:
        """空列表。"""
        repo = MagicMock()
        repo.list_by_user.return_value = []
        repo.count_by_user.return_value = 0

        use_case = ListVoiceClonesUseCase(repo)
        items, total = use_case.execute("user_001")

        assert items == []
        assert total == 0
        repo.list_by_user.assert_called_once_with("user_001", status=None, limit=50, offset=0)

    def test_list_with_pagination(self) -> None:
        """分页查询。"""
        repo = MagicMock()
        profiles = [_make_profile(id=f"clone_{i}") for i in range(3)]
        repo.list_by_user.return_value = profiles
        repo.count_by_user.return_value = 10

        use_case = ListVoiceClonesUseCase(repo)
        items, total = use_case.execute("user_001", skip=5, limit=3)

        assert len(items) == 3
        assert total == 10
        repo.list_by_user.assert_called_once_with("user_001", status=None, limit=3, offset=5)

    def test_list_with_status_filter(self) -> None:
        """按状态过滤。"""
        repo = MagicMock()
        repo.list_by_user.return_value = []
        repo.count_by_user.return_value = 0

        use_case = ListVoiceClonesUseCase(repo)
        use_case.execute("user_001", status="pending")

        repo.list_by_user.assert_called_once_with("user_001", status="pending", limit=50, offset=0)


class TestGetVoiceCloneUseCase:
    """测试获取音色克隆详情。"""

    def test_get_success(self) -> None:
        """正常获取。"""
        profile = _make_profile()
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = GetVoiceCloneUseCase(repo)
        result = use_case.execute("test_clone_001", "user_001")

        assert result.id == "test_clone_001"
        assert result.user_id == "user_001"

    def test_get_not_found(self) -> None:
        """档案不存在。"""
        repo = MagicMock()
        repo.get.return_value = None

        use_case = GetVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("nonexistent", "user_001")

    def test_get_wrong_user(self) -> None:
        """用户不匹配。"""
        profile = _make_profile(user_id="other_user")
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = GetVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("test_clone_001", "user_001")


class TestGetVoiceCloneStatusUseCase:
    """测试查询音色克隆状态。"""

    def test_status_success(self) -> None:
        """正常查询状态。"""
        profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = GetVoiceCloneStatusUseCase(repo)
        result = use_case.execute("test_clone_001", "user_001")

        assert result.status == VoiceCloneStatus.PROCESSING

    def test_status_not_found(self) -> None:
        """档案不存在。"""
        repo = MagicMock()
        repo.get.return_value = None

        use_case = GetVoiceCloneStatusUseCase(repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("nonexistent", "user_001")


class TestDeleteVoiceCloneUseCase:
    """测试删除音色克隆。"""

    def test_delete_success(self) -> None:
        """正常删除。"""
        profile = _make_profile()
        repo = MagicMock()
        repo.get.return_value = profile
        repo.delete.return_value = True

        use_case = DeleteVoiceCloneUseCase(repo)
        result = use_case.execute("test_clone_001", "user_001")

        assert result is True
        repo.delete.assert_called_once_with("test_clone_001")

    def test_delete_not_found(self) -> None:
        """档案不存在。"""
        repo = MagicMock()
        repo.get.return_value = None

        use_case = DeleteVoiceCloneUseCase(repo)
        result = use_case.execute("nonexistent", "user_001")

        assert result is False

    def test_delete_wrong_user(self) -> None:
        """用户不匹配。"""
        profile = _make_profile(user_id="other_user")
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = DeleteVoiceCloneUseCase(repo)
        result = use_case.execute("test_clone_001", "user_001")

        assert result is False


class TestRetryVoiceCloneUseCase:
    """测试重试音色克隆。"""

    def test_retry_success(self) -> None:
        """正常重试（failed → pending）。"""
        profile = _make_profile(status=VoiceCloneStatus.FAILED, error_message="API error")
        repo = MagicMock()
        repo.get.return_value = profile
        repo.update.side_effect = lambda p: p

        use_case = RetryVoiceCloneUseCase(repo)
        result = use_case.execute("test_clone_001", "user_001")

        assert result.status == VoiceCloneStatus.PENDING
        assert result.error_message == ""
        assert result.retry_count == 1
        repo.update.assert_called_once()

    def test_retry_not_found(self) -> None:
        """档案不存在。"""
        repo = MagicMock()
        repo.get.return_value = None

        use_case = RetryVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("nonexistent", "user_001")

    def test_retry_not_retryable_pending(self) -> None:
        """pending 状态不可重试。"""
        profile = _make_profile(status=VoiceCloneStatus.PENDING)
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = RetryVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotRetryableError):
            use_case.execute("test_clone_001", "user_001")

    def test_retry_not_retryable_ready(self) -> None:
        """ready 状态不可重试。"""
        profile = _make_profile(status=VoiceCloneStatus.READY)
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = RetryVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotRetryableError):
            use_case.execute("test_clone_001", "user_001")

    def test_retry_not_retryable_processing(self) -> None:
        """processing 状态不可重试。"""
        profile = _make_profile(status=VoiceCloneStatus.PROCESSING)
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = RetryVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotRetryableError):
            use_case.execute("test_clone_001", "user_001")

    def test_retry_wrong_user(self) -> None:
        """用户不匹配。"""
        profile = _make_profile(user_id="other_user", status=VoiceCloneStatus.FAILED)
        repo = MagicMock()
        repo.get.return_value = profile

        use_case = RetryVoiceCloneUseCase(repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("test_clone_001", "user_001")