如何在Celery中测试on_failure方法的触发逻辑？

我之前也碰到过一模一样的问题，核心原因其实很简单：直接调用Celery任务函数（比如multiple(1,2)）并不会触发Celery任务的完整生命周期流程，包括on_failure这类钩子方法的调用。哪怕你开启了CELERY_ALWAYS_EAGER，直接调用函数本质上只是执行了任务里的原生Python代码，没有经过Celery的任务封装逻辑。

为什么直接调用不行？ #

当你用@celery.task装饰器定义任务后，multiple这个变量是一个任务实例的代理对象，但直接调用它时，Celery只会执行装饰器包裹的原函数逻辑，不会触发on_failure、on_success这些生命周期钩子——这些钩子只有在通过Celery的任务执行机制（比如apply()、apply_async()）调用时才会被触发，哪怕是在eager同步模式下。

解决办法：用Celery任务的apply()或apply_async()方法调用 #

调整你的测试代码，改用apply()（同步执行，非常适合测试场景）来调用任务，这样就能完整触发任务的生命周期，包括on_failure的调用。同时确保你的测试环境中CELERY_ALWAYS_EAGER确实被设置为True（不过即使不手动设置，apply()默认也是同步执行的）。

修改后的测试代码示例 #

如果你需要验证异常抛出，同时确保on_failure被执行：

@patch('tasks.logic.method')
def test_something(self, mock):
    # arrange
    mock.side_effect = NotImplementedError
    # act
    # 用apply()调用任务，而非直接调用函数
    result = multiple.apply(args=(1, 2))
    
    # 验证任务执行失败
    self.assertTrue(result.failed())
    
    # 通过get()获取结果时会抛出原异常，满足你的断言需求
    with self.assertRaises(NotImplementedError):
        result.get()
    
    # 此时on_failure已经被执行了，如果需要验证它的行为（比如打印的内容），可以捕获输出
    with self.assertLogs(level='INFO') as log_capture:
        multiple.apply(args=(1, 2))
        self.assertIn('error', log_capture.output[0])

额外确认：你的BaseTask配置是正确的 #

你的BaseTask已经设置了abstract = True，这一点做得很对——Celery要求抽象基类才能被作为任务的base参数，这样才能正确继承生命周期方法，这部分不需要调整。

总结一下，核心就是不要直接调用任务函数，而是用Celery提供的apply()或apply_async()方法触发任务执行流程，这样on_failure钩子才会被正常调用。

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