如何在 celery 中测试 on_failure答案

【问题标题】：How can I test on_failure in celery如何在 celery 中测试 on_failure
【发布时间】：2018-11-07 04:08:01
【问题描述】：

我的 celery 任务有一个基类，实现了 on_failure 方法。

在我的测试中，我修补了任务调用的方法之一，以引发异常，但从未调用过 on_faliure。

基类

class BaseTask(celery.Task):
    abstract = True 

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        print("error")

任务

@celery.task(bind=True, base=BaseTask)
def multiple(self, a, b):
    logic.method(a, b)

测试

@patch('tasks.logic.method')
def test_something(self, mock):
    # arrange
    mock.side_effect = NotImplementedError

    # act
    with self.assertRaises(NotImplementedError):
        multiple(1, 2)

当运行 celery 并引发异常时，一切正常。 CELERY_ALWAYS_EAGER 已激活。

如何让on_faliure 运行？

【问题讨论】：

我的解决方案不是按原样运行multiple，而是通过同步芹菜

标签： python unit-testing celery celery-task

【解决方案1】：

来自discussion on a issue in celery GitHub：on_failure 测试“已经在 Celery 级别完成（验证是否调用了 on_failure）” 和 “编写一个测试来测试你的 on_failure 所做的任何事情而是”。您可以在on_failure 方法中定义一个函数并对其进行测试，或者像类方法一样调用on_failure：

import TestCase
from billiard.einfo import ExceptionInfo

class TestTask(TestCase):
    def test_on_failure(self):
        "Testing on failure method"

        exc = Exception("Test")
        task_id = "test_task_id"
        args = ["argument 1"]
        kwargs = {"key": "value"}
        einfo = ExceptionInfo

        # call on_failure method
        multiple.on_failure(exc, task_id, args, kwargs, einfo)

        # assert something appened

ExceptionInfo 与 celery 使用的对象类型相同； multiple 是您在问题中定义的任务。

希望对你有帮助

【讨论】：