TopolearnTasks are business logic that runs asynchronously, and testing them shouldn't require a broker in every test.
The simplest and best tests call the task's function directly: add(2, 3) runs the body synchronously like a normal function. Keep task bodies thin, delegating to plain functions you can test without Celery at all.
To test the Celery call path without a broker:
task.apply(args=(...))executes the task locally and returns anEagerResult, including retries and state handling.task_always_eager=Truemakes everydelay()run inline. Pair it withtask_eager_propagates=Trueso exceptions surface in tests. It's convenient, but eager mode doesn't serialize arguments or go through routing, so it can hide bugs like passing non-JSON objects.- For code that enqueues tasks, mock
task.delayand assert it was called with the expected arguments; that tests the caller without running the task.
Celery ships a pytest plugin (celery.contrib.pytest, enable with -p celery.contrib.pytest or in conftest) with fixtures like celery_app and celery_worker, which start a real in-process worker using an in-memory broker. Use those for a few integration tests of workflows.
Test retries by mocking the failing dependency and asserting on the number of calls, and test idempotency by running a task twice with the same arguments.
Example
# test_tasks.py
from unittest.mock import patch
import pytest
from celery import Celery
app = Celery("t", broker="memory://")
@app.task
def add(x, y):
return x + y
def signup(user_id):
add.delay(user_id, 1) # code under test enqueues a task
def test_logic_directly():
assert add(2, 3) == 5 # plain function call
def test_apply_runs_locally():
r = add.apply(args=(2, 3))
assert r.get() == 5 and r.successful()
def test_signup_enqueues():
with patch.object(add, "delay") as delay:
signup(7)
delay.assert_called_once_with(7, 1)
# $ pytest -q -> 3 passedRemember
- Call the task function directly to unit-test its logic
- task.apply() runs through Celery locally without a broker
- Mock .delay to test code that enqueues tasks
- Eager mode skips serialization and routing: it can hide bugs
Watch out for
- Relying only on task_always_eager and missing a non-serializable argument that fails in production.