This document describes the current stable version of pytest_celery (1.0). For development docs, go here.
django¶
- Release:
1.0
- Date:
May 16, 2024
Description¶
This example project demonstrates how to test the official Celery django example using pytest-celery. For more about Django and Celery, please see the official documentation.
For the purposes of this example, we will focus on the setup and configuration side of things.
Warning
Django support is currently experimental. Please report any issues you encounter.
Breakdown¶
File Structure¶
The following diagram lists the relevant files in the project.
django/
├── demoapp/
│ ├── tasks.py
├── tests/
│ ├── __init__.py
│ ├── conftest.py
│ └── DjangoWorker.Dockerfile
│ └── test_tasks.py
└── requirements.txt
DjangoWorker.Dockerfile¶
FROM python:3.11-bookworm
test_user
is created to run the worker.
# Create a user to run the worker
RUN adduser --disabled-password --gecos "" test_user
# Install system dependencies
RUN apt-get update && apt-get install -y build-essential git
CELERY_LOG_LEVEL
, CELERY_WORKER_NAME
and CELERY_WORKER_QUEUE
are set as build arguments.
These will be used to configure the worker for the tests.
# Set arguments
ARG CELERY_LOG_LEVEL=INFO
ARG CELERY_WORKER_NAME=my_worker
ARG CELERY_WORKER_QUEUE=celery
ENV LOG_LEVEL=$CELERY_LOG_LEVEL
ENV WORKER_NAME=$CELERY_WORKER_NAME
ENV WORKER_QUEUE=$CELERY_WORKER_QUEUE
EXPOSE 5678
/src
is arbitrarily chosen as the working directory to install the Django project from.
# Install packages
WORKDIR /src
COPY --chown=test_user:test_user requirements.txt .
RUN pip install --no-cache-dir --upgrade pip
RUN pip install -r ./requirements.txt
# Switch to the test_user
USER test_user
# Start the celery worker
CMD celery -A proj worker --loglevel=$LOG_LEVEL -n $WORKER_NAME@%h -Q $WORKER_QUEUE
conftest.py¶
The DjangoWorkerContainer
class is used to configure the worker container and acts as the interface
to the container instance.
class DjangoWorkerContainer(CeleryWorkerContainer):
@property
def client(self) -> Any:
return self
@classmethod
def version(cls) -> str:
return celery.__version__
@classmethod
def log_level(cls) -> str:
return "INFO"
@classmethod
def worker_name(cls) -> str:
return "django_tests_worker"
@classmethod
def worker_queue(cls) -> str:
return "celery"
Next, we build our worker image using the build and container fixtures.
worker_image = build(
path=".",
dockerfile="tests/DjangoWorker.Dockerfile",
tag="pytest-celery/examples/django:example",
buildargs=DjangoWorkerContainer.buildargs(),
)
default_worker_container = container(
image="{worker_image.id}",
ports=fxtr("default_worker_ports"),
environment=fxtr("default_worker_env"),
network="{default_pytest_celery_network.name}",
volumes={
"{default_worker_volume.name}": defaults.DEFAULT_WORKER_VOLUME,
os.path.abspath(os.getcwd()): {
"bind": "/src",
"mode": "rw",
},
},
wrapper_class=DjangoWorkerContainer,
timeout=defaults.DEFAULT_WORKER_CONTAINER_TIMEOUT,
)
In this case, we also mount the project directory to /src
in the container, so that we can install the project
inside the container and access the Django project files.
Lastly, we override the default worker container class with our custom class.
Note
This is only required when overriding the default worker.
@pytest.fixture
def default_worker_container_cls() -> type[CeleryWorkerContainer]:
return DjangoWorkerContainer
@pytest.fixture(scope="session")
def default_worker_container_session_cls() -> type[CeleryWorkerContainer]:
return DjangoWorkerContainer
test_tasks.py¶
The test_tasks.py
file contains the tests for the demoapp
tasks.
It can directly import the tasks and the celery_setup
will
run the django app worker so the tasks can be tested.
from demoapp.tasks import add
from demoapp.tasks import count_widgets
def test_add(celery_setup):
assert add.s(1, 2).delay().get() == 3
def test_count_widgets(celery_setup):
assert count_widgets.s().delay().get() == 0
Note
Don’t forget to export DJANGO_SETTINGS_MODULE=proj.settings
and run migration
when running the example locally.
See CI for a usage example.