取决于你到底想测试什么。

• 直接测试任务代码。不要调用“ task.late (...)”，只要从单元测试中调用“ task (...)”即可。
• 使用 芹菜，永远，渴望。这将导致您的任务在您说“ task.late (...)”时立即被调用，因此您可以测试整个路径(但不是任何异步行为)。

可以使用任何单元测试库同步测试任务。在处理芹菜任务时，我通常会做两个不同的测试。第一个(正如我在下面建议的那样)是完全同步的，应该是确保算法完成其应该完成的任务的那个。第二个会话使用整个系统(包括代理) ，并确保我没有出现序列化问题或任何其他分发、通信问题。

所以:

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)

希望能帮上忙！

以下是我七年前的答案:

您可以通过 pytest fixture在单独的线程中运行 worker:

Https://docs.celeryq.dev/en/v5.2.6/userguide/testing.html#celery-worker-embed-live-worker

根据文件，你不应该使用 "always_eager"(见上面链接页面的顶部)。

老答案:

我用这个:

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

医生: https://docs.celeryq.dev/en/3.1/configuration.html#celery-always-eager

CELERY_ALWAYS_EAGER允许您同步运行任务，并且不需要芹菜服务器。

至于 芹菜3.0，在 姜戈中设置 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())

团结一致

import unittest


from myproject.myapp import celeryapp


class TestMyCeleryWorker(unittest.TestCase):


def setUp(self):
celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)

测试装置

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

由于设置名称已更改，如果选择升级，则需要更新，请参见

Https://docs.celeryproject.org/en/latest/history/whatsnew-4.0.html?highlight=what%20is%20new#lowercase-setting-names

在我的例子中(我假设还有许多其他的例子) ，我所需要的只是使用 pytest 测试任务的内部逻辑。

结果嘲笑了一切(选择二)

@shared_task(bind=True)
def add_task(self, a, b):
return a+b;

from proj import add_task


def test_add():
assert add_task(1, 2) == 3, '1 + 2 should equal 3'

def internal_add(a, b):
return a + b;

from .tasks_logic import internal_add


@shared_task(bind=True)
def add_task(self, a, b):
return internal_add(a, b);

# noinspection PyUnresolvedReferences
from celery import shared_task


from mock import patch




def mock_signature(**kwargs):
return {}




def mocked_shared_task(*decorator_args, **decorator_kwargs):
def mocked_shared_decorator(func):
func.signature = func.si = func.s = mock_signature
return func


return mocked_shared_decorator


patch('celery.shared_task', mocked_shared_task).start()

from proj import add_task


class MockedRequest:
def __init__(self, id=None):
self.id = id or 1




class MockedTask:
def __init__(self, id=None):
self.request = MockedRequest(id=id)




def test_add():
mocked_task = MockedTask(id=3)
assert add_task(mocked_task, 1, 2) == 3, '1 + 2 should equal 3'