Celery 5 tasks, chains and groups tested in eager mode: working versions (tested 2026-10-15)
Define Celery tasks, chain and fan them out with groups, then test them offline with task_always_eager and the in-memory broker.
passed4 passed · 0 failed2026-10-15 11:34 UTC- Status
- passed
- Last passed
- 2026-10-15 11:34 UTC
- Tests
- 4 / 4
- Language
- Python
- Runtime
- Python 3.13.16
pinned versions2
celery@5.6.3pytest@9.1.1
run it
uv venv -q .venv && uv pip install -q --python .venv/bin/python -r requirements.txt .venv/bin/python -m pytest -q
test environment
Python 3.13.16 linux 6.17.0-1022-azure x64 runner: github-actions 38728226240
The code
Exactly the files that passed. Your agent can pull them with get_recipe("celery-tasks-eager-testing") (install), or see them on GitHub.
requirements.txt
celery==5.6.3 pytest==9.1.1
tasks.py
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())
tests/test_tasks.py
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.ini
[pytest] pythonpath = .
Connect to ProvenPin
Add the free MCP server so your agent can pull this recipe and others like it. Same setup as the docs.
terminal
claude mcp add --transport http provenpin https://mcp.provenpin.com/mcp
~/.cursor/mcp.json
{ "mcpServers": { "provenpin": { "url": "https://mcp.provenpin.com/mcp" } } }.vscode/mcp.json
{ "servers": { "provenpin": { "type": "http", "url": "https://mcp.provenpin.com/mcp" } } }~/.codeium/windsurf/mcp_config.json
{ "mcpServers": { "provenpin": { "serverUrl": "https://mcp.provenpin.com/mcp" } } }MCP settings
{ "mcpServers": { "provenpin": { "type": "streamableHttp", "url": "https://mcp.provenpin.com/mcp" } } }Any other MCP client: Streamable HTTP at https://mcp.provenpin.com/mcp. No API key, no signup.