Celery单元测试：如何验证配置autoretry_for的任务可自动重试

你当前直接调用my_task()函数的方式只会执行函数本身的代码，不会触发Celery的自动重试装饰器逻辑，要验证自动重试配置生效，需要走完整的Celery任务执行流程，操作步骤如下：

1 调整测试环境配置 #

在测试配置中新增/保留以下参数：

• 保留ALWAYS_EAGER = True（Celery 5+版本对应配置为task_always_eager = True），让任务同步执行方便测试
• 新增EAGER_PROPAGATES_EXCEPTIONS = False（Celery 5+版本对应配置为task_eager_propagates_exceptions = False），避免任务抛出的异常直接向上传递打断重试流程

2 编写测试用例 #

核心逻辑为Mock会抛出异常的业务函数，通过Celery提供的任务触发方法执行任务，验证重试次数符合配置：

from unittest import mock

@mock.patch("替换为你代码里function_call_that_can_raise_exception的导入路径")
@mock.patch.object(my_task, 'retry')
def test_my_task_auto_retry(self, mock_retry, mock_faulty_func):
    # 让业务函数一直抛出异常，触发重试逻辑
    mock_faulty_func.side_effect = Exception("测试异常，触发重试")

    # 使用apply方法触发任务，走Celery完整任务生命周期
    my_task.apply(args=(12,))

    # 验证重试次数符合配置：max_retries=3，所以会触发3次重试
    self.assertEqual(mock_retry.call_count, 3)
    # 可选：验证重试参数是否符合配置，比如退避时间为3秒
    mock_retry.assert_called_with(exc=mock.ANY, countdown=3, max_retries=3)

如果你需要同时验证最后一次重试的业务处理逻辑，可以用以下写法：

from unittest import mock

@mock.patch("替换为你代码里function_call_that_can_raise_exception的导入路径")
def test_my_task_last_retry_logic(self, mock_faulty_func):
    mock_faulty_func.side_effect = Exception("测试异常")

    # 执行任务
    result = my_task.apply(args=(12,))

    # 确认任务最终执行失败
    self.assertTrue(result.failed())
    # 确认重试次数达到配置的最大值3次
    self.assertEqual(result.task.request.retries, 3)
    # 这里补充你自己的最后一次重试业务逻辑断言

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