你如何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())
