unittest 芹菜任务 assertRaises

Question

我有一些芹菜任务。我想通过单元测试来测试它。

我正在做的事情非常类似于：

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

我觉得奇怪的是：

此断言失败，因为 ValueError not raised，但在执行过程中，我可以看到此芹菜任务导致的 ValueError。

我不确定，但看起来 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 等系统的实际 'alive' 部分测试代码单元实际上是不正确的。这是从 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))

unittest 芹菜任务 assertRaises

unittest celery task assertRaises

python

unit-testing

assert

celery