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

1. Confirm the repo has Celery tasks (@app.task / @shared_task) and pick the level: unit (task body) vs integration (real worker).
2. For unit tests, call the task function directly and assert its side effects; do NOT enable task_always_eager (Step 1, Step 2).
3. For retry logic, unittest.mock.patch the <task>.retry and the failing dependency, then assert Retry is raised (Step 3).
4. For an in-process synchronous run with no broker, invoke task.apply(args=[...]) and assert on result.successful() / result.result (Step 5).
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).
6. Test chord / chain / group end-to-end with celery_worker, or mock at boundaries for unit level (Step 6).
7. 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:

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.

1. 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'))

2. 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

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 tools
• idempotency-test-author, cron-job-test-author - build-an-X authors