xiaoxia-saas/tests/unit/test_tts_synthesis_failure.py

"""TTS 合成失败场景单元测试。

验证 TTS 合成接口在 CosyVoice 失败时返回 201 + failed 状态（与音色克隆一致），
而不是抛 500 错误。
"""

from __future__ import annotations

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

import pytest

from packages.application.cosyvoice_service import (
    CosyVoiceAuthError,
    CosyVoiceError,
    CosyVoiceService,
    CosyVoiceTimeoutError,
)
from packages.application.tts_job.workflow import TTSWorkflowService
from packages.domain.tts_job import TTSJob, TTSJobStatus


def _make_job(**kwargs) -> TTSJob:
    defaults = {
        "id": "test_job_001",
        "user_id": "user_001",
        "input_text": "测试文本",
        "voice_id": "voice_001",
        "voice_model": "",
        "project_id": "",
        "voice_clone_profile_id": "",
        "status": TTSJobStatus.PENDING,
        "output_audio_url": "",
        "output_audio_key": "",
        "duration": 0.0,
        "file_size": 0,
        "sample_rate": 22050,
        "format": "mp3",
        "error_message": "",
        "retry_count": 0,
        "max_retries": 3,
        "metadata": {},
        "started_at": None,
        "completed_at": None,
        "created_at": datetime.now(timezone.utc),
        "updated_at": datetime.now(timezone.utc),
    }
    defaults.update(kwargs)
    return TTSJob(**defaults)


class TestTTSWorkflowFailureScenarios:
    """测试 TTSWorkflowService 在各种失败场景下的行为。

    核心要求：CosyVoice 失败时，job 应被标记为 failed，而不是抛异常。
    """

    def _make_workflow(self, cosyvoice_service: MagicMock, repo: MagicMock | None = None) -> TTSWorkflowService:
        if repo is None:
            repo = MagicMock()
            repo.get.return_value = _make_job()
            repo.update.side_effect = lambda j: j
        return TTSWorkflowService(repository=repo, cosyvoice_service=cosyvoice_service)

    def test_auth_error_marks_job_failed(self) -> None:
        """CosyVoiceAuthError（API Key 未配置）→ job 标记为 failed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = CosyVoiceAuthError("CosyVoice API Key 未配置")

        workflow = self._make_workflow(service)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.FAILED
        assert "API Key 未配置" in job.error_message

    def test_api_error_marks_job_failed(self) -> None:
        """CosyVoiceError（API 调用失败）→ job 标记为 failed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = CosyVoiceError("CosyVoice API 服务端错误: HTTP 500")

        workflow = self._make_workflow(service)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.FAILED
        assert "服务端错误" in job.error_message

    def test_timeout_error_marks_job_failed(self) -> None:
        """CosyVoiceTimeoutError（超时）→ job 标记为 failed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = CosyVoiceTimeoutError("请求超时")

        workflow = self._make_workflow(service)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.FAILED
        assert "超时" in job.error_message

    def test_value_error_marks_job_failed(self) -> None:
        """ValueError（参数错误）→ job 标记为 failed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = ValueError("voice_id 不能为空")

        workflow = self._make_workflow(service)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.FAILED
        assert "voice_id" in job.error_message

    def test_unexpected_exception_propagates(self) -> None:
        """非预期异常（如 DB 错误）→ workflow 不捕获，由路由层兜底。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = RuntimeError("unexpected DB error")

        workflow = self._make_workflow(service)

        with pytest.raises(RuntimeError, match="unexpected DB error"):
            workflow.start_synthesis("test_job_001")

    def test_sync_success_marks_job_completed(self) -> None:
        """CosyVoice 同步返回 audio_url → job 标记为 completed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.return_value = {
            "task_id": "",
            "audio_url": "https://example.com/audio.mp3",
            "duration": 3.5,
            "file_size": 12345,
            "request_id": "req_001",
        }

        workflow = self._make_workflow(service)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.COMPLETED
        assert job.output_audio_url == "https://example.com/audio.mp3"
        assert job.duration == 3.5

    def test_async_success_marks_job_processing(self) -> None:
        """CosyVoice 异步模式（返回 task_id）→ job 标记为 processing。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.return_value = {
            "task_id": "cosy_task_001",
            "audio_url": "",
            "duration": 0.0,
            "file_size": 0,
            "request_id": "req_002",
        }

        workflow = self._make_workflow(service)
        job = workflow.start_synthesis("test_job_001")

        assert job.status == TTSJobStatus.PROCESSING
        assert job.metadata.get("cosyvoice_task_id") == "cosy_task_001"

    def test_process_synthesis_failure(self) -> None:
        """process_synthesis_failure 将 job 标记为 failed。"""
        repo = MagicMock()
        job = _make_job(status=TTSJobStatus.PROCESSING)
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j

        service = MagicMock(spec=CosyVoiceService)
        workflow = TTSWorkflowService(repository=repo, cosyvoice_service=service)

        result = workflow.process_synthesis_failure("test_job_001", "Celery 任务调度失败")

        assert result.status == TTSJobStatus.FAILED
        assert "Celery 任务调度失败" in result.error_message


class TestTTSRouteFailureScenarios:
    """测试 TTS 路由层在各种失败场景下的行为。

    核心要求：即使 workflow 抛出意外异常，路由也应返回 201 + failed，不抛 500。
    """

    def test_route_catches_unexpected_exception_from_workflow(self) -> None:
        """路由层兜底：workflow 抛出非预期异常时，job 仍标记为 failed。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = RuntimeError("unexpected")

        repo = MagicMock()
        job = _make_job()
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j
        repo.create.side_effect = lambda j: j

        # 模拟路由逻辑
        from packages.application.tts_job.use_cases import CreateTTSJobUseCase
        from packages.application.tts_job.workflow import TTSWorkflowService

        use_case = CreateTTSJobUseCase(repo)
        created_job = use_case.execute(
            user_id="user_001",
            input_text="测试文本",
            voice_id="voice_001",
        )

        workflow = TTSWorkflowService(repository=repo, cosyvoice_service=service)

        # 模拟路由的 try/except 兜底逻辑
        try:
            result_job = workflow.start_synthesis(created_job.id)
        except Exception as e:
            result_job = workflow.process_synthesis_failure(created_job.id, str(e))

        assert result_job.status == TTSJobStatus.FAILED
        assert "unexpected" in result_job.error_message

    def test_route_handles_double_failure_gracefully(self) -> None:
        """路由层兜底：即使 process_synthesis_failure 也失败，不崩溃。"""
        service = MagicMock(spec=CosyVoiceService)
        service.submit_synthesize_task.side_effect = RuntimeError("unexpected")

        repo = MagicMock()
        job = _make_job()
        repo.get.return_value = job
        repo.update.side_effect = lambda j: j
        repo.create.side_effect = lambda j: j

        from packages.application.tts_job.use_cases import CreateTTSJobUseCase
        from packages.application.tts_job.workflow import TTSWorkflowService

        use_case = CreateTTSJobUseCase(repo)
        created_job = use_case.execute(
            user_id="user_001",
            input_text="测试文本",
            voice_id="voice_001",
        )

        # 让 process_synthesis_failure 也失败
        call_count = 0
        original_update = repo.update

        def failing_update(j):
            nonlocal call_count
            call_count += 1
            if call_count > 2:  # 第3次调用时失败（mark_failed 时的 update）
                raise RuntimeError("DB connection lost")
            return original_update(j)

        repo.update.side_effect = failing_update

        workflow = TTSWorkflowService(repository=repo, cosyvoice_service=service)

        # 模拟路由的 try/except 兜底逻辑（包含内层 try/except）
        try:
            result_job = workflow.start_synthesis(created_job.id)
        except Exception as e:
            try:
                result_job = workflow.process_synthesis_failure(created_job.id, str(e))
            except Exception:
                # 路由层日志记录后，使用原始 job 对象
                result_job = created_job

        # 不应崩溃，即使状态可能不是 failed
        assert result_job is not None