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

pytest_celery.vendors.worker package

Subpackages

Submodules

pytest_celery.vendors.worker.container module

The pytest-celery plugin provides a set of built-in components called Vendors.

This module is part of the Built-in Celery Worker vendor.

class pytest_celery.vendors.worker.container.CeleryWorkerContainer(container)[source]

Bases: CeleryTestContainer

This is the base class for all Celery worker containers. It is preconfigured for a built-in Celery worker image and should be customized for your own worker image.

The purpose of this class is manipulating the container volume and configurations to warm up the worker container according to the test case requirements.

Responsibility Scope:

Prepare the worker container with the required filesystem, configurations and dependencies of your project.

classmethod app_module() → module[source]

A preconfigured module that contains the Celery app instance.

The module is manipulated at runtime to inject the required configurations from the test case.

classmethod buildargs() → dict[source]

Build arguments for the built-in worker image.

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

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]

classmethod initial_content(worker_tasks: set | None = None, worker_signals: set | None = None, worker_app: Celery | None = None, app_module: module | None = None, utils_module: module | None = None) → dict[source]

Defines the initial content of the worker container.

See more: pytest_docker_tools.volume()

Parameters:

  • worker_tasks (set | None, optional) – Set of tasks modules. Defaults to None.

  • worker_signals (set | None, optional) – Set of signals handlers modules. Defaults to None.

  • worker_app (Celery | None, optional) – Celery app instance. Defaults to None.

  • app_module (ModuleType | None, optional) – app module. Defaults to None.

  • utils_module (ModuleType | None, optional) – utils module. Defaults to None.

Returns:

Custom volume content for the worker container.

Return type:

dict

classmethod initial_env(celery_worker_cluster_config: dict, initial: dict | None = None) → dict[source]

Defines the environment variables for the worker container.

See more: pytest_docker_tools.container()

Parameters:

  • celery_worker_cluster_config (dict) – Environment variables to set.

  • initial (dict | None, optional) – Additional variables. Defaults to None.

Returns:

Environment variables set for the worker container from the test case.

Return type:

dict

classmethod log_level() → str[source]

Celery worker log level.

classmethod ports() → dict | None[source]

Ports to expose from the worker container.

property ready_prompt: str

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

classmethod signals_modules() → set[source]

Signals handlers modules.

This is an optional feature that can be used to inject signals handlers that needs to in the context of the worker container.

classmethod tasks_modules() → set[source]

Tasks modules.

classmethod utils_module() → module[source]

A utility helper module for running python code in the worker container context.

classmethod version() → str[source]

Celery version to use for the worker container.

classmethod worker_name() → str[source]

Celery worker name.

classmethod worker_queue() → str[source]

Celery worker queue.

pytest_celery.vendors.worker.defaults module

The pytest-celery plugin provides a set of built-in components called Vendors.

This module is part of the Built-in Celery Worker vendor.

pytest_celery.vendors.worker.fixtures module

The pytest-celery plugin provides a set of built-in components called Vendors.

This module is part of the Built-in Celery Worker vendor.

pytest_celery.vendors.worker.fixtures.celery_setup_worker(default_worker_cls: type[CeleryTestWorker], default_worker_container: CeleryWorkerContainer, default_worker_app: Celery) → CeleryTestWorker[source]

Creates a CeleryTestWorker instance. Responsible for tearing down the node.

Parameters:

  • default_worker_cls (type[CeleryTestWorker]) – Interface class.

  • default_worker_container (CeleryWorkerContainer) – Instantiated CeleryWorkerContainer.

  • default_worker_app (Celery) – Celery app instance.

pytest_celery.vendors.worker.fixtures.default_worker_app(celery_setup_app: Celery) → Celery[source]

Celery app instance for this worker.

Parameters:

celery_setup_app (Celery) – See also: Vendor Class.

Returns:

Celery app instance.

Return type:

Celery

pytest_celery.vendors.worker.fixtures.default_worker_app_module(default_worker_container_cls: type[CeleryWorkerContainer]) → module[source]

App module for this worker.

Parameters:

default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

App module.

Return type:

ModuleType

pytest_celery.vendors.worker.fixtures.default_worker_celery_log_level(default_worker_container_session_cls: type[CeleryWorkerContainer]) → str[source]

Log level for this worker.

Parameters:

default_worker_container_session_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Log level.

Return type:

str

pytest_celery.vendors.worker.fixtures.default_worker_celery_version(default_worker_container_session_cls: type[CeleryWorkerContainer]) → str[source]

Celery version for this worker.

Parameters:

default_worker_container_session_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Celery version.

Return type:

str

pytest_celery.vendors.worker.fixtures.default_worker_celery_worker_name(default_worker_container_session_cls: type[CeleryWorkerContainer]) → str[source]

Name of the worker.

Parameters:

default_worker_container_session_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Worker name.

Return type:

str

pytest_celery.vendors.worker.fixtures.default_worker_celery_worker_queue(default_worker_container_session_cls: type[CeleryWorkerContainer]) → str[source]

Worker queue for this worker.

Parameters:

default_worker_container_session_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Worker queue.

Return type:

str

pytest_celery.vendors.worker.fixtures.default_worker_cls() → type[CeleryTestWorker][source]

Default worker class. Override to apply custom configuration globally.

See also: Vendor Class.

Returns:

API for managing the vendor’s node.

Return type:

type[CeleryTestWorker]

pytest_celery.vendors.worker.fixtures.default_worker_command(default_worker_container_cls: type[CeleryWorkerContainer]) → list[str][source]

Command to run the container.

Parameters:

default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Docker CMD instruction.

Return type:

list[str]

pytest_celery.vendors.worker.fixtures.default_worker_container_cls() → type[CeleryWorkerContainer][source]

Default worker container class. Override to apply custom configuration globally.

See also: Vendor Class.

Returns:

API for managing the vendor’s container.

Return type:

type[CeleryWorkerContainer]

pytest_celery.vendors.worker.fixtures.default_worker_container_session_cls() → type[CeleryWorkerContainer][source]

Default worker container session class. Override to apply custom configuration globally.

See also: Vendor Class.

Returns:

API for managing the vendor’s container.

Return type:

type[CeleryWorkerContainer]

pytest_celery.vendors.worker.fixtures.default_worker_env(default_worker_container_cls: type[CeleryWorkerContainer], celery_worker_cluster_config: dict) → dict[source]

Environment variables for this worker.

Parameters:
Returns:

Items to pass to the container’s environment.

Return type:

dict

pytest_celery.vendors.worker.fixtures.default_worker_initial_content(default_worker_container_cls: type[CeleryWorkerContainer], default_worker_app_module: module, default_worker_utils_module: module, default_worker_tasks: set, default_worker_signals: set, default_worker_app: Celery) → dict[source]

Initial content for this worker’s volume.

This is applied on a worker container when using the following volume configuration:

default_worker_container = container(
    ...
    volumes={"{default_worker_volume.name}": DEFAULT_WORKER_VOLUME},
    ...
)

Note

More volumes may be added additionally.

Parameters:

  • default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

  • default_worker_app_module (ModuleType) – App module to inject.

  • default_worker_utils_module (ModuleType) – Utils module to inject.

  • default_worker_tasks (set) – Tasks modules to inject.

  • default_worker_signals (set) – Signals modules to inject.

  • default_worker_app (Celery) – Celery app to initialize the worker with.

Returns:

Initial volume content (dict of files).

Return type:

dict

pytest_celery.vendors.worker.fixtures.default_worker_ports(default_worker_container_cls: type[CeleryWorkerContainer]) → dict | None[source]

Port bindings for this vendor.

Parameters:

default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Port bindings.

Return type:

dict

pytest_celery.vendors.worker.fixtures.default_worker_signals(default_worker_container_cls: type[CeleryWorkerContainer]) → set[source]

Signals modules set for this worker.

Parameters:

default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Signals modules.

Return type:

set

pytest_celery.vendors.worker.fixtures.default_worker_tasks(default_worker_container_cls: type[CeleryWorkerContainer]) → set[source]

Tasks modules set for this worker.

Parameters:

default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Tasks modules.

Return type:

set

pytest_celery.vendors.worker.fixtures.default_worker_utils_module(default_worker_container_cls: type[CeleryWorkerContainer]) → module[source]

Utils module for this worker.

Parameters:

default_worker_container_cls (type[CeleryWorkerContainer]) – See also: Vendor Class.

Returns:

Utils module.

Return type:

ModuleType

pytest_celery.vendors.worker.tasks module

The pytest-celery plugin provides a set of built-in components called Vendors.

This module is part of the Built-in Celery Worker vendor.

(task)pytest_celery.vendors.worker.tasks.add(x: 'int | float', y: 'int | float', z: 'int | float | None' = None) → 'int | float'[source]

Pytest-celery internal task.

This task adds two or three numbers together.

Parameters:

  • x (int | float) – The first number.

  • y (int | float) – The second number.

  • z (int | float | None, optional) – The third number. Defaults to None.

Returns:

The sum of the numbers.

Return type:

int | float

(task)pytest_celery.vendors.worker.tasks.add_replaced(x: 'int | float', y: 'int | float', z: 'int | float | None' = None, *, queue: 'str | None' = None) → 'None'[source]

Pytest-celery internal task.

This task replaces itself with the add task for the given arguments.

Parameters:

  • x (int | float) – The first number.

  • y (int | float) – The second number.

  • z (int | float | None, optional) – The third number. Defaults to None.

Raises:

Ignore – Always raises Ignore.

(task)pytest_celery.vendors.worker.tasks.fail(*args: 'tuple') → 'None'[source]

Pytest-celery internal task.

This task raises a RuntimeError with the given arguments.

Parameters:

*args (tuple) – Arguments to pass to the RuntimeError.

Raises:

RuntimeError – Always raises a RuntimeError.

(task)pytest_celery.vendors.worker.tasks.identity(x: 'Any') → 'Any'[source]

Pytest-celery internal task.

This task returns the input as is.

Parameters:

x (Any) – Any value.

Returns:

The input value.

Return type:

Any

(task)pytest_celery.vendors.worker.tasks.noop(*args: 'tuple', **kwargs: 'dict') → 'None'[source]

Pytest-celery internal task.

This is a no-op task that does nothing.

Returns:

Always returns None.

Return type:

None

(task)pytest_celery.vendors.worker.tasks.ping() → 'str'[source]

Pytest-celery internal task.

Used to check if the worker is up and running.

Returns:

Always returns “pong”.

Return type:

str

(task)pytest_celery.vendors.worker.tasks.sleep(seconds: 'float' = 1, **kwargs: 'dict') → 'bool'[source]

Pytest-celery internal task.

This task sleeps for the given number of seconds.

Parameters:

  • seconds (float, optional) – The number of seconds to sleep. Defaults to 1.

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

Returns:

Always returns True.

Return type:

bool

(task)pytest_celery.vendors.worker.tasks.xsum(nums: 'Iterable') → 'int'[source]

Pytest-celery internal task.

This task sums a list of numbers, but also supports nested lists.

Parameters:

nums (Iterable) – A list of numbers or nested lists.

Returns:

The sum of the numbers.

Return type:

int

pytest_celery.vendors.worker.volume module

The pytest-celery plugin provides a set of built-in components called Vendors.

This module is part of the Built-in Celery Worker vendor.

class pytest_celery.vendors.worker.volume.WorkerInitialContent(app_module: module | None = None, utils_module: module | None = None)[source]

Bases: object

This class is responsible for generating the initial content for the worker container volume.

Responsibility Scope:

Prepare the worker container with the required filesystem, configurations and dependencies for your project.

class Parser[source]

Bases: object

Parser for the initial content of the worker container.

app_name(name: str | None = None) → str[source]

Generates the Celery app initialization string.

Parameters:

name (str | None, optional) – The app name. Defaults to None.

Returns:

app = Celery(name)

Return type:

str

config(app: Celery | None = None) → str[source]

Generates the Celery app configuration changes.

Parameters:

app (Celery | None, optional) – Celery app with possibly changed config. Defaults to None.

Raises:

TypeError – If the app.conf.changes property is not a dict.

Returns:

config = {key: value, …} or config = None

Return type:

str

imports_src(modules: set[module]) → dict[source]

Parse the given modules and return a dict with the source code of the modules.

Parameters:

modules (set[ModuleType]) – A set of modules to parse.

Returns:

A dict with the source code of the modules.

Return type:

dict

imports_str(modules: set[module]) → str[source]

Parse the given modules and return a string with the import statements.

Parameters:

modules (set[ModuleType]) – A set of modules to parse.

Returns:

“from module import *” statements.

Return type:

str

add_modules(name: str, modules: set[module]) → None[source]

Adds a set of modules to the initial content.

Parameters:

  • name (str) – Arbitrary unique name for the set of modules.

  • modules (set[ModuleType]) – A set of modules to add.

generate() → dict[source]

Generates the initial content for the worker container.

Returns:

Initial content volume for the worker container.

Return type:

dict

set_app_module(app_module: module | None = None) → None[source]

Injects an app module into the initial content.

set_app_name(name: str | None = None) → None[source]

Sets the Celery app name.

Parameters:

name (str | None, optional) – The app name. Defaults to None.

set_config_from_object(app: Celery | None = None) → None[source]

Sets the Celery app configuration from the given app.

Parameters:

app (Celery | None, optional) – Celery app with possibly changed config. Defaults to None.

set_utils_module(utils_module: module | None = None) → None[source]

Injects a utils module into the initial content.

Module contents

The pytest-celery plugin provides a set of built-in components called Vendors.

This module is part of the Built-in Celery Worker vendor.