【发布时间】:2015-01-17 17:35:07
【问题描述】:
我的 Django 测试中包含了几个 celery 任务。不幸的是,通过 .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
当删除 .delay 时,测试退出并出现预期的错误:
ERROR: test_exception_in_task
Exception: foo
版本
celery==3.1.4
Django==1.6.4
【问题讨论】:
标签: python django testing celery celery-task