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

为什么直接用 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

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

• 安装命令必须带 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 模式。

python-parallelization

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

下载

• 最简方案：设 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，并完整复制关键配置。

立即学习“Python免费学习笔记（深入）”；

• 必须显式传 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 必须严格匹配。漏掉任意一环，测试就变成“看起来过了，其实没跑”。