相关疑难解决方法(0)

我正在尝试为我们的django-celery项目提出测试方法.我已经阅读了文档中的注释,但它并没有让我对实际操作有什么了解.我并不担心测试实际守护进程中的任务,只是我的代码的功能.主要是我想知道:

1. 我们如何task.delay()在测试期间绕过(我尝试过设置,CELERY_ALWAYS_EAGER = True但没有区别)？
2. 我们如何在不实际更改settings.py的情况下使用推荐的测试设置(如果这是最佳方式)？
3. 我们还可以使用manage.py test或者我们必须使用自定义跑步者吗？

总的来说,使用芹菜进行测试的任何提示或技巧都会非常有用.

我试图在Django的单元测试框架中测试一些芹菜功能,但每当我尝试检查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选项,测试永远不会运行,所以我不知所措.

谢谢!