{"id":3769,"library":"pytest-celery","title":"Pytest Celery Plugin","description":"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.","status":"active","version":"1.3.0","language":"en","source_language":"en","source_url":"https://github.com/celery/pytest-celery","tags":["pytest","celery","testing","async","fixtures"],"install":[{"cmd":"pip install \"pytest-celery[all]\"","lang":"bash","label":"Install with all extras"},{"cmd":"pip install pytest-celery","lang":"bash","label":"Basic install (without optional backends)"}],"dependencies":[{"reason":"Core testing framework.","package":"pytest","optional":false},{"reason":"Core async task queue framework.","package":"celery","optional":false},{"reason":"Required for Redis broker/backend support (via 'redis' extra).","package":"redis","optional":true},{"reason":"Required for AWS SQS broker support (via 'sqs' extra).","package":"boto3","optional":true},{"reason":"Re-added in v1.3.0 for SQS transport compatibility with specific Kombu versions. May be required for SQS broker support depending on Celery/Kombu version.","package":"pycurl","optional":true}],"imports":[{"note":"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`.","symbol":"pytest_celery fixtures","correct":"import pytest; # Fixtures like celery_app, celery_worker are auto-discovered"}],"quickstart":{"code":"import pytest\nfrom celery import Celery\n\n@pytest.fixture(scope=\"session\")\ndef celery_config():\n    return {\n        \"broker_url\": \"redis://localhost:6379/0\",\n        \"result_backend\": \"redis://localhost:6379/0\",\n        \"task_always_eager\": False,\n        \"task_eager_propagates\": False\n    }\n\n@pytest.fixture(scope=\"session\")\ndef celery_enable_logging():\n    return True\n\n@pytest.fixture(scope=\"session\")\ndef celery_worker_parameters():\n    return {\n        'queue': ['celery'],\n        'pool': 'solo'\n    }\n\n# Assuming you have a tasks.py module with a simple task\n# e.g., tasks.py:\n# from celery import shared_task\n# @shared_task\n# def add(x, y):\n#     return x + y\n\ndef test_add_task(celery_app: Celery, celery_worker):\n    # Ensure tasks.py is discoverable, e.g., by adding a custom Celery app fixture\n    # or importing tasks directly if they are in the test file.\n    from my_app.tasks import add # Replace with your actual task module\n    result = add.delay(2, 3)\n    assert result.get(timeout=10) == 5\n    assert result.successful()\n","lang":"python","description":"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."},"warnings":[{"fix":"Upgrade Python to 3.9 or higher, or pin `pytest-celery<1.3.0` in your project dependencies.","message":"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.","severity":"breaking","affected_versions":">=1.3.0"},{"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.","message":"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.","severity":"gotcha","affected_versions":"1.2.x, 1.3.x"},{"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'`.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"fix":"Ensure `setuptools` is explicitly listed as a dependency if your project's build process requires it, rather than relying on transitive installation.","message":"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.","severity":"gotcha","affected_versions":">=1.3.0"}],"env_vars":null,"last_verified":"2026-04-11T00:00:00.000Z","next_check":"2026-07-10T00:00:00.000Z"}