【问题标题】:Celery tasks not throwing exception in Django TestsCelery 任务在 Django 测试中没有抛出异常
【发布时间】: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


    【解决方案1】:

    似乎我还必须将 CELERY_EAGER_PROPAGATES_EXCEPTIONS 设置为 True。

    【讨论】:

    • 这仍然适用于 Celery 5。
    【解决方案2】:

    在celery 4.0下,我不得不使用CELERY_TASK_EAGER_PROPAGATES

    【讨论】:

      猜你喜欢
      • 2020-12-21
      • 2013-07-17
      • 2018-09-11
      • 1970-01-01
      • 1970-01-01
      • 2021-03-26
      • 1970-01-01
      • 1970-01-01
      • 2023-01-25
      相关资源
      最近更新 更多