FastAPI+Beanie项目中异步辅助函数的测试方案咨询

一、单独单元测试异步函数（聚焦逻辑本身） #

因为_create_product是独立的异步辅助函数，不需要依赖FastAPI后台任务上下文，直接单独测试是最高效的方式，能精准验证函数逻辑。

测试步骤： #

• 初始化测试用Beanie连接：使用独立的测试数据库（比如本地MongoDB的测试库，或内存版mongomock），测试前初始化Beanie，测试后清理数据。
• Mock外部依赖：用unittest.mock或pytest-mock模拟create_stripe_product函数，返回预设的Stripe产品对象，避免调用真实API。
• 准备测试数据：创建并保存Plan文档实例，作为函数入参。
• 调用函数并断言：验证返回的FixedProduct字段正确，同时检查Beanie数据库中是否成功写入Product文档。

代码示例（基于pytest）： #

import pytest
from unittest.mock import Mock
from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
# 替换成你的模块路径
from your_app.models import Plan, Product
from your_app.tasks import _create_product

@pytest.fixture(scope="module")
async def test_db():
    # 连接测试数据库
    client = AsyncIOMotorClient("mongodb://localhost:27017/fastapi_test")
    await init_beanie(database=client.fastapi_test, document_models=[Plan, Product])
    yield
    # 测试结束后清理所有数据
    await Product.delete_all()
    await Plan.delete_all()
    client.close()

@pytest.mark.asyncio
async def test_create_product_logic(test_db):
    # 1. 准备测试用Plan
    test_plan = Plan(name="基础版", price=9.99)
    await test_plan.save()

    # 2. Mock Stripe API调用
    mock_stripe_product = Mock()
    mock_stripe_product.id = "stripe_test_123"
    mock_create_stripe = Mock(return_value=mock_stripe_product)

    # 3. 调用待测试函数
    result_product = await _create_product(test_plan, "sk_test_xxx", mock_create_stripe)

    # 4. 断言结果正确性
    assert result_product.stripe_id == "stripe_test_123"
    assert result_product.plan.id == test_plan.id

    # 5. 验证数据库写入成功
    saved_product = await Product.find_one(Product.stripe_id == "stripe_test_123")
    assert saved_product is not None
    assert saved_product.plan.id == test_plan.id

    # 6. 验证Stripe函数被正确调用
    mock_create_stripe.assert_called_once_with("sk_test_xxx")

二、集成测试（验证后台任务触发） #

如果需要验证FastAPI路由确实触发了后台任务中的_create_product，可以结合TestClient做集成测试：

测试思路： #

• 用FastAPI的TestClient发送触发后台任务的请求。
• 替换_create_product为带调用追踪的mock函数，验证它被正确触发。
• 注意：同步TestClient不会自动等待异步后台任务完成，需要手动触发事件循环等待任务执行。

代码示例： #

import pytest
import asyncio
from fastapi.testclient import TestClient
# 替换成你的模块路径
from your_app.main import app
from your_app.tasks import _create_product

@pytest.mark.asyncio
async def test_background_task_trigger():
    client = TestClient(app)
    call_count = 0

    # 1. 封装原函数，添加调用计数
    original_func = _create_product
    async def tracked_create_product(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        return await original_func(*args, **kwargs)

    # 2. 替换模块中的函数
    import your_app.tasks
    your_app.tasks._create_product = tracked_create_product

    # 3. 发送请求触发路由
    response = client.post("/plans", json={"name": "专业版", "price": 19.99})
    assert response.status_code == 200

    # 4. 等待后台任务执行（根据实际情况调整等待时间）
    await asyncio.sleep(0.2)

    # 5. 验证函数被调用
    assert call_count == 1

    # 6. 恢复原函数
    your_app.tasks._create_product = original_func

关键注意事项 #

• 使用内存MongoDB：可以用mongomock替代真实MongoDB，测试更快且无需本地数据库，只需把AsyncIOMotorClient换成mongomock.AsyncMongoClient。
• 严格隔离测试环境：测试数据库和生产数据库必须分开，避免测试数据污染生产环境。
• 避免真实外部调用：所有第三方API（比如Stripe）必须mock，防止产生真实费用或触发API限制。

内容的提问来源于stack exchange，提问作者override