Celery 文档提到在 Django 中测试 Celery,但没有说明如果您不使用 Django,如何测试 Celery 任务。你怎么做到这一点?
10 回答
可以使用任何 unittest 库同步测试任务。在处理 celery 任务时,我通常会进行 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)
希望有帮助!
更新我七岁的答案:
您可以通过 pytest 夹具在单独的线程中运行工作人员:
https://docs.celeryproject.org/en/stable/userguide/testing.html#celery-worker-embed-live-worker
根据文档,您不应使用“always_eager”(请参阅上述链接的页面顶部)。
老答案:
我用这个:
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 让您可以同步运行任务,并且不需要 celery 服务器。
取决于你到底想测试什么。
- 直接测试任务代码。不要调用“task.delay(...)”,只需从单元测试中调用“task(...)”。
- 使用CELERY_ALWAYS_EAGER。这将导致您的任务在您说“task.delay(...)”时立即被调用,因此您可以测试整个路径(但不能测试任何异步行为)。
单元测试
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
对于 Celery 4 上的用户,它是:
@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
因为设置名称已更改,如果您选择升级需要更新,请参阅
从Celery 3.0CELERY_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())
从 Celery v4.0开始,提供了py.test 固定装置来启动一个 celery worker 来进行测试,并在完成后关闭:
def test_myfunc_is_executed(celery_session_worker):
# celery_session_worker: <Worker: gen93553@mymachine.local (running)>
assert myfunc.delay().wait(3)
在http://docs.celeryproject.org/en/latest/userguide/testing.html#py-test上描述的其他固定装置中,您可以通过重新定义固定装置来更改 celery 默认选项celery_config
:
@pytest.fixture(scope='session')
def celery_config():
return {
'accept_content': ['json', 'pickle'],
'result_serializer': 'pickle',
}
默认情况下,测试工作者使用内存中的代理和结果后端。如果不测试特定功能,则无需使用本地 Redis 或 RabbitMQ。
使用 pytest参考。
def test_add(celery_worker):
mytask.delay()
如果您使用烧瓶,请设置应用程序配置
CELERY_BROKER_URL = 'memory://'
CELERY_RESULT_BACKEND = 'cache+memory://'
并且在conftest.py
@pytest.fixture
def app():
yield app # Your actual Flask application
@pytest.fixture
def celery_app(app):
from celery.contrib.testing import tasks # need it
yield celery_app # Your actual Flask-Celery application
就我而言(我假设还有很多其他人),我想要的只是使用 pytest 测试任务的内部逻辑。
TL;博士; 最终嘲笑一切(选项2)
示例用例:
proj/tasks.py
@shared_task(bind=True)
def add_task(self, a, b):
return a+b;
tests/test_tasks.py
from proj import add_task
def test_add():
assert add_task(1, 2) == 3, '1 + 2 should equal 3'
但是,由于shared_task
装饰器做了很多 celery 内部逻辑,它并不是真正的单元测试。
所以,对我来说,有两个选择:
选项 1:独立的内部逻辑
proj/tasks_logic.py
def internal_add(a, b):
return a + b;
proj/tasks.py
from .tasks_logic import internal_add
@shared_task(bind=True)
def add_task(self, a, b):
return internal_add(a, b);
这看起来很奇怪,除了使其可读性降低之外,它还需要手动提取和传递作为请求的一部分的属性,例如task_id
以防万一,这会使逻辑不那么纯粹。
选项 2:嘲笑
嘲笑芹菜内部
tests/__init__.py
# 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()
然后允许我模拟请求对象(同样,如果您需要请求中的内容,例如 id 或重试计数器。
tests/test_tasks.py
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'
该解决方案更加手动,但是,它为我提供了实际单元测试所需的控制权,无需重复自己,也不会丢失 celery 范围。
我在单元测试中看到了很多CELERY_ALWAYS_EAGER = true
方法作为单元测试的解决方案,但是由于版本 5.0.5 可用,因此有很多变化使得大多数旧答案被弃用,对我来说是一个耗时的废话,所以对于每个人在这里搜索解决方案,转到文档并阅读新版本的有据可查的单元测试示例:
https://docs.celeryproject.org/en/stable/userguide/testing.html
对于带有单元测试的 Eager Mode,这里引用了实际文档的引用:
渴望模式
task_always_eager 设置启用的渴望模式根据定义不适合单元测试。
在使用 Eager 模式进行测试时,您只是在测试工作人员中发生的情况的仿真,并且仿真与实际发生的情况之间存在许多差异。