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 任务的底层函数签名为 (self, profile_id)，
CosyVoiceService 在 voice_clone.py 中被实例化传入 workflow，必须 mock 防止真实初始化。

跨环境兼容（_resolve_task）：
  不同 Celery 版本 / Python 版本 / 是否有 active Celery app，task 对象形态不同：
    1) Celery Proxy（LocalProxy/LazyProxy）：import 结果是代理对象，调用
       _get_current_object() 可能抛 RuntimeError（无 active context），必须 try 保护。
       成功取到真实 Task 实例后，使用 bound method .run。
    2) Celery Task 实例（bind=True 时 @task 返回的典型形态）：直接有 .run/.retry。
    3) 原始函数（某些环境装饰器未生效或 patch 时序问题）：需手动传 mock_self。
  统一返回 (callable, mock_self, real_task)，调用方不需要重复解析。
"""

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


def _resolve_task(task_obj):
    """解析 Celery 任务对象，兼容 Proxy / Task 实例 / 原始函数三种形态。

    所有分支均做异常保护，避免因 Celery Proxy 在无 app context 时抛错导致测试挂掉。

    Returns:
        tuple: (callable, mock_self, real_task)
            - callable: 最终执行用的可调用对象
            - mock_self: 仅原始函数分支需要手动传入 mock self；其他分支为 None
            - real_task: 真实 Task 实例（Proxy 分支为 _get_current_object() 结果；
              Task 分支为 task_obj 本身；原始函数分支为 None）。用于 patch .retry。
    """
    # Case 1: Celery Proxy → 安全尝试 _get_current_object()
    if hasattr(task_obj, "_get_current_object"):
        try:
            real_task = task_obj._get_current_object()
            if real_task is not None and hasattr(real_task, "run"):
                return real_task.run, None, real_task
        except Exception:
            # 无 active app context 或 Proxy 未绑定，退化为其他分支处理
            pass

    # Case 2: Celery Task 实例（非 Proxy）
    if hasattr(task_obj, "run") and hasattr(task_obj, "retry"):
        return task_obj.run, None, task_obj

    # Case 3: 原始函数（装饰器未生效）
    return task_obj, MagicMock(), None


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


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

    @patch("worker_app.tasks.voice_clone.SessionLocal")
    @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,
        mock_session_local: 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_session_local.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        func, mock_self, _ = _resolve_task(process_voice_clone)
        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
        result = func(*args)

        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.SessionLocal")
    @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,
        mock_session_local: 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

        mock_workflow.poll_and_process_clone.side_effect = VoiceCloneNotFoundError(
            "Voice clone nonexistent not found"
        )

        mock_session_local.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        func, mock_self, _ = _resolve_task(process_voice_clone)
        args = (mock_self, "nonexistent") if mock_self is not None else ("nonexistent",)
        result = func(*args)

        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.SessionLocal")
    @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,
        mock_session_local: 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_session_local.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        func, mock_self, real_task = _resolve_task(process_voice_clone)

        if mock_self is not None:
            # 原始函数环境：retry 在 mock_self 上
            mock_self.retry.side_effect = Retry("retrying")
            with pytest.raises(Retry):
                func(mock_self, "profile-123")
            mock_self.retry.assert_called_once()
        else:
            # Proxy/Task 环境：retry 在 Task 实例上。用 _resolve_task 返回的 real_task，
            # 避免再次 _get_current_object() 在无 context 时抛 AttributeError。
            retry_target = real_task if real_task is not None else process_voice_clone
            mock_retry = MagicMock()
            mock_retry.side_effect = Retry("retrying")
            with patch.object(retry_target, "retry", mock_retry):
                with pytest.raises(Retry):
                    func("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.SessionLocal")
    @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,
        mock_session_local: 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_session_local.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        func, mock_self, _ = _resolve_task(process_voice_clone)
        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
        result = func(*args)

        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.SessionLocal")
    @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,
        mock_session_local: 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_session_local.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        func, mock_self, _ = _resolve_task(process_voice_clone)
        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
        result = func(*args)

        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.SessionLocal")
    @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,
        mock_session_local: 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_session_local.return_value = mock_session

        from worker_app.tasks.voice_clone import process_voice_clone

        func, mock_self, _ = _resolve_task(process_voice_clone)
        args = (mock_self, "profile-123") if mock_self is not None else ("profile-123",)
        result = func(*args)

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