Сельдерей задачи не бросая исключение в тестах 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
2 ответов:
Кажется мне дополнительно пришлось установить CELERY_EAGER_PROPAGATES_EXCEPTIONS в значение true.
Под celery 4.0 мне пришлось использовать CELERY_TASK_EAGER_PROPAGATES
Comments