如何测试FastAPI端点中celery_client.send_task是否被调用？

问题背景 #

项目文件结构：

endpoints.py 
celery_conf.py

其中celery_conf.py负责Celery客户端的配置与实例化，endpoints.py导入该客户端并实现FastAPI端点：

# endpoints.py
from celery_conf import celery_client

@router.post(
    include_in_schema=True,
    status_code=status.HTTP_200_OK,
    name="some_name:post"
)
def foo_endpoint(
    item: PydanticModel, db: Session = Depends(get_database)
) -> dict:
    tmp = <some preprocessing of item>
    celery_client.send_task(...)
    return 200

需要编写pytest测试用例，验证端点中celery_client.send_task()是否被正确调用，已尝试使用pytest的patch但不清楚具体验证方式，现有测试代码框架：

client = TestClient(app) # 需传入FastAPI app实例
def test_endpoint():
    # patch where celery client is imported
    with patch('endpoints.celery_client') as mock_celery:
        client.post(url=app.url_path_for("some_name:post"), json={})
        ...

测试实现步骤 #

1. 确认patch路径：由于endpoints.py中导入了celery_client，所以需要patchendpoints.celery_client（你已选对路径）。
2. 获取send_task的mock对象：patch后的mock_celery是celery_client的替身，它的send_task属性是一个mock函数，可通过mock_celery.send_task访问。
3. 验证调用情况：使用unittest.mock的断言方法，验证send_task的调用状态与参数。

完整测试代码示例：

from fastapi.testclient import TestClient
from unittest.mock import patch
from your_app_module import app  # 替换为你的FastAPI app所在模块

client = TestClient(app)

def test_foo_endpoint_calls_send_task():
    with patch('endpoints.celery_client') as mock_celery:
        # 构造符合PydanticModel要求的请求数据
        test_data = {"key": "value"}  # 替换为你的模型对应字段
        # 发送匹配端点的POST请求
        response = client.post(
            url=app.url_path_for("some_name:post"),
            json=test_data
        )
        
        # 验证接口返回状态码正确
        assert response.status_code == 200
        
        # 验证send_task至少被调用一次
        mock_celery.send_task.assert_called()
        
        # 可选：验证send_task的调用参数是否符合预期
        # 示例：假设send_task被调用时传入任务名和处理后的tmp参数
        # mock_celery.send_task.assert_called_once_with("your_task_name", args=(expected_tmp_value,))

关键注意事项 #

• 请求方法匹配：端点是@router.post，测试时必须用client.post而非client.put。
• 请求数据格式：FastAPI接收JSON数据需用json参数，data参数用于表单数据。
• 常用mock断言方法：
  • assert_called()：验证函数至少被调用一次
  • assert_called_once()：验证函数恰好被调用一次
  • assert_called_with(*args, **kwargs)：验证函数调用参数完全匹配
  • assert_any_call(*args, **kwargs)：验证函数曾以指定参数被调用

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