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

vhost

Release:

1.0

Date:

May 16, 2024

Description

This example project demonstrates how to use a single Redis container as both a broker and a result backend, using different vhosts for each purpose.

Breakdown

File Structure

The following diagram lists the relevant files in the project.

rabbitmq_management/
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_vhost.py
└── requirements.txt

conftest.py

We create our own Redis container, and then use it as both a broker and a result backend.

redis_image = fetch(repository=REDIS_IMAGE)
redis_test_container: RedisContainer = container(
    image="{redis_image.id}",
    ports=REDIS_PORTS,
    environment=REDIS_ENV,
    network="{default_pytest_celery_network.name}",
    wrapper_class=RedisContainer,
    timeout=REDIS_CONTAINER_TIMEOUT,
)

As the default vhost is “0”, we can use it as the broker, and create a new vhost for the result backend.

@pytest.fixture
def redis_broker(redis_test_container: RedisContainer) -> RedisTestBroker:
    broker = RedisTestBroker(redis_test_container)
    yield broker
    broker.teardown()


@pytest.fixture
def celery_broker_cluster(redis_broker: RedisTestBroker) -> CeleryBrokerCluster:
    cluster = CeleryBrokerCluster(redis_broker)
    yield cluster
    cluster.teardown()

For the backend, we need to change some settings to use a different vhost, so we create our own Redis backend and work it out there.

class MyRedisTestBackend(RedisTestBackend):
    def config(self, *args: tuple, **kwargs: dict) -> dict:
        return super().config(vhost=1, *args, **kwargs)

Lastly, we add our backend that uses the same Redis container as the broker and our MyRedisTestBackend class.

@pytest.fixture
def redis_backend(redis_test_container: RedisContainer) -> MyRedisTestBackend:
    backend = MyRedisTestBackend(redis_test_container)
    yield backend
    backend.teardown()


@pytest.fixture
def celery_backend_cluster(redis_backend: MyRedisTestBackend) -> CeleryBackendCluster:
    cluster = CeleryBackendCluster(redis_backend)
    yield cluster
    cluster.teardown()

test_vhost.py

We can now run tests that will share the same Redis container for both the broker and the result backend components.

examples.vhost.tests.test_vhost.py

from pytest_celery import RESULT_TIMEOUT
from pytest_celery import CeleryTestSetup
from pytest_celery import ping


class TestVhost:
    def test_ping(self, celery_setup: CeleryTestSetup):
        assert ping.s().delay().get(timeout=RESULT_TIMEOUT) == "pong"

    def test_vhost(self, celery_setup: CeleryTestSetup):
        assert celery_setup.app.conf.broker_url[:-1] == celery_setup.app.conf.result_backend[:-1]
        assert celery_setup.app.conf.broker_url.endswith("/0")
        assert celery_setup.app.conf.result_backend.endswith("/1")

    def test_single_redis(self, celery_setup: CeleryTestSetup):
        assert celery_setup.broker.container.id == celery_setup.backend.container.id