你如何unit testing芹菜任务?

Celery文档提到在Django中testingCelery,但是如果不使用Django,则不会解释如何testingCelery任务。 你怎么做到这一点?

有可能使用任何unittest lib在那里同步testing任务。 芹菜任务时,我通常会做2个不同的testing会话。 第一个(正如我所build议的那样)是完全同步的,应该是确保algorithm做它应该做的。 第二次会议使用整个系统(包括经纪人),并确保我没有序列化问题或任何其他分布,通信问题。

所以:

from celery import Celery celery = Celery() @celery.task def add(x, y): return x + y 

而你的testing:

 from nose.tools import eq_ def test_add_task(): rst = add.apply(args=(4, 4)).get() eq_(rst, 8) 

希望有所帮助!

我使用这个:

 with mock.patch('celeryconfig.CELERY_ALWAYS_EAGER', True, create=True): ... 

文档: http : //docs.celeryproject.org/en/3.1/configuration.html#celery-always-eager

CELERY_ALWAYS_EAGER允许您同步运行您的任务,而且您不需要芹菜服务器。

取决于你想要testing的是什么。

  • 直接testing任务代码。 不要调用“task.delay(…)”,只需从unit testing中调用“task(…)”即可。
  • 使用CELERY_ALWAYS_EAGER 。 这将导致你的任务立即被称为“task.delay(…)”,所以你可以testing整个path(但不是任何asynchronous行为)。

unit testing

 import unittest from myproject.myapp import celeryapp class TestMyCeleryWorker(unittest.TestCase): def setUp(self): celeryapp.conf.update(CELERY_ALWAYS_EAGER=True) 

py.test夹具

 # conftest.py from myproject.myapp import celeryapp @pytest.fixture(scope='module') def celery_app(request): celeryapp.conf.update(CELERY_ALWAYS_EAGER=True) return celeryapp # test_tasks.py def test_some_task(celery_app): ... 

附录:使send_task尊重渴望

 from celery import current_app def send_task(name, args=(), kwargs={}, **opts): # https://github.com/celery/celery/issues/581 task = current_app.tasks[name] return task.apply(args, kwargs, **opts) current_app.send_task = send_task 

对于芹菜4这是:

 @override_settings(CELERY_TASK_ALWAYS_EAGER=True) 

由于设置名称已更改,如果select升级,则需要更新,请参阅

http://docs.celeryproject.org/en/latest/whatsnew-4.0.html#lowercase-setting-names

Celery 3.0开始 ,在Django中设置CELERY_ALWAYS_EAGER一种方法是:

 from django.test import TestCase, override_settings from .foo import foo_celery_task class MyTest(TestCase): @override_settings(CELERY_ALWAYS_EAGER=True) def test_foo(self): self.assertTrue(foo_celery_task.delay())