unittest芹菜任务assertRaises

Question

我有一些芹菜任务。我想通过unittest测试它。

我做的非常相似的东西：

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）更好的选择是嘲笑芹菜任务。从单元测试的角度来看，用芹菜等系统中实际的“活着”部分来测试代码单元并不是真实的。 这是一个取自芹菜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))