flakytest.dev
Sign In Get started

Speed up your test suite

Run Celery tasks eagerly in tests

Tests that call task.delay() publish to a broker, retry against a missing one or wait on a worker. Run tasks eagerly with CELERY_TASK_ALWAYS_EAGER, or mock delay.

Application code queues work with send_receipt.delay(order.pk). In production a worker picks it up. In a test, that same call has to go somewhere: to a real broker, which makes the test slow and dependent on Redis or RabbitMQ running; to a broker that isn't there, which Celery retries before failing; or to a worker started for the test, which the test then waits on.

Tests rarely need any of that. Most want the task to run right away, in the same process, or only want to check that it was queued.

Why Celery is slow in tests

  • Publishing to a broker. delay() and apply_async() serialize the arguments and send a message over the network. The first call in a process also connects.
  • No broker running. Celery retries publishing by default. With the default retry policy (3 retries, waiting 0, 0.2 and 0.4 seconds) that's about 0.6 seconds of waiting per call before it raises OperationalError, and a test that catches or ignores the error just gets slower.
  • A real worker. celery.contrib.pytest's celery_worker fixture is function-scoped, so it starts a worker thread for every test that uses it, and the test then blocks on result.get() while the worker picks the task up.

The fix: eager mode in test settings

With task_always_eager, delay() and apply_async() run the task immediately, in the calling process, and return an EagerResult. No broker is involved. In a Django project that loads Celery's configuration from Django settings (app.config_from_object("django.conf:settings", namespace="CELERY")), set it in test settings:

# settings.py
if TESTING:
    CELERY_TASK_ALWAYS_EAGER = True
    CELERY_TASK_EAGER_PROPAGATES = True

TESTING is the flag from Django test settings. Without Django, configure the app in conftest.py:

# conftest.py
from myapp.celery import app

app.conf.update(task_always_eager=True, task_eager_propagates=True)

Always set task_eager_propagates too. Without it, an exception inside an eager task is stored in the EagerResult (its status becomes FAILURE) and the test carries on as if the task succeeded.

Cheaper still: call the task or mock delay

Eager mode runs everything a task does, including the slow parts. Often a test only needs one side of it.

To test what the task does, call it like a function. A Celery task called directly runs synchronously, without a broker and without eager mode:

from django.test import TestCase
from model_bakery import baker

from shop.models import Order
from shop.tasks import send_receipt


class SendReceiptTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.order = baker.make(Order)

    def test_marks_order_as_receipted(self):
        send_receipt(self.order.pk)
        self.order.refresh_from_db()
        self.assertTrue(self.order.receipt_sent)

To test that code queues a task, patch delay and assert on the call:

from unittest import mock

from django.test import TestCase
from model_bakery import baker

from shop.models import Order
from shop.services import place_order
from shop.tasks import send_receipt


class PlaceOrderTests(TestCase):
    @classmethod
    def setUpTestData(cls):
        cls.order = baker.make(Order)

    def test_queues_receipt(self):
        with mock.patch.object(send_receipt, "delay") as delay:
            place_order(self.order)
        delay.assert_called_once_with(self.order.pk)

Tasks queued on commit

Code that queues tasks only after the transaction commits, with transaction.on_commit(lambda: send_receipt.delay(order.pk)) or Celery's send_receipt.delay_on_commit(order.pk), never queues them inside Django's TestCase, because the test's transaction never commits. Capture the callbacks and run them:

    def test_sends_receipt_after_commit(self):
        with self.captureOnCommitCallbacks(execute=True):
            place_order(self.order)
        self.order.refresh_from_db()
        self.assertTrue(self.order.receipt_sent)

With eager mode on, the callbacks run the task right away. This is much cheaper than switching the test to TransactionTestCase to get a real commit; see TestCase instead of TransactionTestCase.

Keep the tests correct

Celery's own documentation warns that eager mode is only an emulation of a worker, so know what it skips:

  • Eager tasks skip serialization: an argument that can't be serialized, like a model instance, works in a test and fails in production. Pass ids.
  • The task runs inside the caller's transaction and sees its uncommitted rows; a real worker runs later, in its own transaction.
  • Routing, queues, rate limits and time limits don't apply, and results aren't stored in the result backend unless you set task_store_eager_result.

Keep a few tests that run a real worker (for example with the celery_session_worker fixture from celery.contrib.pytest) for the behaviour you depend on.

How flakytest finds it

In a Django project, flakytest measures the time spent in Celery's code and reads CELERY_TASK_ALWAYS_EAGER from your settings. When Celery takes at least half a second and eager mode isn't on, it suggests Run Celery tasks eagerly in tests, with the tests and lines that called into Celery. Time a test spends blocked waiting on a worker's result can also show up as waiting: see tests blocked on threads, queues and the event loop.

Further reading

Related guides

Updated October 18, 2026 · Suggest a change