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

myutils

Release:

1.0

Date:

May 16, 2024

Description

This example project shows How to inject your own utility functions to add custom APIs to the utils.py module that is being injected into the worker container.

Breakdown

File Structure

The following diagram lists the relevant files in the project.

myutils/
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── myutils.py
│   └── test_myutils.py
└── requirements.txt

myutils.py

For this example, we’ll add a myfunc() using our own myutils.py module.

from pytest_celery.vendors.worker.content.utils import get_running_processes_info  # noqa


def myfunc():
    return "foo"

Note

The get_running_processes_info function is imported from the original utils.py module, which means the plugin need to be installed in the worker container. Otherwise, you may provide your own implementation.

conftest.py

To inject our own module, we use the default_worker_utils_module fixture.

@pytest.fixture
def default_worker_utils_module() -> ModuleType:
    from tests import myutils

    return myutils

Warning

The module will be renamed inside the worker container as utils.py, regardless of the name of the module you provide.

Then, we create an API to access the myfunc() function from the testing environment.

class MyWorker(CeleryTestWorker):

    def myfunc(self) -> bool:
        exit_code, output = self.container.exec_run(
            'python -c "from utils import myfunc; print(myfunc())"',
        )
        if exit_code != 0:
            raise RuntimeError(f"Error: {output}")
        output = output.decode("utf-8")
        return output.strip()


@pytest.fixture
def default_worker_cls() -> type[CeleryTestWorker]:
    return MyWorker

Which uses the new myfunc() function from the provided myutils.py module.

test_myutils.py

Our test file tests the new myfunc() function in three different ways.

Directly.

def test_myfunc():
    assert myfunc() == "foo"

Using a single worker component.

def test_myfunc_in_worker(celery_worker: MyWorker):
    assert celery_worker.myfunc() == "foo"
    assert celery_worker.get_running_processes_info()

Using a full setup.

def test_myfunc_in_setup_worker(celery_setup: CeleryTestSetup):
    celery_worker: MyWorker = celery_setup.worker
    assert celery_worker.myfunc() == "foo"
    assert celery_worker.get_running_processes_info()