celery-tests
Overview
Per docs.celeryq.dev/en/stable/userguide/testing.html:
"Testing with Celery is divided into two parts: Unit & Integration:
Using celery.contrib.pytest. Smoke / Production: Using
pytest-celery >= 1.0.0"
The test model uses pytest fixtures from the celery.contrib.pytest
plugin (or the standalone pytest-celery package for newer versions).
When to use
- The repo has Celery task definitions (
@app.task or
@shared_task).
- The user writes unit tests for the task body (no need for a real
worker).
- The user writes integration tests that require a real Celery
worker thread.
- A test verifies retry / chord / chain semantics.
How to use
- Confirm the repo has Celery tasks (
@app.task / @shared_task) and pick
the level: unit (task body) vs integration (real worker).
- For unit tests, call the task function directly and assert its side
effects; do NOT enable
task_always_eager (Step 1, Step 2).
- For retry logic,
unittest.mock.patch the <task>.retry and the failing
dependency, then assert Retry is raised (Step 3).
- For an in-process synchronous run with no broker, invoke
task.apply(args=[...]) and assert on result.successful() /
result.result (Step 5).
- For integration tests that need a real worker, add the
celery_worker
(per-test) or celery_session_worker (per-session) fixture, call
task.delay(...), and read result.get(timeout=N) (Step 4).
- Test
chord / chain / group end-to-end with celery_worker, or mock
at boundaries for unit level (Step 6).
- Wire
pytest into CI, adding redis / rabbitmq service containers only
when a test needs a real broker (Step 7).
Step 1 - Don't rely on task_always_eager for unit tests
Per cel-test:
"The eager mode enabled by the [task_always_eager] setting is by
definition not suitable for unit tests."
Reason (per cel-test): "eagerly executed tasks don't
write results to backend by default."
For unit tests, prefer to call the task function directly (test the
logic) and mock the dispatch where queue interaction matters.
Step 2 - Direct task-function unit tests
Test the task function as if it were a regular function:
from proj.tasks import send_order
from decimal import Decimal
def test_send_order_calls_product_order():
product = Product.objects.create(name='Foo')
send_order(product.pk, 3, Decimal('30.30'))
# Assert side effects (DB row, external call, etc.)
This bypasses Celery's dispatch entirely - fastest, most direct.
Step 3 - Mock retry behavior
Per cel-test (verbatim retry-test pattern):
from pytest import raises
from celery.exceptions import Retry
from unittest.mock import patch
from proj.models import Product
from proj.tasks import send_order
class test_send_order:
@patch('proj.tasks.Product.order')
def test_success(self, product_order):
product = Product.objects.create(name='Foo')
send_order(product.pk, 3, Decimal(30.3))
product_order.assert_called_with(3, Decimal(30.3))
@patch('proj.tasks.send_order.retry')
def test_failure(self, send_order_retry, product_order):
send_order_retry.side_effect = Retry()
product_order.side_effect = OperationalError()
with raises(Retry):
send_order(product.pk, 3, Decimal(30.6))
Patch <task>.retry to assert the task triggers a retry; patch the
side-effect dependency to control failure mode.
Step 4 - pytest-celery fixtures (integration tests)
Per cel-test, the canonical pytest-celery fixtures:
| Fixture | Use |
|---|
celery_app | "This fixture returns a Celery app you can use for testing." |
celery_worker | "This fixture starts a Celery worker instance that you can use for integration tests. The worker will be started in a separate thread." |
celery_session_worker | "This fixture starts a worker that lives throughout the testing session (it won't be started/stopped for every test)." |
Choose celery_worker for tests that need clean worker state per
test; celery_session_worker for fast suites where the worker can
be reused.
def test_task_runs_via_real_worker(celery_app, celery_worker):
@celery_app.task
def add(x, y):
return x + y
result = add.delay(2, 3)
assert result.get(timeout=10) == 5
Step 5 - apply() for synchronous test invocation
When you want to invoke the task synchronously without spawning a
worker:
from proj.tasks import send_order
result = send_order.apply(args=[product_id, qty, amount])
assert result.successful()
assert result.result == expected_value
apply() runs the task in-process; delay() enqueues for a worker.
For tests, prefer apply() (no Redis / RabbitMQ dependency) unless
testing the dispatch path itself.
Step 6 - Test chord / chain / group
Celery's primitives compose:
from celery import chain, group, chord
# Chain: A -> B -> C
chain(task_a.s(1), task_b.s(), task_c.s())()
# Group: parallel execution
group(task_a.s(i) for i in range(3))()
# Chord: parallel group + callback
chord([task_a.s(i) for i in range(3)])(callback.s())
For unit tests, mock the primitives at boundaries; for integration
tests, use celery_worker fixture (Step 4) - primitives execute
end-to-end.
Step 7 - CI integration
- run: pip install -r requirements-dev.txt # includes pytest, celery, pytest-celery
- run: pytest -v
For tests requiring real broker (RabbitMQ / Redis):
services:
redis: { image: redis:7, ports: [6379:6379] }
# or rabbitmq:
rabbitmq: { image: rabbitmq:3-management, ports: [5672:5672, 15672:15672] }
Worked example
An order-processing task must run its side effect on success and retry on a
transient DB error. Two fast unit tests cover both paths, with no broker.
- Success path - call the task function directly, patch the dependency, and
assert the side effect:
@patch('proj.tasks.Product.order')
def test_send_order_success(product_order):
product = Product.objects.create(name='Foo')
send_order(product.pk, 3, Decimal('30.30'))
product_order.assert_called_with(3, Decimal('30.30'))
- Retry path - patch
send_order.retry to raise Retry, force the
dependency to fail, then assert the task retries:
@patch('proj.tasks.send_order.retry', side_effect=Retry())
@patch('proj.tasks.Product.order', side_effect=OperationalError())
def test_send_order_retries_on_db_error(order, retry):
with raises(Retry):
send_order(1, 3, Decimal('30.30'))
Running pytest -v reports both tests passing in milliseconds with no Redis
/ RabbitMQ dependency - the success side effect and the retry trigger are both
proven without spawning a worker.
Anti-patterns
| Anti-pattern | Why it fails | Fix |
|---|
task_always_eager = True for unit tests | Per cel-test, "not suitable for unit tests"; results don't write to backend | Direct function call (Step 2) or apply() (Step 5) |
Use delay() in tests without a real worker | Tasks enqueue but never execute; tests hang or never assert | Use apply() for sync; celery_worker fixture for real-worker integration |
| Skip patching the broker side-effect | Tests hit real broker / DB | unittest.mock.patch the boundary (Step 3) |
Reuse celery_session_worker for tests with conflicting task definitions | Worker has stale task registry; later tests fail | Use celery_worker for changing-registry tests; celery_session_worker for stable |
Limitations
- Worker-thread fixtures slow tests substantially vs direct function
calls - use them only when integration matters.
result.get(timeout=N) with too-short N causes intermittent
CI failures on slow runners; pick generous timeouts.- pytest-celery v1.0+ has a different API than legacy
celery.contrib.pytest;
pin one and stick with it per project.
References
- cel-test - official testing guide, fixtures, retry
patterns, eager-mode warning
- docs.celeryq.dev - full Celery documentation
- pypi.org/project/pytest-celery - pytest-celery package
sidekiq-tests,
bullmq-tests,
sqs-tests,
rabbitmq-tests - sister toolsidempotency-test-author,
cron-job-test-author - build-an-X authors