单元测试芹菜任务断言

Question

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

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

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

对我来说很奇怪的是：

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

我不确定，但看起来断言的检查速度比 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来启用渴望模式，但这并不能保证您的代码能够赶上。

3）更好的选择是模拟芹菜任务。从单元测试的角度来看，用芹菜等系统的实际"活动"部分测试代码单元实际上是不正确的。下面是从芹菜测试文档中获取的代码示例。

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))