如何在Python中测试基于Celery的异步任务_利用celery.contrib.pytest-Python教程-PHP中文网

celery 5.0起移除了celery.contrib.pytest，需改用celery[pytest]包并显式配置pytest.ini或pyproject.toml启用celery.contrib.testing.pytest插件，否则celery_app、celery_worker等fixture无法加载。

如何在python中测试基于celery的异步任务_利用celery.contrib.pytest

为什么直接用 `celery.contrib.pytest` 会报 `ImportError: No module named 'celery.contrib.pytest'`

因为从 Celery 5.0 开始，celery.contrib.pytest 已被移除，不再随 Celery 发布。官方推荐改用独立包 celery[pytest]（即安装时显式启用 pytest 插件），但实际安装后仍需手动启用插件，且默认不自动注册 fixture。

运行 pip install "celery[pytest]" 后，插件代码在 celery.contrib.testing 下，而非旧路径
pytest 不会自动发现并加载该插件，必须在 pytest.ini 或 pyproject.toml 中显式启用
常见错误：只装了包、没配配置，然后调用 celery_worker fixture 报 fixture not found

如何正确启用并使用 `celery_worker` 和 `celery_app` fixture

核心是两步：安装 + 配置。缺一不可。

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

安装命令必须带 extras：pip install "celery[pytest]"（注意引号，避免 shell 解析失败）

在项目根目录加 pytest.ini，内容为：

[tool:pytest]
addopts = --tb=short
plugins = celery.contrib.testing.pytest

或用 pyproject.toml（Pytest ≥ 7.0）：

[tool.pytest.ini_options]
addopts = ["--tb=short"]
plugins = ["celery.contrib.testing.pytest"]

fixture 名字是 celery_app 和 celery_worker，不是 app 或 worker —— 拼错就找不到

测试异步任务时，为什么 `celery_worker` 执行完就退出，任务没跑？

因为 celery_worker fixture 默认启动的是「临时 worker」，它只处理同步发送的 task（即 .apply()），不监听 broker。想测真实异步流程，得自己控制 worker 生命周期或改用 eager 模式。

最简方案：设 task_always_eager=True，让 .delay() 等价于 .apply()，无需 worker —— 适合逻辑校验
真 worker 场景：用 celery_worker fixture 启动后，需显式调用 task.delay().get(timeout=3)，否则测试函数结束、worker 就停了
注意 timeout：Broker（如 Redis）连接慢时，.get() 可能超时，建议设 timeout=5 并捕获 celery.exceptions.TimeoutError
若任务依赖外部服务（如 DB），确保 celery_app 的 broker_url 和 result_backend 指向测试专用实例（如 redis://localhost:6379/15）

自定义 `celery_app` fixture 时，哪些参数容易漏掉？

直接复用默认 fixture 往往不够——比如你用了 SqlAlchemyTask 或自定义序列化器，就必须重定义 app fixture，并完整复制关键配置。

必须显式传 broker_url 和 result_backend，否则 fallback 到 memory://，导致结果无法读取
若任务用了 bind=True，要加 task_track_started=True，否则 task.status 始终是 PENDING
修改 task_serializer 或 result_serializer 后，务必同步更新 accept_content，否则 worker 拒收消息

示例重定义：

@pytest.fixture()
def celery_app(celery_config):
    app = Celery("test")
    app.config_from_object(celery_config)
    app.conf.update(
        broker_url="redis://localhost:6379/14",
        result_backend="redis://localhost:6379/14",
        task_track_started=True,
        accept_content=["json"],
        task_serializer="json",
        result_serializer="json",
    )
    yield app

Celery 的 pytest 集成表面简单，实则每层都有隐含约束：插件加载靠配置文件、worker 生命周期靠手动同步、序列化和 backend 必须严格匹配。漏掉任意一环，测试就变成“看起来过了，其实没跑”。