Сельдерей задачи не бросая исключение в тестах Django



У меня есть пара заданий на сельдерей, которые включены в мои тесты Django. К сожалению, исключения не создаются, когда задачи вызываются через .delay () . Я устанавливаю CELERY_ALWAYS_EAGER в True.



Tasks.py



import celeryapp as app

@app.task()
def exception_task():
print 'CELERY_ALWAYS_EAGER:', app.conf['CELERY_ALWAYS_EAGER']
raise Exception('foo')


Tests.py



def test_exception_in_task(self):
from tasks import exception_task
exception_task.delay()


Вывод



CELERY_ALWAYS_EAGER: True
.
----------------------------------------------------------------------
Ran 1 test in 0.686s


При удалении .задержка тест завершается с ошибкой, как исключено:



ERROR: test_exception_in_task
Exception: foo


Версии



celery==3.1.4
Django==1.6.4
726   2  

2 ответов:

Кажется мне дополнительно пришлось установить CELERY_EAGER_PROPAGATES_EXCEPTIONS в значение true.

Под celery 4.0 мне пришлось использовать CELERY_TASK_EAGER_PROPAGATES

Comments

    Ничего не найдено.