Celery任务.delay方法非异步：pytest测试场景下的问题求助

问题核心 #

• 设置task_always_eager=True时，.delay()会同步执行任务，后续代码无法立即运行，违背异步测试的初衷
• 不设置该参数时任务代码无法执行，说明celery_worker fixture未正确配置异步运行环境

修复步骤 #

1. 重构celery_app和celery_worker fixture #

放弃依赖task_always_eager，改用标准测试worker配置：

import pytest
from celery import Celery
from celery.contrib.testing.worker import start_worker

@pytest.fixture(scope="module")
def celery_app():
    # 初始化测试用Celery应用，使用内存broker和rpc后端
    app = Celery(
        "test_tasks",
        broker="memory://",
        backend="rpc://"
    )
    # 关闭同步执行模式
    app.conf.task_always_eager = False
    # 关闭异常自动传播，保持异步隔离
    app.conf.task_eager_propagates = False
    return app

@pytest.fixture(scope="module")
def celery_worker(celery_app):
    # 启动本地测试worker
    worker = start_worker(celery_app, concurrency=1)
    worker.start()
    yield worker
    worker.stop()

2. 调整测试代码适配异步逻辑 #

异步执行后，后续代码会立即运行，可按需验证任务状态或等待结果：

def test_selected_rows_are_blocked(celery_worker):
    task_id, select_only = 0, ['a', 'b']
    # 提交异步任务
    task = tasks.update_table.delay(task_id, select_only)
    
    # 此处代码会立即执行，比如验证任务处于待执行状态
    assert task.state == 'PENDING'
    
    # 若需等待任务完成并获取结果，可调用get（会阻塞至任务结束）
    # result = task.get(timeout=5)
    # assert result == 预期结果

3. 关键配置解释 #

• memory:// broker：无需额外服务，测试环境轻量高效
• rpc:// backend：支持任务状态查询和结果获取
• task_eager_propagates = False：避免任务异常直接中断测试主线程，保证异步行为的真实性

验证方式 #

可以在Celery任务开头添加临时标记（比如写入一个临时文件、更新内存中的状态变量），测试后续代码执行时检查该标记是否存在，以此确认任务已启动且后续代码是异步执行的。

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