如何使用pytest测试Celery实例中app.task装饰的任务？

1. 单元测试：直接验证任务逻辑 #

用@app.task装饰的任务本质是被Celery包装后的函数，你可以直接调用任务对象（或其run()方法）跳过Celery异步机制，快速测试核心业务逻辑，不需要启动worker或依赖消息队列。

举个任务定义示例：

# tasks.py
from celery import Celery

app = Celery('myapp', broker='redis://localhost:6379/0')

@app.task
def add(x, y):
    return x + y

对应的pytest测试用例：

# test_tasks.py
from tasks import add

def test_add_task():
    # 直接调用任务对象，等同于调用add.run(1, 2)
    result = add(1, 2)
    assert result == 3

2. 集成测试：模拟异步执行流程 #

如果要测试任务的提交、状态跟踪等Celery异步特性，可以配置Celery为测试模式，让任务同步执行，无需额外启动worker。

方式1：单测试用例配置 #

# test_tasks.py
from tasks import app, add

def test_add_task_async():
    # 开启同步执行+异常直接抛出
    app.conf.update(
        CELERY_ALWAYS_EAGER=True,
        CELERY_EAGER_PROPAGATES_EXCEPTIONS=True
    )
    
    # 提交任务
    task_result = add.delay(3, 4)
    
    # 直接验证结果和状态
    assert task_result.successful()
    assert task_result.get() == 7

方式2：全局fixture复用配置 #

在conftest.py中定义全局fixture，避免重复配置：

# conftest.py
import pytest
from tasks import app

@pytest.fixture(autouse=True)
def celery_test_env():
    app.conf.update(
        CELERY_ALWAYS_EAGER=True,
        CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
        CELERY_BROKER_URL='memory://',  # 用内存broker，不依赖外部服务
        CELERY_RESULT_BACKEND='cache+memory://'
    )

之后测试用例可以简化：

# test_tasks.py
from tasks import add

def test_add_task_async_simple():
    task_result = add.delay(5, 6)
    assert task_result.get() == 11
    assert task_result.state == 'SUCCESS'

3. 测试带Celery特性的任务 #

如果任务用到了重试、绑定实例等Celery特性，直接在测试中模拟或验证即可：

示例：测试绑定实例的重试任务 #

任务定义：

# tasks.py
@app.task(bind=True)
def retry_task(self, x):
    if x < 0:
        self.retry(countdown=1)
    return x * 2

测试用例：

# test_tasks.py
import pytest
from tasks import retry_task
from celery.exceptions import Retry
from unittest.mock import patch

def test_retry_task_normal_case():
    assert retry_task(4) == 8

def test_retry_task_trigger_retry():
    # 验证重试逻辑触发
    with pytest.raises(Retry):
        retry_task(-1)

def test_retry_task_with_mock():
    # 用mock验证重试参数
    with patch.object(retry_task, 'retry') as mock_retry:
        retry_task(-1)
        mock_retry.assert_called_once_with(countdown=1)

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