This document describes the current stable version of pytest_celery (1.0). For development docs, go here.

pytest_celery.api package

Submodules

pytest_celery.api.backend module

Backend components represents Celery’s result backend instances.

This module provides the base API for creating new backend components by defining the base classes for backend nodes and clusters.

class pytest_celery.api.backend.CeleryBackendCluster(*nodes: tuple[CeleryTestNode | CeleryTestContainer])

Bases: CeleryTestCluster

This is a specialized cluster type for handling celery backends. It is used to define which backend instances are available for the test.

Responsibility Scope:

Provude useful methods for managing a cluster of celery backends.

classmethod default_config() → dict

Default cluster configurations if not overridden by the user.

class pytest_celery.api.backend.CeleryTestBackend(container: CeleryTestContainer, app: Celery = None)

Bases: CeleryTestNode

This is specialized node type for handling celery backends nodes. It is used to encapsulate a backend instance.

Responsibility Scope:

Handling backend specific requirements and configuration.

classmethod default_config() → dict

Default node configurations if not overridden by the user.

restart(reload_container: bool = True, force: bool = False) → None

Override restart method to update the app result backend with new container values.

pytest_celery.api.base module

The node/cluster set are designed to provide a common interface for encapsulating the implementation of test components and their integration with the test framework.

This module provides the base API for creating new components by defining the base classes for nodes and clusters.

class pytest_celery.api.base.CeleryTestCluster(*nodes: tuple[CeleryTestNode | CeleryTestContainer])

Bases: object

This is a collection of CeleryTestNodes. It is used to collect the test nodes into a single object for easier management.

Responsibility Scope:

The cluster’s responsibility is to define which nodes will be used for the test.

config(*args: tuple, **kwargs: dict) → dict

Compile the configurations required for Celery from this cluster.

classmethod default_config() → dict

Default cluster configurations if not overridden by the user.

property nodes: tuple[CeleryTestNode]

Get the nodes of the cluster.

ready() → bool

Waits until the cluster is ready or raise an exception if any of the nodes fail to boot up.

teardown() → None

Teardown the cluster.

class pytest_celery.api.base.CeleryTestNode(container: CeleryTestContainer, app: Celery = None)

Bases: object

This is the logical representation of a container instance. It is used to provide a common interface for interacting with the container regardless of the underlying implementation.

Responsibility Scope:

The node’s responsibility is to wrap the container and provide useful methods for interacting with it.

property app: Celery

Celery app for the node if available.

assert_log_does_not_exist(log: str, message: str = '', timeout: int = 1) → None

Assert that a log does not exist in the container.

Parameters:

• log (str) – Log to assert.

• message (str, optional) – Message to display while waiting. Defaults to “”.

• timeout (int, optional) – Timeout in seconds. Defaults to 1.

assert_log_exists(log: str, message: str = '', timeout: int = 60) → None

Assert that a log exists in the container.

Parameters:

• log (str) – Log to assert.

• message (str, optional) – Message to display while waiting. Defaults to “”.

• timeout (int, optional) – Timeout in seconds. Defaults to RESULT_TIMEOUT.

config(*args: tuple, **kwargs: dict) → dict

Compile the configurations required for Celery from this node.

property container: CeleryTestContainer

Underlying container for the node.

classmethod default_config() → dict

Default node configurations if not overridden by the user.

hostname() → str

Get the hostname of this node.

kill(signal: str | int = 'SIGKILL', reload_container: bool = True) → None

Kill the underlying container.

Parameters:

• signal (str | int, optional) – Signal to send to the container. Defaults to “SIGKILL”.

• reload_container (bool, optional) – Reload the container object after killing it. Defaults to True.

logs() → str

Get the logs of the underlying container.

name() → str

Get the name of this node.

ready() → bool

Waits until the node is ready or raise an exception if it fails to boot up.

restart(reload_container: bool = True, force: bool = False) → None

Restart the underlying container.

Parameters:

• reload_container (bool, optional) – Reload the container object after restarting it. Defaults to True.

• force (bool, optional) – Kill the container before restarting it. Defaults to False.

teardown() → None

Teardown the node.

wait_for_log(log: str, message: str = '', timeout: int = 60) → None

Wait for a log to appear in the container.

Parameters:

• log (str) – Log to wait for.

• message (str, optional) – Message to display while waiting. Defaults to “”.

• timeout (int, optional) – Timeout in seconds. Defaults to RESULT_TIMEOUT.

pytest_celery.api.broker module

Broker components represents Celery’s broker instances.

This module provides the base API for creating new broker components by defining the base classes for broker nodes and clusters.

class pytest_celery.api.broker.CeleryBrokerCluster(*nodes: tuple[CeleryTestNode | CeleryTestContainer])

Bases: CeleryTestCluster

This is a specialized cluster type for handling celery brokers. It is used to define which broker instances are available for the test.

Responsibility Scope:

Provude useful methods for managing a cluster of celery brokers.

classmethod default_config() → dict

Default cluster configurations if not overridden by the user.

class pytest_celery.api.broker.CeleryTestBroker(container: CeleryTestContainer, app: Celery = None)

Bases: CeleryTestNode

This is specialized node type for handling celery brokers nodes. It is used to encapsulate a broker instance.

Responsibility Scope:

Handling broker specific requirements and configuration.

classmethod default_config() → dict

Default node configurations if not overridden by the user.

restart(reload_container: bool = True, force: bool = False) → None

Override restart method to update the app broker url with new container values.

pytest_celery.api.container module

The pytest_docker_tools package powers the Docker container management for the plugin.

This module extends the Container class from the package to provide the base API for creating new test containers in a Celery test environment.

class pytest_celery.api.container.CeleryTestContainer(container)

Bases: Container

This is an extension of pytest_docker_tools.wrappers.Container, adding improved control over the container lifecycle.

Responsibility Scope:

Provide useful methods for managing a container instance.

property celeryconfig: dict

Each container is responsible for providing the configuration values required for Celery. This property should be implemented to return the configuration values for the specific container.

Raises:

NotImplementedError – There is no config available by default.

Returns:

Configuration values required for Celery.

Return type:

dict

property client: Any

Provides an API client for interacting with the container, if available.

Subclasses should implement this to return an instance of the client specific to the service running in the container.

Raises:

NotImplementedError – There is not client available by default.

Returns:

Client instance.

Return type:

Any

classmethod command(*args: str, debugpy: bool = False, wait_for_client: bool = True, **kwargs: dict) → list[str]

Override the CMD instruction in the Dockerfile.

This method should be overridden in derived classes to provide the specific command and its arguments required to start the container.

Parameters:

• *args (str) – Additional command-line arguments.

• debugpy (bool) – Enable debugpy. Defaults to False.

• wait_for_client (bool) – Wait for a debugger to be attached. Defaults to True.

• **kwargs (dict) – Additional keyword arguments.

Raises:

NotImplementedError – Rely on the Dockerfile if not set otherwise by default.

Returns:

A list containing the command to run in the container as

the first element, followed by the command-line arguments.

Return type:

list[str]

ready() → bool

Override the default ready() method to wait for the container to be using our waiting logic on top of the default implementation.

When this method returns False, other attempts will be made until the container is ready or stop if other conditions do not allow for it to be ready. The underlying implementation of pytest_docker_tools is currently responsible for error raising.

Returns:

True if ready, False otherwise.

Return type:

bool

property ready_prompt: str | None

A log string that indicates the container has finished starting up and is ready to use.

Returns:

A string to wait for or None for no wait. Defaults to None.

Return type:

str | None

teardown() → None

Teardown the container instance.

pytest_celery.api.setup module

The test setup represents the main entry point for accessing the celery architecture from the test.

This module provides the base API for creating new test setups.

class pytest_celery.api.setup.CeleryTestSetup(worker_cluster: CeleryWorkerCluster, broker_cluster: CeleryBrokerCluster, backend_cluster: CeleryBackendCluster, app: Celery = None)

Bases: object

The test setup is the main entry point for accessing the celery architecture from the test. It is the glue that holds all of the relevant components of the specific test case environment.

Each test case will have its own test setup instance, which is created for the test case by the plugin and is configured for the specific run and its given configurations.

Responsibility Scope:

Provide useful access to the celery architecture from the test.

property app: Celery

The celery app configured for all of the nodes.

property backend: CeleryTestBackend | None

The first backend node of the backend cluster.

property backend_cluster: CeleryBackendCluster | None

The backend cluster of this setup.

property broker: CeleryTestBroker | None

The first broker node of the broker cluster.

property broker_cluster: CeleryBrokerCluster | None

The broker cluster of this setup.

classmethod config(celery_worker_cluster_config: dict) → dict

Creates a configuration dict to be used by app.config_from_object(). The configuration is compiled from all of the nodes in the setup.

Parameters:

celery_worker_cluster_config (dict) – The configuration of the worker cluster.

Returns:

Celery-aware configuration dict.

Return type:

dict

classmethod create_setup_app(celery_setup_config: dict, celery_setup_app_name: str) → Celery

Creates a celery app for the setup.

Parameters:

• celery_setup_config (dict) – Celery configuration dict.

• celery_setup_app_name (str) – Celery app name.

Returns:

Celery app configured for this setup.

Return type:

Celery

classmethod name() → str

The name of the setup.

ready(ping: bool = False, control: bool = False, docker: bool = True) → bool

Check if the setup is ready for testing. All of the confirmations are optional.

Warning

Enabling additional confirmations may hurt performance. Disabling all confirmations may result in false positive results. Use with caution.

Parameters:

• ping (bool, optional) – Confirm via ping task. Defaults to False.

• control (bool, optional) – Confirm via ping control. Defaults to False.

• docker (bool, optional) – Confirm via docker container status. Defaults to True.

Returns:

True if the setup is ready for testing (all required confirmations passed).

Return type:

bool

teardown() → None

Teardown the setup.

classmethod update_app_config(app: Celery) → None

Hook for updating the app configuration in a subclass.

Parameters:

app (Celery) – App after initial configuration.

property worker: CeleryTestWorker | None

The first worker node of the worker cluster.

property worker_cluster: CeleryWorkerCluster | None

The worker cluster of this setup.

pytest_celery.api.worker module

Worker components represents Celery’s worker instances.

This module provides the base API for creating new worker components by defining the base classes for worker nodes and clusters.

class pytest_celery.api.worker.CeleryTestWorker(container: CeleryTestContainer, app: Celery)

Bases: CeleryTestNode

This is specialized node type for handling celery worker nodes. It is used to encapsulate a worker instance.

Responsibility Scope:

Managing a celery worker.

get_running_processes_info(columns: list[str] | None = None, filters: dict[str, str] | None = None) → list[dict]

Get running processes info on the container of this node.

Possible columns:

• pid

• name

• username

• cmdline

• cpu_percent

• memory_percent

• create_time

Parameters:

• columns (list[str] | None, optional) – Columns to query. Defaults to None (all).

• filters (dict[str, str] | None, optional) – Filters to apply. Defaults to None.

Raises:

RuntimeError – If the command fails.

Returns:

List of processes info per requested columns.

Return type:

list[dict]

hostname() → str

Hostname of the worker node.

property log_level: str

Celery log level of this worker node.

property version: str

Celery version of this worker node.

property worker_name: str

Celery test worker node name.

property worker_queue: str

Celery queue for this worker node.

class pytest_celery.api.worker.CeleryWorkerCluster(*nodes: tuple[CeleryTestNode | CeleryTestContainer])

Bases: CeleryTestCluster

This is a specialized cluster type for handling celery workers. It is used to define which worker instances are available for the test.

Responsibility Scope:

Provude useful methods for managing a cluster of celery workers.

property versions: set[str]

Celery versions of all workers in this cluster.

Module contents