Celery 5 tasks, chains and groups tested in eager mode: working versions (tested 2026-10-15)

celery==5.6.3
pytest==9.1.1

from celery import Celery, chain, group

app = Celery("tasks", broker="memory://", backend="cache+memory://")

@app.task
def add(x, y):
    return x + y

@app.task
def double(x):
    return x * 2

def add_then_double(x, y):
    return chain(add.s(x, y), double.s())

def sum_of_doubles(nums):
    # fan out with a group, then reduce the results
    return sum(group(double.s(n) for n in nums).apply_async().get())

import pytest
from tasks import app, add, add_then_double, sum_of_doubles

@pytest.fixture(autouse=True)
def eager():
    app.conf.update(task_always_eager=True, task_eager_propagates=True)
    yield
    app.conf.update(task_always_eager=False)

def test_delay():
    assert add.delay(2, 3).get() == 5

def test_chain():
    assert add_then_double(2, 3).apply_async().get() == 10

def test_group():
    assert sum_of_doubles([1, 2, 3]) == 12

def test_errors_propagate():
    with pytest.raises(TypeError):
        add.delay(1, "a").get()

[pytest]
pythonpath = .