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