Я пытаюсь придумать методологию тестирования для нашего проекта django-celery . Я прочитал примечания в документации , но это не дало мне хорошего представления о том, что на самом деле делать. Я не беспокоюсь о тестировании задач в реальных демонах, а только о функциональности моего кода. В основном мне интересно:
- Как мы можем обойтись
task.delay()
во время теста (я пробовал настройку,CELERY_ALWAYS_EAGER = True
но без разницы)? - Как мы можем использовать рекомендуемые тестовые настройки (если это лучший способ) без фактического изменения нашего settings.py?
- Можем ли мы по-прежнему использовать
manage.py test
или должны использовать пользовательский раннер?
В целом, любые советы или подсказки по тестированию с сельдереем были бы очень полезны.
python
django
unit-testing
celery
Джейсон Уэбб
источник
источник
CELERY_ALWAYS_EAGER
без разницы?.delay
попытки установить соединение.BROKER_BACKEND=memory
В этом случае может помочь настройка .BROKER_BACKEND=memory
починил это. Если вы поставите это как ответ, я отмечу, что это правильно.Ответы:
Попробуйте установить:
BROKER_BACKEND = 'memory'
(Спасибо за комментарий asksol .)
источник
Мне нравится использовать декоратор override_settings в тестах, для завершения которых нужны результаты сельдерея.
from django.test import TestCase from django.test.utils import override_settings from myapp.tasks import mytask class AddTestCase(TestCase): @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True, CELERY_ALWAYS_EAGER=True, BROKER_BACKEND='memory') def test_mytask(self): result = mytask.delay() self.assertTrue(result.successful())
Если вы хотите применить это ко всем тестам, вы можете использовать средство запуска тестов сельдерея, как описано на http://docs.celeryproject.org/en/2.5/django/unit-testing.html, которое в основном устанавливает те же настройки, кроме (
BROKER_BACKEND = 'memory'
).В настройках:
TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'
Посмотрите на исходный код CeleryTestSuiteRunner, и становится ясно, что происходит.
источник
djcelery
.Вот отрывок из моего базового класса тестирования, в котором этот
apply_async
метод заменяется заглушкой и записываются вызовы к нему (в том числеTask.delay
.) Это немного грубо, но в течение последних нескольких месяцев мне удалось удовлетворить мои потребности.from django.test import TestCase from celery.task.base import Task # For recent versions, Task has been moved to celery.task.app: # from celery.app.task import Task # See http://docs.celeryproject.org/en/latest/reference/celery.app.task.html class CeleryTestCaseBase(TestCase): def setUp(self): super(CeleryTestCaseBase, self).setUp() self.applied_tasks = [] self.task_apply_async_orig = Task.apply_async @classmethod def new_apply_async(task_class, args=None, kwargs=None, **options): self.handle_apply_async(task_class, args, kwargs, **options) # monkey patch the regular apply_sync with our method Task.apply_async = new_apply_async def tearDown(self): super(CeleryTestCaseBase, self).tearDown() # Reset the monkey patch to the original method Task.apply_async = self.task_apply_async_orig def handle_apply_async(self, task_class, args=None, kwargs=None, **options): self.applied_tasks.append((task_class, tuple(args), kwargs)) def assert_task_sent(self, task_class, *args, **kwargs): was_sent = any(task_class == task[0] and args == task[1] and kwargs == task[2] for task in self.applied_tasks) self.assertTrue(was_sent, 'Task not called w/class %s and args %s' % (task_class, args)) def assert_task_not_sent(self, task_class): was_sent = any(task_class == task[0] for task in self.applied_tasks) self.assertFalse(was_sent, 'Task was not expected to be called, but was. Applied tasks: %s' % self.applied_tasks)
Вот пример того, как вы могли бы использовать его в своих тестовых случаях:
mymodule.py
from my_tasks import SomeTask def run_some_task(should_run): if should_run: SomeTask.delay(1, some_kwarg=2)
test_mymodule.py
class RunSomeTaskTest(CeleryTestCaseBase): def test_should_run(self): run_some_task(should_run=True) self.assert_task_sent(SomeTask, 1, some_kwarg=2) def test_should_not_run(self): run_some_task(should_run=False) self.assert_task_not_sent(SomeTask)
источник
поскольку я все еще вижу это в результатах поиска, настройки отменяются
TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'
работал у меня в соответствии с документами Celery
источник
Для всех, кто попадает сюда в 2019 году: ознакомьтесь с этой статьей, в которой рассматриваются различные стратегии, включая синхронный вызов задач.
источник
Вот что я сделал
Внутри myapp.tasks.py у меня есть:
from celery import shared_task @shared_task() def add(a, b): return a + b
Внутри myapp.test_tasks.py у меня есть:
from django.test import TestCase, override_settings from myapp.tasks import add class TasksTestCase(TestCase): def setUp(self): ... @override_settings(CELERY_TASK_ALWAYS_EAGER=True,CELERY_TASK_EAGER_PROPOGATES=True) def test_create_sections(self): result= add.delay(1,2) assert result.successful() == True assert result.get() == 3
источник