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

```python
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`.

```python
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()`.

```python
from tasks import add

def test_add_direct():
    # apply() executes the task inline
    result = add.apply(args=(2, 2))
    assert result.get() == 4
```

## 3. 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*.

```python
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')
```

[[programming/python/celery]]