unittest celery 任务 assertRaises

Question

我有一些芹菜任务。我想通过 unittest 对其进行测试。

我正在做一些非常相似的事情：

class TestMe(unittest.TestCase):
    def test_celery_task(self):
        self.assertRaises(ValueError, celery_task.apply, args)

对我来说有什么奇怪的：

这个断言失败，因为ValueError not raised，但是在执行过程中我可以看到 ValueError 作为这个 celery 任务的结果。

我不确定，但看起来 assert 的检查速度快于 ValueError 的上升速度。 是否可以检查执行的芹菜任务的结果？ 或者如何测试？

Answer 1

这不可能。当您将 Celery 任务排入队列时，所发生的一切就是您将消息放入队列以供单独的进程提取；正是该进程运行任务并可能引发异常。

如果你想检查任务本身是否引发了 ValueError，那么你应该调用任务，而不是延迟函数：

self.assertRaises(ValueError, celery_task, args)

Answer 2

我在这里看到 3 个选项。

1) 尝试在 apply() 上调用 get()。以下是您将获得的：

class TestMe(unittest.TestCase):
    def test_celery_task(self):
        self.assertRaises(ValueError, celery_task.apply().get(), args)

2) 您可以通过将 'task_always_eager' 设置为 True 来启用 Eager 模式，但这并不能保证您的代码能够赶上。

3) 更好的选择是模拟 celery 任务。从单元测试的角度来看，用 celery 等系统的实际“活动”部分测试代码单元实际上是不正确的。 这是取自 celery testing documentation 的代码示例。

from pytest import raises

from celery.exceptions import Retry

# for python 2: use mock.patch from `pip install mock`.
from unittest.mock import patch

from proj.models import Product
from proj.tasks import send_order

class test_send_order:

    @patch('proj.tasks.Product.order')  # < patching Product in module above
    def test_success(self, product_order):
        product = Product.objects.create(
            name='Foo',
        )
        send_order(product.pk, 3, Decimal(30.3))
        product_order.assert_called_with(3, Decimal(30.3))

    @patch('proj.tasks.Product.order')
    @patch('proj.tasks.send_order.retry')
    def test_failure(send_order_retry, product_order):
        product = Product.objects.create(
            name='Foo',
        )

        # set a side effect on the patched method
        # so that it raises the error we want.
        product_order.side_effect = OperationalError()

        with raises(Retry):
            send_order(product.pk, 3, Decimal(30.6))