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

Это не может работать. Когда вы ставите в очередь задачу сельдерея, все, что происходит, это то, что вы помещаете сообщение в очередь для отдельного процесса, чтобы забрать; это тот процесс, который выполняет задачу и, потенциально, вызывает исключение.

Если вы хотите проверить, что сама задача возникает ValueError, то вы должны вызвать задачу, а не функция задержки:

self.assertRaises(ValueError, celery_task, args) 

Answer 2

Я вижу 3 варианта здесь.

1) Попробуйте позвонить get() на apply(). Вот что вы получите:

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

2) Вы можете включить режим нетерпеливый, установив «task_always_eager» в True, однако это не гарантирует, что ваш код будет в состоянии догнать.

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