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()