无法在 Django 测试中同步调用 celery 任务
Can't invoke celery task in Django tests synchronously
我想在我的 Django 测试期间同步调用 celery 任务,而不需要 运行 celery worker。为此,我在 settings.py 中指定了 CELERY_ALWAYS_EAGER=True
,但它似乎不起作用。所以我决定将 override_settings 装饰器应用到看起来像这样的特定测试
@override_settings(CELERY_ALWAYS_EAGER=True, BROKER_BACKEND='memory',
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True)
def test_foo(self):
...
不幸的是,这个测试仍然在我的 celery worker 中调用任务。我可以缺少什么?具体来说,我将 Django 1.10 与 Celery 4.0.0 一起使用。
在celery 4.0配置参数已经改变,
在你的测试中试试这些,
@override_settings(
task_eager_propagates=True,
task_always_eager=True,
broker_url='memory://',
backend='memory'
)
我遇到了同样的问题,使用新的小写名称进行测试以及在默认的芹菜设置中解决了这个问题。
这是原始设置映射的新设置,
http://docs.celeryproject.org/en/latest/userguide/configuration.html#new-lowercase-settings
Celery 设置更改信息:
http://docs.celeryproject.org/en/latest/whatsnew-4.0.html#lowercase-setting-names
我想在我的 Django 测试期间同步调用 celery 任务,而不需要 运行 celery worker。为此,我在 settings.py 中指定了 CELERY_ALWAYS_EAGER=True
,但它似乎不起作用。所以我决定将 override_settings 装饰器应用到看起来像这样的特定测试
@override_settings(CELERY_ALWAYS_EAGER=True, BROKER_BACKEND='memory',
CELERY_EAGER_PROPAGATES_EXCEPTIONS=True)
def test_foo(self):
...
不幸的是,这个测试仍然在我的 celery worker 中调用任务。我可以缺少什么?具体来说,我将 Django 1.10 与 Celery 4.0.0 一起使用。
在celery 4.0配置参数已经改变,
在你的测试中试试这些,
@override_settings(
task_eager_propagates=True,
task_always_eager=True,
broker_url='memory://',
backend='memory'
)
我遇到了同样的问题,使用新的小写名称进行测试以及在默认的芹菜设置中解决了这个问题。
这是原始设置映射的新设置,
http://docs.celeryproject.org/en/latest/userguide/configuration.html#new-lowercase-settings
Celery 设置更改信息:
http://docs.celeryproject.org/en/latest/whatsnew-4.0.html#lowercase-setting-names