芹菜文档mentions testing Celery within Django但不解释如何测试芹菜任务，如果你不使用Django。你怎么做到这一点？你如何测试芹菜任务？

2012-08-22 DavidM

的链接的文档已经消失，似乎并没有在当前的Celery 3.1文档中取代。 – keturn

@打开[使用Celery 2.5进行单元测试]的文档（http://docs.celeryproject.org/en/2.5/django/unit-testing.html）仍然可用 – Merwan

链接到Django文档已损坏。 – sheldonkreger

可以使用任何unittest lib在那里同步测试任务。在使用芹菜任务时，我通常会做2个不同的测试会话。第一个（正如我所建议的那样）是完全同步的，应该是确保算法完成它应该做的工作。第二场会议使用整个系统（包括经纪人），并确保我没有序列化问题或任何其他分布，通信问题。

所以：

from celery import Celery 

celery = Celery() 

@celery.task 
def add(x, y): 
    return x + y

而且测试：

from nose.tools import eq_ 

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

希望帮助！

来源

2012-08-22 20:33:59 FlaPer87

除了使用HttpDispatchTask的任务外， http://docs.celeryproject.org/en/latest/userguide/remote-tasks.html我必须设置celery.conf.CELERY_ALWAYS_EAGER = True，但即使同时设置了celery.conf.CELERY_IMPORTS =（'celery.task。 http'）测试失败NotRegistered：celery.task.http.HttpDispatchTask – DavidM

奇怪，你确定你没有一些导入问题？这[测试]（http://pastebin.com/TXkZxb6t）的作品（注意，我假装的响应，所以它返回芹菜期望）。此外，CELERY_IMPORTS中定义的模块将在[工作程序初始化]期间导入（https://github.com/celery/celery/blob/master/celery/loaders/base.py#L107），为了避免这种情况，我建议你可以调用'celery.loader.import_default_modules（）'。 – FlaPer87

我也建议你看看[这里]（https://github.com/celery/celery/blob/master/celery/tests/tasks/test_http.py）。它嘲笑http请求。不知道它是否有帮助，我想你想测试一个正在运行的服务，不是吗？ – FlaPer87

取决于你想要测试的是什么。

直接测试任务代码。不要调用“task.delay（...）”，只需在单元测试中调用“task（...）”即可。
使用CELERY_ALWAYS_EAGER。这会导致你的任务在你说“task.delay（...）”时被立即调用，因此你可以测试整个路径（但不是任何异步行为）。

来源

2012-08-22 20:29:48 slacy

CELERY_ALWAYS_EAGER链接在你的答案现在已经死了。 –

Celery 4.0+的文档链接在这里：http://docs.celeryproject.org/en/v4.1.0/userguide/configuration.html#task-always-eager – krassowski

我用这个：

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让你运行你的任务同步，你不需要芹菜服务器。

来源

2014-01-10 10:05:23 guettli

我认为这已经过时了 - 我得到了'ImportError：No module named celeryconfig'。 – Daenyth

我相信上面假设模块'celeryconfig.py'存在于一个包中。请参阅http://docs.celeryproject.org/en/latest/getting-started/first-steps-with-celery.html#configuration。 – ksindi

我知道它已经很老了，但是你能否提供一个完整的例子来说明如何在一个TestCase类中从OP的问题中启动任务'add'？ –

作为芹菜3.0，设置在CELERY_ALWAYS_EAGER Django的单程的是：

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())

来源

2015-10-27 13:30:32

不要以这种方式修改django设置。它会对所有其他测试产生副作用。 CELERY_ALWAYS_EAGER在测试之后不会恢复，因此它将在所有下列测试中设置为True。请参阅https://docs.djangoproject.com/en/1.9/topics/testing/tools/#overriding-settings 使用@override_settings（CELERY_ALWAYS_EAGER = True）而不是 – ornoone

@ornoone您是正确的。我更新了我的答案。感谢您的反馈意见。 –

要保存额外的谷歌搜索，你需要的导入是：'from django.test import override_settings' – Marshall

单元测试

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_ta SK尊重渴望

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

来源

2016-03-04 22:10:47 ksindi

对于那些对芹菜4是：

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)

由于设置名称已更改，且需要的，如果你选择升级更新，请参阅

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

来源

2016-12-20 18:49:26 okrutny

根据[官方文档]（http://docs.celeryproject.org/en/latest/userguide/testing.html），使用“task_always_eager”（早期的“CELERY_ALWAYS_EAGER”）不适合单元测试。相反，他们提出了其他一些很好的方法来测试你的Celery应用程序。 – krassowski

你如何测试芹菜任务？

回答

py.test夹具

附录：使send_ta SK尊重渴望

相关问题