xiaoxia-saas/tests/unit/test_voice_clone_use_cases.py

"""音色克隆 Voice Clone UseCase 单元测试."""

from __future__ import annotations

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


def _make_profile(
    id: str = "vc1",
    user_id: str = "user_1",
    name: str = "我的音色",
    status: str = "completed",
) -> VoiceCloneProfile:
    p = VoiceCloneProfile.create(
        user_id=user_id,
        name=name,
        description="测试音色",
        source_audio_url="https://oss.example.com/source.wav",
        voice_model="cosyvoice",
        language="zh-CN",
        gender="female",
        max_retries=3,
    )
    p.id = id
    # 直接设状态绕过状态机
    object.__setattr__(p, "status", status)
    return p


@pytest.fixture
def mock_repo():
    return MagicMock()


@pytest.fixture
def sample_profile():
    return _make_profile()


class TestCreateVoiceCloneUseCase:
    """CreateVoiceCloneUseCase 测试"""

    def test_create_success(self, mock_repo, sample_profile):
        """创建成功"""
        mock_repo.create.return_value = sample_profile
        use_case = CreateVoiceCloneUseCase(mock_repo)

        result = use_case.execute(
            "user_1",
            "新音色",
            description="自定义音色",
            source_audio_url="https://oss.example.com/src.wav",
            voice_model="cosyvoice",
            language="zh-CN",
            gender="female",
        )

        assert result.id == "vc1"
        assert result.name == "我的音色"
        mock_repo.create.assert_called_once()
        created = mock_repo.create.call_args[0][0]
        assert isinstance(created, VoiceCloneProfile)

    def test_create_with_defaults(self, mock_repo):
        """使用默认参数创建"""
        mock_repo.create.side_effect = lambda x: x
        use_case = CreateVoiceCloneUseCase(mock_repo)

        result = use_case.execute("user_1", "极简音色")

        assert result.user_id == "user_1"
        assert result.name == "极简音色"
        assert result.language == "zh-CN"
        assert result.gender == "unknown"
        assert result.max_retries == 3


class TestListVoiceClonesUseCase:
    """ListVoiceClonesUseCase 测试"""

    def test_list_returns_items_and_total(self, mock_repo, sample_profile):
        """返回 (items, total) 元组"""
        mock_repo.list_by_user.return_value = [sample_profile]
        mock_repo.count_by_user.return_value = 1
        use_case = ListVoiceClonesUseCase(mock_repo)

        items, total = use_case.execute("user_1")

        assert len(items) == 1
        assert total == 1
        mock_repo.list_by_user.assert_called_once_with(
            "user_1", status=None, limit=50, offset=0
        )
        mock_repo.count_by_user.assert_called_once_with("user_1", status=None)

    def test_list_with_status(self, mock_repo, sample_profile):
        """按状态过滤"""
        mock_repo.list_by_user.return_value = [sample_profile]
        mock_repo.count_by_user.return_value = 1
        use_case = ListVoiceClonesUseCase(mock_repo)

        use_case.execute("user_1", status="processing")

        mock_repo.list_by_user.assert_called_once_with(
            "user_1", status="processing", limit=50, offset=0
        )
        mock_repo.count_by_user.assert_called_once_with("user_1", status="processing")

    def test_list_with_pagination(self, mock_repo, sample_profile):
        """带分页参数"""
        mock_repo.list_by_user.return_value = [sample_profile]
        mock_repo.count_by_user.return_value = 10
        use_case = ListVoiceClonesUseCase(mock_repo)

        use_case.execute("user_1", skip=5, limit=10)

        mock_repo.list_by_user.assert_called_once_with(
            "user_1", status=None, limit=10, offset=5
        )

    def test_empty_list(self, mock_repo):
        """空列表"""
        mock_repo.list_by_user.return_value = []
        mock_repo.count_by_user.return_value = 0
        use_case = ListVoiceClonesUseCase(mock_repo)

        items, total = use_case.execute("user_1")

        assert items == []
        assert total == 0


class TestGetVoiceCloneUseCase:
    """GetVoiceCloneUseCase 测试"""

    def test_get_existing(self, mock_repo, sample_profile):
        """获取存在的音色克隆"""
        mock_repo.get.return_value = sample_profile
        use_case = GetVoiceCloneUseCase(mock_repo)

        result = use_case.execute("vc1", "user_1")

        assert result.id == "vc1"
        mock_repo.get.assert_called_once_with("vc1")

    def test_get_nonexistent_raises(self, mock_repo):
        """获取不存在的抛出 VoiceCloneNotFoundError"""
        mock_repo.get.return_value = None
        use_case = GetVoiceCloneUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotFoundError, match="not found"):
            use_case.execute("noexist", "user_1")

    def test_get_wrong_user_raises(self, mock_repo, sample_profile):
        """非本人的音色克隆抛出 NotFoundError"""
        sample_profile.user_id = "other_user"
        mock_repo.get.return_value = sample_profile
        use_case = GetVoiceCloneUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("vc1", "user_1")


class TestGetVoiceCloneStatusUseCase:
    """GetVoiceCloneStatusUseCase 测试"""

    def test_get_status_completed(self, mock_repo, sample_profile):
        """获取 completed 状态"""
        mock_repo.get.return_value = sample_profile
        use_case = GetVoiceCloneStatusUseCase(mock_repo)

        result = use_case.execute("vc1", "user_1")

        assert result.status == "completed"

    def test_get_status_not_found(self, mock_repo):
        """不存在抛出"""
        mock_repo.get.return_value = None
        use_case = GetVoiceCloneStatusUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("noexist", "user_1")

    def test_get_status_wrong_user(self, mock_repo, sample_profile):
        """非本人抛出"""
        sample_profile.user_id = "other"
        mock_repo.get.return_value = sample_profile
        use_case = GetVoiceCloneStatusUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("vc1", "user_1")


class TestDeleteVoiceCloneUseCase:
    """DeleteVoiceCloneUseCase 测试"""

    def test_delete_success(self, mock_repo, sample_profile):
        """删除成功"""
        mock_repo.get.return_value = sample_profile
        mock_repo.delete.return_value = True
        use_case = DeleteVoiceCloneUseCase(mock_repo)

        result = use_case.execute("vc1", "user_1")

        assert result is True
        mock_repo.get.assert_called_once_with("vc1")
        mock_repo.delete.assert_called_once_with("vc1")

    def test_delete_nonexistent_returns_false(self, mock_repo):
        """不存在返回 False"""
        mock_repo.get.return_value = None
        use_case = DeleteVoiceCloneUseCase(mock_repo)

        result = use_case.execute("noexist", "user_1")

        assert result is False
        mock_repo.delete.assert_not_called()

    def test_delete_wrong_user_returns_false(self, mock_repo, sample_profile):
        """非本人返回 False"""
        sample_profile.user_id = "other"
        mock_repo.get.return_value = sample_profile
        use_case = DeleteVoiceCloneUseCase(mock_repo)

        result = use_case.execute("vc1", "user_1")

        assert result is False
        mock_repo.delete.assert_not_called()


class TestRetryVoiceCloneUseCase:
    """RetryVoiceCloneUseCase 测试"""

    def test_retry_failed_profile(self, mock_repo):
        """失败的音色克隆可以重试"""
        profile = _make_profile(status="failed")
        mock_repo.get.return_value = profile
        mock_repo.update.side_effect = lambda x: x
        use_case = RetryVoiceCloneUseCase(mock_repo)

        result = use_case.execute("vc1", "user_1")

        assert result is not None
        assert result.status == "pending"
        assert result.retry_count >= 1
        mock_repo.update.assert_called_once()

    def test_retry_not_found(self, mock_repo):
        """不存在抛出"""
        mock_repo.get.return_value = None
        use_case = RetryVoiceCloneUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("noexist", "user_1")

        mock_repo.update.assert_not_called()

    def test_retry_wrong_user(self, mock_repo, sample_profile):
        """非本人抛出"""
        sample_profile.user_id = "other"
        mock_repo.get.return_value = sample_profile
        use_case = RetryVoiceCloneUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotFoundError):
            use_case.execute("vc1", "user_1")

        mock_repo.update.assert_not_called()

    def test_retry_completed_not_retryable(self, mock_repo, sample_profile):
        """completed 状态不可重试"""
        mock_repo.get.return_value = sample_profile
        use_case = RetryVoiceCloneUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotRetryableError):
            use_case.execute("vc1", "user_1")

        mock_repo.update.assert_not_called()

    def test_retry_processing_not_retryable(self, mock_repo):
        """processing 状态不可重试"""
        profile = _make_profile(status="processing")
        mock_repo.get.return_value = profile
        use_case = RetryVoiceCloneUseCase(mock_repo)

        with pytest.raises(VoiceCloneNotRetryableError):
            use_case.execute("vc1", "user_1")