Сообщить об ошибкеКороткий ответ
Для unit-тестирования логики задачи её можно вызвать напрямую как обычную функцию (task.run(...) или task(...)) либо включить режим task_always_eager=True, при котором Celery выполняет задачу синхронно в том же процессе без реального брокера.
Как это работает подробнее
При task_always_eager=True вызов task.delay()/apply_async() выполняет тело задачи немедленно и синхронно в текущем процессе, возвращая EagerResult, что позволяет проверить результат без Redis/RabbitMQ. Альтернативно можно тестировать саму функцию задачи напрямую, игнорируя декоратор Celery, если вся логика вынесена в обычную функцию, а задача — лишь тонкая обёртка.
Пример кода
testing_example_316.py
from myapp.tasks import send_report
def test_send_report_task_runs_synchronously(celery_app, celery_worker):
celery_app.conf.task_always_eager = True
result = send_report.delay(user_id=1)
assert result.get(timeout=5) == "sent"Что использовать на практике
- Практика. Eager-режим хорош для unit-тестирования бизнес-логики задачи, но не проверяет поведение, специфичное для реальной очереди: retry-политики, маршрутизацию по очередям, сериализацию аргументов, поведение при падении воркера. Для этого нужны отдельные интеграционные тесты с реальным (тестовым) брокером.
Подводные камни
- Риск. Полагаться только на eager-тесты и считать, что retry/countdown/routing уже проверены — eager-режим их фактически обходит стороной и не выполняет реальную маршрутизацию. Не мокать побочные эффекты внутри задачи (отправку email, внешние вызовы), из-за чего unit-тест задачи превращается в полноценный, медленный и нестабильный интеграционный тест без явного на то намерения.