blackbull.testing.mqtt

blackbull.testing.mqtt

In-process MQTT tap tests.

Taps run inline and failures propagate regardless of configured tap mode. This does not exercise routing, CONNECT, QoS or retained messages.

MQTTTestBroker

Test environment for an app's MQTT taps (no socket, no client).

Parameters

app: The BlackBull app the MQTT extension is registered on.

Async-only: taps are coroutines and publish awaits them on the caller's event loop, so tests are written as async def. A synchronous façade of the blackbull.testing.native.NativeClient kind would have no socket, broker or background loop to bridge to.

async with is supported for a consistent test idiom, but the broker owns no resource, so entering and exiting are no-ops.

__init__(app)

Bind to app's MQTT extension.

The extension must already be registered on the app; without it this raises RuntimeError at construction rather than failing later on a publish.

publish(topic, payload=b'', *, qos=0, retain=False, properties=None) async

Dispatch one PUBLISH to every matching tap and await them.

topic is matched against the registered filters — + / # wildcards and {name} captures apply — and captured segments are bound as keyword arguments. Taps run inline and are complete when this returns.

A tap that raises propagates its exception, and the remaining taps in this publish are skipped — the first-failure semantics a sequential test loop already has. Production, by contrast, logs and isolates a raising tap; publish deliberately does not.

qos, retain and properties are passed through to the Message without validation, so a test may exercise values the wire layer would reject. Wire-level QoS validation is the conformance suite's territory, not this helper's.