Pytest Celery Plugin

1.3.0 · active · verified Sat Apr 11

pytest-celery is a pytest plugin that provides fixtures and utilities for testing Celery applications. It streamlines setting up and tearing down Celery workers and brokers (like Redis, RabbitMQ, SQS) within your pytest test suite. Currently at version 1.3.0, it maintains an active development pace with frequent updates to support new Python and Celery versions, and address compatibility issues.

Warnings

breaking Python 3.8 support was dropped in `pytest-celery` v1.3.0. Users on Python 3.8 must upgrade their Python environment or pin `pytest-celery` to a version prior to 1.3.0.
Fix: Upgrade Python to 3.9 or higher, or pin `pytest-celery<1.3.0` in your project dependencies.
gotcha The `pycurl` dependency for SQS transport was removed in v1.2.0 (in favor of `urllib3`) and then re-added in v1.3.0 to match `celery`/`kombu` v5.6 SQS transport requirements. This flip-flop can cause unexpected dependency issues for users relying on SQS brokers, depending on their `pytest-celery` and `celery`/`kombu` versions.
Fix: If experiencing SQS transport issues, check your installed `celery` and `kombu` versions against `pytest-celery`'s dependency requirements. You may need to manually install `pycurl` (e.g., `pip install 'pytest-celery[sqs]'`) or ensure `urllib3` is correctly used, depending on the `pytest-celery` version.
gotcha By default, `celery_app` and `celery_worker` fixtures have a `function` scope, meaning they are created and torn down for each test. For performance, especially with many tests, consider using the `session` scoped fixtures (`celery_session_app`, `celery_session_worker`) or customizing the scope of the default fixtures.
Fix: Use `celery_session_app` and `celery_session_worker` fixtures for session-scoped Celery instances, or define your own fixtures with `scope='session'` or `scope='module'`.
gotcha From v1.3.0, `setuptools` was removed from dependencies. While not directly breaking for users, projects with custom build systems or highly specific environments that previously relied on `pytest-celery` implicitly bringing `setuptools` might need adjustments.
Fix: Ensure `setuptools` is explicitly listed as a dependency if your project's build process requires it, rather than relying on transitive installation.

Install

pip install "pytest-celery[all]" Install with all extras
pip install pytest-celery Basic install (without optional backends)

Imports

pytest_celery fixtures
```
import pytest; # Fixtures like celery_app, celery_worker are auto-discovered
```
pytest-celery primarily provides fixtures (e.g., `celery_app`, `celery_worker`, `celery_session_app`, `celery_session_worker`) that are automatically discovered and made available to your test functions. Direct imports from `pytest_celery` modules are less common for basic usage, but may be used for advanced customization like `CeleryTestCluster`.

Quickstart

This quickstart demonstrates a basic test for a Celery task using pytest-celery's fixtures. It configures Celery to use Redis, starts a worker, and executes a task. Ensure your test environment has a running Redis instance or adjust the `celery_config` fixture accordingly. Remember to replace `my_app.tasks` with the actual path to your Celery tasks.

import pytest
from celery import Celery

@pytest.fixture(scope="session")
def celery_config():
    return {
        "broker_url": "redis://localhost:6379/0",
        "result_backend": "redis://localhost:6379/0",
        "task_always_eager": False,
        "task_eager_propagates": False
    }

@pytest.fixture(scope="session")
def celery_enable_logging():
    return True

@pytest.fixture(scope="session")
def celery_worker_parameters():
    return {
        'queue': ['celery'],
        'pool': 'solo'
    }

# Assuming you have a tasks.py module with a simple task
# e.g., tasks.py:
# from celery import shared_task
# @shared_task
# def add(x, y):
#     return x + y

def test_add_task(celery_app: Celery, celery_worker):
    # Ensure tasks.py is discoverable, e.g., by adding a custom Celery app fixture
    # or importing tasks directly if they are in the test file.
    from my_app.tasks import add # Replace with your actual task module
    result = add.delay(2, 3)
    assert result.get(timeout=10) == 5
    assert result.successful()

view raw JSON →