I have a couple of celery tasks that are included in my Django tests. Unfortunately exceptions are not thrown when tasks are invoked via .delay(). I am setting CELERY_ALWAYS_EAGER to 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()
Output
CELERY_ALWAYS_EAGER: True
.
----------------------------------------------------------------------
Ran 1 test in 0.686s
When removing the .delay the test exits with an error as excpected:
ERROR: test_exception_in_task
Exception: foo
Versions
celery==3.1.4
Django==1.6.4
Yes you can. Celery is a generic asynchronous task queue. In place of "django_project" you would point to your module.
Celery makes it easier to implement the task queues for many workers in a Django application.
Seems I additionally had to set CELERY_EAGER_PROPAGATES_EXCEPTIONS to True.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With