Unit Testing Celery Tasks
Testing distributed tasks can be tricky because they usually run asynchronously in a separate worker process. However, for unit tests, you typically want to run tasks synchronously (locally) to verify their logic without needing a running message broker or worker.
1. Using task_always_eager
The most common way to test Celery tasks is to set the task_always_eager configuration setting to True. This tells Celery to execute tasks locally and synchronously (blocking) instead of sending them to the queue.
Configuration
You can set this in your test setup or override the settings temporarily.
from celery import Celery
import unittest
app = Celery('tests')
class TestCeleryTasks(unittest.TestCase):
def setUp(self):
# Enable eager mode
app.conf.update(task_always_eager=True)
def test_add_task(self):
from tasks import add
# This will run locally and return the result immediately
result = add.delay(2, 2)
self.assertEqual(result.get(), 4)
self.assertTrue(result.successful())task_eager_propagates
By default, if a task raises an exception in eager mode, it will be stored in the result backend but not raised in the test. To make tests fail immediately on task errors, set task_eager_propagates=True.
app.conf.update(task_always_eager=True, task_eager_propagates=True)2. Using .apply() (Direct Execution)
If you don't want to change the global configuration, you can execute a task directly using the .apply() method instead of .delay() or .apply_async().
from tasks import add
def test_add_direct():
# apply() executes the task inline
result = add.apply(args=(2, 2))
assert result.get() == 43. Mocking Tasks
Sometimes you don't want to execute the task at all (e.g., if it sends an email or charges a credit card). You just want to ensure that the task would have been called.
from unittest.mock import patch
from my_app import register_user
@patch('my_app.send_welcome_email.delay')
def test_register_user(mock_send_email):
register_user('alice@example.com')
# Verify the task was queued
mock_send_email.assert_called_with('alice@example.com')