Testing with TestKubeMQBroker

TestKubeMQBroker lets you unit-test FastStream handlers without a running KubeMQ broker. It wraps your real KubeMQBroker in an in-memory message router: when you publish inside the test context, the message is delivered directly to any subscriber whose pattern and channel match — no gRPC connection, no network, no Docker container.

That makes the whole pattern suite — Events, Events Store, Queues, Commands, and Queries — testable in milliseconds, in CI, with nothing to stand up first.

Prerequisites

• kubemq-faststream installed alongside faststream, with KubeMQBroker and subscribers already defined (see Configuration & Security)
• pytest and pytest-asyncio installed (uv add --dev pytest pytest-asyncio) — no live broker is needed for the in-memory tests below

Why an In-Memory Broker

A normal KubeMQBroker opens three gRPC channels to a live broker on startup. In a unit test that is slow, flaky, and forces every contributor (and your CI runner) to have a broker on localhost:50000.

TestKubeMQBroker replaces the broker's producer and connection with in-memory fakes. Published messages are routed straight to the matching subscriber handlers, so your tests exercise handler logic deterministically and in isolation. You only need a real broker for end-to-end smoke checks — covered in Health Check Against a Live Broker below.

A Basic Test

Import both KubeMQBroker and TestKubeMQBroker from kubemq_faststream. Define your broker and subscribers exactly as you would in production, then enter the TestKubeMQBroker context manager and publish to a channel a subscriber is listening on.

import pytest

from kubemq_faststream import KubeMQBroker, TestKubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(queues="tasks")
async def handle_task(msg: dict) -> None:
    assert msg["type"] == "test"


@pytest.mark.asyncio
async def test_task_handling():
    async with TestKubeMQBroker(broker) as br:
        await br.publish({"type": "test"}, queues="tasks")

The URL passed to KubeMQBroker is never dialed — inside the async with block the connection is mocked, so the address is irrelevant for the test. br is the same broker instance, now backed by the in-memory router; call br.publish(...), br.publish_batch(...), or br.request(...) on it just as you would the real thing.

What Actually Runs

TestKubeMQBroker does not stub your handlers — it stubs the transport. Every published message flows through the full FastStream pipeline: the parser, the decoder, and all broker and subscriber middleware execute in-memory before your handler is invoked, exactly as they would against a live broker.

That means a test will catch the same parsing, decoding, validation, and middleware bugs you would otherwise only discover at runtime — you are testing the real message path, just without the network.

Coverage Across Patterns

All five KubeMQ messaging patterns are supported in-memory. The fake producer routes by pattern and channel, so the keyword you use (events, events_store, queues, commands, queries) decides which subscriber receives the message.

Request-reply works too: br.request(...) is valid for the Commands and Queries patterns and returns the handler's result. The handler return value is delivered back to the caller in-memory.

import pytest

from kubemq_faststream import KubeMQBroker, TestKubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")


@broker.subscriber(queries="product.lookup")
async def lookup_product(msg: dict) -> dict:
    return {"name": "Widget", "price": 29.99}


@pytest.mark.asyncio
async def test_query_request():
    async with TestKubeMQBroker(broker) as br:
        result = await br.request({"product_id": "SKU-100"}, queries="product.lookup")
        assert result.body == b'{"name": "Widget", "price": 29.99}'

For the publish/subscribe patterns, publish and let the handler assert:

import pytest

from kubemq_faststream import KubeMQBroker, TestKubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")
received: list[dict] = []


@broker.subscriber(events="notifications")
async def on_notification(msg: dict) -> None:
    received.append(msg)


@pytest.mark.asyncio
async def test_event_delivery():
    async with TestKubeMQBroker(broker) as br:
        await br.publish({"type": "alert", "text": "Hello"}, events="notifications")
    assert received == [{"type": "alert", "text": "Hello"}]

Configuring pytest-asyncio

TestKubeMQBroker is an async context manager, so your tests are async def and need pytest-asyncio. The package's own test suite sets asyncio's auto mode, which collects async def tests without a per-test @pytest.mark.asyncio decorator:

[tool.pytest.ini_options]
asyncio_mode = "auto"

Mirror this in your own project's pyproject.toml (or pytest.ini) and you can drop the @pytest.mark.asyncio markers from the examples above. Add pytest-asyncio to your dev dependencies:

uv add --dev pytest pytest-asyncio

Health Check Against a Live Broker

In-memory tests never touch a real broker, so they cannot tell you whether your deployment can actually reach KubeMQ. For that, use the broker's health check as a smoke test against a live instance.

Enter the broker as a context manager — which opens the real connection — and call await broker.ping(...) with a timeout. It returns True when the broker responds in time and False otherwise.

from kubemq_faststream import KubeMQBroker

broker = KubeMQBroker("kubemq://localhost:50000")

async with broker:
    is_healthy = await broker.ping(timeout=5.0)
    print(f"Broker healthy: {is_healthy}")

This is a live check — unlike the in-memory tests above, it requires a reachable broker. Start one locally with Docker:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

KubeMQ exposes native gRPC on port 50000; kubemq-faststream connects there directly, so no HTTP connector or server-side enable flag is involved. See the runnable examples/connection/health_check.py script in the repository for a full app that pings before publishing.

Shipping the Test Utility

TestKubeMQBroker lives in kubemq_faststream.testing and is intended only for tests. In the package's own coverage configuration, testing.py is omitted from coverage measurement because it is a test helper, not production code:

[tool.coverage.run]
source = ["kubemq_faststream"]
omit = ["*/testing.py"]

Treat it the same way in your projects: import TestKubeMQBroker from your test modules only, and keep it out of application startup paths.

Related