15

我正在尝试在 Django 的单元测试框架中测试一些 celery 功能，但是每当我尝试检查 AsyncResult 时，测试的行为就像它从未开始一样。

我知道这段代码在使用 RabbitMQ 的真实环境中工作，所以我只是想知道为什么在使用测试框架时它不起作用。

这是一个例子：

@override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS = True,
                   CELERY_ALWAYS_EAGER = True,
                   BROKER_BACKEND = 'memory',)
def test_celery_do_work(self):
    result = myapp.tasks.celery_do_work.AsyncResult('blat')
    applied_task = myapp.tasks.celery_do_work.apply_async((), task_id='blat')
    applied_task.wait()
    # THIS SUCCEEDS
    self.assertTrue(applied_task.successful())
    # THIS FAILS
    self.assertTrue(result.successful())

使用 ALWAYS_EAGER 选项是否会禁用 AsyncResult 功能，因为它会立即执行？如果是这样，有没有办法对 AsyncResult 状态检查进行单元测试？如果我尝试去掉 ALWAYS_EAGER 选项，测试永远不会运行，所以我很茫然。

谢谢！

4

1 回答 1

20

当CELERY_ALWAYS_EAGER是True时，调用apply_async()实际上被替换为apply()。返回的结果是一个EagerResult，它已经包含了你的任务的结果。

所以，是的，设置ALWAYS_EAGER = True会禁用整个AsyncResult功能。整个异步过程被绕过，实际上没有任务发送到代理，这就是为什么您无法通过AsyncResult.

CELERY_ALWAYS_EAGER = True当您测试只需要 Celery 结果的代码路径时使用，并且使用相同的方式处理 anEagerResult或AsyncResult.

如果需要，也可以使用 with 运行测试AsyncResult，CELERY_ALWAYS_EAGER = False但为此，您需要在调用测试用例中的任务之前启动一个 worker。然后，工人将能够执行您的任务，并且AsyncResult工作得很好。你可以看看django-celery-testworker似乎就是这样做的，虽然我没有测试过。

于 2013-06-17T20:09:10.123 回答