xiaoxia-saas/tests/unit/test_voice_clone_task.py

"""process_voice_clone Celery 任务单元测试。

关键：voice_clone.py 在模块级别 import worker_app.db.SessionLocal，
而 worker_app.db 会在导入时调用 ensure_database_exists() 尝试连接 PostgreSQL。
因此必须在 @patch 装饰器解析模块路径之前，将 worker_app.db 预注入 sys.modules。

注意：production code 使用 VoiceCloneWorkflowService（非直接 CosyVoiceService），
Celery bind=True 任务通过 __call__ 调用时自动传入 self，不需要手动传 mock_self。
CosyVoiceService 在 voice_clone.py 中被实例化传入 workflow，必须 mock 防止真实初始化。
"""

from __future__ import annotations

import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# worker_app 在 apps/worker 下，需要加入 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))

# ── 预注入 mock 模块，防止 worker_app.db 触发真实数据库连接 ──
_mock_db_module = MagicMock()
_mock_db_module.SessionLocal = MagicMock()
sys.modules.setdefault("worker_app.db", _mock_db_module)

if "worker_app" in sys.modules:
    sys.modules["worker_app"].db = _mock_db_module

from celery.exceptions import Retry

from packages.application.cosyvoice_service import (
    CosyVoiceError,
    CosyVoiceTimeoutError,
)
from packages.application.voice_clone.use_cases import VoiceCloneNotFoundError


def _make_mock_profile(
    *,
    voice_id: str = "voice-xyz",
    status: str = "ready",
) -> MagicMock:
    """创建测试用 mock profile。"""
    profile = MagicMock()
    profile.voice_id = voice_id
    profile.status = status
    return profile


# ── 成功场景 ──────────────────────────────────────────────


class TestProcessVoiceCloneSuccess:
    """测试成功场景。"""

    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
    @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
    def test_process_voice_clone_success(
        self,
        mock_repo_cls: MagicMock,
        mock_workflow_cls: MagicMock,
        mock_cosy_cls: MagicMock,
    ) -> None:
        """克隆成功：轮询返回 voice_id，profile 标记为 ready。"""
        mock_session = MagicMock()
        mock_repo = MagicMock()
        mock_workflow = MagicMock()

        mock_repo_cls.return_value = mock_repo
        mock_workflow_cls.return_value = mock_workflow

        mock_result = _make_mock_profile(voice_id="voice-xyz")
        mock_workflow.poll_and_process_clone.return_value = mock_result

        _mock_db_module.SessionLocal.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        # bind=True: Celery __call__ 自动传入 self，只需传 profile_id
        result = process_voice_clone("profile-123")

        assert result["ok"] is True
        assert result["voice_id"] == "voice-xyz"
        mock_workflow.poll_and_process_clone.assert_called_once_with(
            "profile-123",
            timeout=300,
        )
        mock_session.commit.assert_called_once()
        mock_session.close.assert_called_once()

    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
    @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
    def test_process_voice_clone_profile_not_found(
        self,
        mock_repo_cls: MagicMock,
        mock_workflow_cls: MagicMock,
        mock_cosy_cls: MagicMock,
    ) -> None:
        """profile 不存在时 workflow 抛 VoiceCloneNotFoundError，返回 failed。"""
        mock_session = MagicMock()
        mock_repo = MagicMock()
        mock_workflow = MagicMock()

        mock_repo_cls.return_value = mock_repo
        mock_workflow_cls.return_value = mock_workflow

        # workflow 是 MagicMock，不会执行真实逻辑，需手动设置 side_effect
        mock_workflow.poll_and_process_clone.side_effect = VoiceCloneNotFoundError("Voice clone nonexistent not found")

        _mock_db_module.SessionLocal.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        result = process_voice_clone("nonexistent")

        assert result["ok"] is False
        assert "not found" in result["error"].lower()
        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()


# ── 超时场景 ──────────────────────────────────────────────


class TestProcessVoiceCloneTimeout:
    """测试超时场景。"""

    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
    @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
    def test_process_voice_clone_timeout_retries(
        self,
        mock_repo_cls: MagicMock,
        mock_workflow_cls: MagicMock,
        mock_cosy_cls: MagicMock,
    ) -> None:
        """超时时调用 self.retry() 进行重试，Retry 异常向上传播。"""
        mock_session = MagicMock()
        mock_repo = MagicMock()
        mock_workflow = MagicMock()

        mock_repo_cls.return_value = mock_repo
        mock_workflow_cls.return_value = mock_workflow

        mock_workflow.poll_and_process_clone.side_effect = CosyVoiceTimeoutError("任务超时")

        _mock_db_module.SessionLocal.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        # mock self.retry：Celery __call__ 传入的 self 是 task 对象本身
        with patch.object(process_voice_clone, "retry") as mock_retry:
            mock_retry.side_effect = Retry("retrying")

            with pytest.raises(Retry):
                process_voice_clone("profile-123")

            mock_retry.assert_called_once()

        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()


# ── 失败场景 ──────────────────────────────────────────────


class TestProcessVoiceCloneFailure:
    """测试失败场景。"""

    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
    @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
    def test_process_voice_clone_cosyvoice_error(
        self,
        mock_repo_cls: MagicMock,
        mock_workflow_cls: MagicMock,
        mock_cosy_cls: MagicMock,
    ) -> None:
        """CosyVoice 错误：profile 标记为 failed。"""
        mock_session = MagicMock()
        mock_repo = MagicMock()
        mock_workflow = MagicMock()

        mock_repo_cls.return_value = mock_repo
        mock_workflow_cls.return_value = mock_workflow

        mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("克隆失败")

        _mock_db_module.SessionLocal.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        result = process_voice_clone("profile-123")

        assert result["ok"] is False
        assert "克隆失败" in result["error"]
        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()

    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
    @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
    def test_process_voice_clone_unexpected_error(
        self,
        mock_repo_cls: MagicMock,
        mock_workflow_cls: MagicMock,
        mock_cosy_cls: MagicMock,
    ) -> None:
        """意外异常：profile 标记为 failed。"""
        mock_session = MagicMock()
        mock_repo = MagicMock()
        mock_workflow = MagicMock()

        mock_repo_cls.return_value = mock_repo
        mock_workflow_cls.return_value = mock_workflow

        mock_workflow.poll_and_process_clone.side_effect = RuntimeError("未知错误")

        _mock_db_module.SessionLocal.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        result = process_voice_clone("profile-123")

        assert result["ok"] is False
        assert "未知错误" in result["error"]
        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()

    @patch("worker_app.tasks.voice_clone.CosyVoiceService")
    @patch("worker_app.tasks.voice_clone.VoiceCloneWorkflowService")
    @patch("worker_app.tasks.voice_clone.SQLAlchemyVoiceCloneProfileRepository")
    def test_process_voice_clone_no_task_id(
        self,
        mock_repo_cls: MagicMock,
        mock_workflow_cls: MagicMock,
        mock_cosy_cls: MagicMock,
    ) -> None:
        """metadata 中没有 cosyvoice_task_id 时返回 failed。"""
        mock_session = MagicMock()
        mock_repo = MagicMock()
        mock_workflow = MagicMock()

        mock_repo_cls.return_value = mock_repo
        mock_workflow_cls.return_value = mock_workflow

        mock_workflow.poll_and_process_clone.side_effect = CosyVoiceError("missing task_id")

        _mock_db_module.SessionLocal.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        result = process_voice_clone("profile-123")

        assert result["ok"] is False
        assert "task_id" in result["error"]
        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()