Skip to content

TestMQTTBroker

faststream.mqtt.TestMQTTBroker #

TestMQTTBroker(
    broker: MQTTBroker,
    /,
    *,
    with_real: bool = False,
    connect_only: bool | None = None,
)

TestMQTTBroker(
    *brokers: MQTTBroker,
    with_real: bool = False,
    connect_only: bool | None = None,
)

TestMQTTBroker(
    *brokers: MQTTBroker,
    with_real: bool = False,
    connect_only: bool | None = None,
)

Bases: TestBroker[MQTTBroker, EnterType]

In-memory test double for MQTTBroker.

Routes published messages to matching subscribers without a real MQTT connection, using MQTT wildcard rules for topic matching. Messages are encoded in the same wire format as the configured broker version (V311 envelope or V5 PublishProperties).

Usage::

async with TestMQTTBroker(broker) as br:
    await br.publish("hello", "sensors/temp")
    handler.mock.assert_called_once_with("hello")
Source code in faststream/mqtt/testing.py 
106
107
108
109
110
111
112
113
114
115
116
def __init__(
    self,
    *brokers: MQTTBroker,
    with_real: bool = False,
    connect_only: bool | None = None,
) -> None:
    super().__init__(
        *brokers,
        with_real=with_real,
        connect_only=connect_only,
    )

with_real instance-attribute #

with_real = with_real

brokers instance-attribute #

brokers = brokers

connect_only instance-attribute #

connect_only = connect_only

create_publisher_fake_subscriber #

create_publisher_fake_subscriber(
    broker: MQTTBroker, publisher: MQTTPublisher
) -> tuple[MQTTBaseSubscriber, bool]
Source code in faststream/mqtt/testing.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def create_publisher_fake_subscriber(
    self,
    broker: MQTTBroker,
    publisher: "MQTTPublisher",
) -> tuple["MQTTBaseSubscriber", bool]:
    sub: MQTTBaseSubscriber | None = None
    for handler in (s for b in self.brokers for s in b.subscribers):
        handler = cast("MQTTBaseSubscriber", handler)
        if mqtt_topic_matches(handler.topic, publisher.topic):
            sub = handler
            break

    if sub is None:
        is_real = False
        sub = broker.subscriber(publisher.topic, persistent=False)
        # Apply the correct version parser so fake subs match FakeProducer output.
        parser = sub._build_parser()
        sub._parser = parser.parse_message
        sub._decoder = parser.decode_message
    else:
        is_real = True

    return sub, is_real