xiaoxia-saas/tests/conftest.py

from __future__ import annotations

import os
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

# 必须在任何 app 模块导入之前设置，否则 pydantic Settings 验证失败
os.environ.setdefault("JWT_SECRET_KEY", "test-secret-key-for-all-tests")
os.environ.setdefault("USE_IN_MEMORY_DB", "True")


# ── Celery 全局 mock ──────────────────────────────────────────────────────
# CI 环境没有 Redis，所有 Celery 异步任务都 mock 掉，避免连接超时报错
# 集成测试只测 API 层逻辑（参数校验、权限、DB 操作），异步任务由 worker 单测覆盖

from unittest.mock import MagicMock, patch


def _mock_celery_task():
    """全局 mock Celery 任务的 delay/apply_async/send_task 方法。"""
    from celery import Celery, Task

    # 保存原始方法
    _orig_delay = Task.delay
    _orig_apply_async = Task.apply_async
    _orig_send_task = Celery.send_task

    def _mock_delay(self, *args, **kwargs):
        mock_result = MagicMock()
        mock_result.id = "mock-task-id"
        mock_result.state = "PENDING"
        mock_result.ready.return_value = False
        mock_result.get.return_value = None
        return mock_result

    def _mock_apply_async(self, *args, **kwargs):
        return _mock_delay(self, *args, **kwargs)

    def _mock_send_task(self, name, *args, **kwargs):
        mock_result = MagicMock()
        mock_result.id = f"mock-{name}"
        mock_result.state = "PENDING"
        mock_result.ready.return_value = False
        mock_result.get.return_value = None
        return mock_result

    Task.delay = _mock_delay
    Task.apply_async = _mock_apply_async
    Celery.send_task = _mock_send_task


# 在任何 app 模块导入之前就 patch 掉
_mock_celery_task()