xiaoxia-saas/tests/unit/test_tts_job_use_cases.py

"""TTS Job Use Cases 单元测试"""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest

from packages.application.tts_job.exceptions import TTSJobNotFoundError
from packages.application.tts_job.use_cases import (
    CreateTTSJobUseCase,
    DeleteTTSJobUseCase,
    GetTTSJobStatusUseCase,
    GetTTSJobUseCase,
    ListTTSJobsUseCase,
)
from packages.domain.tts_job import TTSJob, TTSJobStatus


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


@pytest.fixture
def sample_job():
    return TTSJob.create(
        user_id="user_001",
        input_text="测试文本",
        voice_id="voice_001",
        voice_model="cosyvoice",
        project_id="proj_001",
        sample_rate=22050,
        format="mp3",
        max_retries=3,
    )


class TestCreateTTSJobUseCase:
    """创建 TTS 任务用例测试"""

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

        use_case = CreateTTSJobUseCase(mock_repo)
        result = use_case.execute(
            user_id="user_001",
            input_text="测试文本",
            voice_id="voice_001",
            voice_model="cosyvoice",
            project_id="proj_001",
        )

        assert result is not None
        assert result.user_id == "user_001"
        assert result.input_text == "测试文本"
        assert result.status == TTSJobStatus.PENDING
        mock_repo.create.assert_called_once()

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

        use_case = CreateTTSJobUseCase(mock_repo)
        result = use_case.execute(user_id="user_001", input_text="hello")

        assert result.voice_id == ""
        assert result.voice_model == ""
        assert result.sample_rate == 22050
        assert result.format == "mp3"
        assert result.max_retries == 3

    def test_create_with_metadata(self, mock_repo):
        """创建时携带 metadata"""
        mock_repo.create.side_effect = lambda x: x

        use_case = CreateTTSJobUseCase(mock_repo)
        result = use_case.execute(
            user_id="user_001",
            input_text="test",
            metadata={"source": "api", "priority": "high"},
        )

        assert result.metadata["source"] == "api"
        assert result.metadata["priority"] == "high"

    def test_create_with_voice_clone_profile(self, mock_repo):
        """使用音色克隆档案创建"""
        mock_repo.create.side_effect = lambda x: x

        use_case = CreateTTSJobUseCase(mock_repo)
        result = use_case.execute(
            user_id="user_001",
            input_text="test",
            voice_clone_profile_id="clone_001",
        )

        assert result.voice_clone_profile_id == "clone_001"


class TestListTTSJobsUseCase:
    """列出 TTS 任务用例测试"""

    def test_list_success(self, mock_repo, sample_job):
        """列出任务成功"""
        mock_repo.list_by_user.return_value = [sample_job]
        mock_repo.count_by_user.return_value = 1

        use_case = ListTTSJobsUseCase(mock_repo)
        items, total = use_case.execute(user_id="user_001")

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

    def test_list_with_status_filter(self, mock_repo):
        """按状态过滤"""
        mock_repo.list_by_user.return_value = []
        mock_repo.count_by_user.return_value = 0

        use_case = ListTTSJobsUseCase(mock_repo)
        items, total = use_case.execute(user_id="user_001", status="completed")

        assert total == 0
        mock_repo.list_by_user.assert_called_once_with("user_001", status="completed", limit=50, offset=0)
        mock_repo.count_by_user.assert_called_once_with("user_001", status="completed")

    def test_list_with_pagination(self, mock_repo):
        """分页参数正确传递"""
        mock_repo.list_by_user.return_value = []
        mock_repo.count_by_user.return_value = 0

        use_case = ListTTSJobsUseCase(mock_repo)
        use_case.execute(user_id="user_001", skip=10, limit=20)

        mock_repo.list_by_user.assert_called_once_with("user_001", status=None, limit=20, offset=10)

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

        use_case = ListTTSJobsUseCase(mock_repo)
        items, total = use_case.execute(user_id="user_001")

        assert items == []
        assert total == 0


class TestGetTTSJobUseCase:
    """获取 TTS 任务详情用例测试"""

    def test_get_success(self, mock_repo, sample_job):
        """获取成功"""
        mock_repo.get.return_value = sample_job

        use_case = GetTTSJobUseCase(mock_repo)
        result = use_case.execute(job_id=sample_job.id, user_id="user_001")

        assert result.id == sample_job.id
        mock_repo.get.assert_called_once_with(sample_job.id)

    def test_get_not_found(self, mock_repo):
        """任务不存在"""
        mock_repo.get.return_value = None

        use_case = GetTTSJobUseCase(mock_repo)

        with pytest.raises(TTSJobNotFoundError):
            use_case.execute(job_id="nonexistent", user_id="user_001")

    def test_get_wrong_user(self, mock_repo, sample_job):
        """用户不匹配"""
        mock_repo.get.return_value = sample_job  # user_001

        use_case = GetTTSJobUseCase(mock_repo)

        with pytest.raises(TTSJobNotFoundError):
            use_case.execute(job_id=sample_job.id, user_id="other_user")


class TestGetTTSJobStatusUseCase:
    """查询 TTS 任务状态用例测试"""

    def test_get_status_success(self, mock_repo, sample_job):
        """获取状态成功"""
        mock_repo.get.return_value = sample_job

        use_case = GetTTSJobStatusUseCase(mock_repo)
        result = use_case.execute(job_id=sample_job.id, user_id="user_001")

        assert result.status == TTSJobStatus.PENDING

    def test_get_status_not_found(self, mock_repo):
        """任务不存在抛异常"""
        mock_repo.get.return_value = None

        use_case = GetTTSJobStatusUseCase(mock_repo)

        with pytest.raises(TTSJobNotFoundError):
            use_case.execute(job_id="nonexistent", user_id="user_001")

    def test_get_status_wrong_user(self, mock_repo, sample_job):
        """用户不匹配抛异常"""
        mock_repo.get.return_value = sample_job

        use_case = GetTTSJobStatusUseCase(mock_repo)

        with pytest.raises(TTSJobNotFoundError):
            use_case.execute(job_id=sample_job.id, user_id="other_user")


class TestDeleteTTSJobUseCase:
    """删除 TTS 任务用例测试"""

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

        use_case = DeleteTTSJobUseCase(mock_repo)
        result = use_case.execute(job_id=sample_job.id, user_id="user_001")

        assert result is True
        mock_repo.delete.assert_called_once_with(sample_job.id)

    def test_delete_not_found(self, mock_repo):
        """任务不存在返回 False"""
        mock_repo.get.return_value = None

        use_case = DeleteTTSJobUseCase(mock_repo)
        result = use_case.execute(job_id="nonexistent", user_id="user_001")

        assert result is False
        mock_repo.delete.assert_not_called()

    def test_delete_wrong_user(self, mock_repo, sample_job):
        """用户不匹配返回 False"""
        mock_repo.get.return_value = sample_job

        use_case = DeleteTTSJobUseCase(mock_repo)
        result = use_case.execute(job_id=sample_job.id, user_id="other_user")

        assert result is False
        mock_repo.delete.assert_not_called()