Flask集成Celery用Pytest测试时countdown致批量用例失败

批量执行测试失败的核心矛盾是fixture作用域不匹配，加上ETA（countdown是ETA的语法糖）任务的调度特性：

• celery.contrib.pytest提供的默认celery_worker是函数级fixture，每个测试用例跑完就会关停当前worker，下一个用例再重新启动新worker
• 配置的Redis broker、result backend、celery_app都是session级，跨测试用例持久存在
• 带countdown的任务不会立刻投递给worker，会先存在Redis的定时调度集合里，等倒计时结束才会进入待执行队列。第一个测试收尾、worker关停的间隙，调度集合里的残留任务、加上第二个测试新提交的倒计时任务，很容易出现新worker启动时错过调度触发点、或者任务被已经销毁的旧worker抢占导致结果丢失的情况
• 不带countdown的任务提交后直接进入待执行队列，新worker启动后立刻就能拉到任务执行，不存在调度窗口错位的问题，所以批量运行可以正常通过
• 单独跑单个测试时，worker从启动到关停的完整生命周期完全覆盖2秒倒计时，没有跨测试的启停打断，所以可以正常通过

方案1：统一用session级worker（最推荐，最接近真实运行环境） #

覆盖默认的celery_worker fixture，把它的作用域提升到session，整个测试会话只启动一次worker，彻底避免测试间启停带来的调度错位：

import pytest
from celery.contrib.pytest import celery_worker as base_celery_worker

pytest_plugins = ('celery.contrib.pytest', )

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'redis://localhost:8001',
        'result_backend': 'redis://localhost:8001',
        'task_always_eager': False,
        'worker_prefetch_multiplier': 1,
        # 缩短Redis队列的可见性超时，避免异常断开的任务长时间锁死
        'broker_transport_options': {'visibility_timeout': 5}
    }

@pytest.fixture(scope='session')
def celery_worker(base_celery_worker):
    yield from base_celery_worker

方案2：测试前清空队列（适合需要测试间隔离配置的场景） #

如果必须用函数级worker做配置隔离，加一个自动执行的fixture，每个测试启动前清空Redis里残留的调度任务和队列消息：

@pytest.fixture(autouse=True)
def purge_celery_tasks(celery_app):
    with celery_app.connection() as conn:
        channel = conn.default_channel
        # 清空所有任务队列
        for queue in celery_app.conf.task_queues:
            channel.queue_purge(queue.name)
        # 清空Redis里存的未调度ETA任务、未确认任务残留
        channel.client.delete('unacked_mutex', 'unacked_index')
    yield

方案3：单元测试场景用eager模式（最快，不需要启动真实worker） #

如果测试目标只是验证任务本身的逻辑正确性，不需要验证Celery的分布式调度流程，可以给测试用例开启eager模式，任务会直接在测试进程内同步执行，跳过所有调度逻辑：

@pytest.fixture
def celery_eager(celery_app):
    old_eager = celery_app.conf.task_always_eager
    old_propagate = celery_app.conf.task_eager_propagates
    celery_app.conf.task_always_eager = True
    celery_app.conf.task_eager_propagates = True
    yield
    celery_app.conf.task_always_eager = old_eager
    celery_app.conf.task_eager_propagates = old_propagate

测试时传入celery_eager fixture即可，此时countdown参数会被忽略，任务立刻执行，不会有调度相关的问题。

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