在芹菜中对AsyncResult进行单元测试

Question

我正在尝试在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选项，则测试将永远不会运行，因此我一头雾水。

Answer 1

如果CELERY_ALWAYS_EAGER为is True，则对的呼叫apply_async()实际上被替换为apply()。返回的结果是EagerResult，其中已经包含任务的结果。

因此，是的，设置ALWAYS_EAGER = True将禁用整个AsyncResult功能。整个异步过程将被绕过，并且实际上没有任何任务发送到代理，这就是为什么您无法通过检索结果的原因AsyncResult。

CELERY_ALWAYS_EAGER = True在测试仅需要Celery结果的代码路径时使用，并与EagerResult或一起使用相同的方式AsyncResult。

如果需要，还有运行与测试的方式AsyncResult也与CELERY_ALWAYS_EAGER = False，但对于这一点，你需要调用任务在你的测试用例之前启动工作。然后，工作人员将能够执行您的任务，并且AsyncResult可以正常工作。您可以看一下django-celery-testworker，它似乎就是这样做的，尽管我尚未对其进行测试。