# Testing with TestKubeMQBroker (/integrations/faststream/how-to/testing)
`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 [#prerequisites]
* `kubemq-faststream` installed alongside `faststream`, with `KubeMQBroker` and subscribers already defined (see [Configuration & Security](/integrations/faststream/how-to/configuration))
* `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 [#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](#health-check-against-a-live-broker) below.
## A Basic Test [#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.
```python title="test_tasks.py"
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.
Assert inside the handler (as above) for a quick check, or capture state in an outer variable / mock and assert after `br.publish(...)` returns when you need richer assertions on what the handler received.
## What Actually Runs [#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 [#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.
```python title="test_queries.py"
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}'
```
`request()` is only routable for **Commands** and **Queries**. Calling it for an Events, Events Store, or Queues channel raises `FeatureNotSupportedException`, matching the broker's real behaviour. Likewise, `publish_batch()` is supported only for the **Queues** pattern.
For the publish/subscribe patterns, publish and let the handler assert:
```python title="test_events.py"
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 [#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:
```toml title="pyproject.toml"
[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:
```bash
uv add --dev pytest pytest-asyncio
```
With `asyncio_mode = "auto"` the `@pytest.mark.asyncio` decorators are optional — they are shown here so the examples run under either configuration.
## Health Check Against a Live Broker [#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.
```python title="health_check.py"
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:
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.
Keep live health checks separate from your in-memory unit tests — gate them behind a marker (the package uses an `integration` marker for tests that require a live broker) so they do not run in environments without a broker.
## Shipping the Test Utility [#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:
```toml title="pyproject.toml"
[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 [#related]