Fixtures do pytest para MQTT: testes isolados com broker

A maioria das suítes de teste MQTT começa do mesmo jeito: um teste que conecta, dorme dois segundos, publica, dorme de novo e confere uma lista global. Funciona no primeiro dia e fica flaky lá pelo décimo. As fixtures do pytest resolvem isso levando a conexão, a nomeação de topics, a espera e a limpeza para alguns blocos reutilizáveis, de modo que o corpo de cada teste vira três linhas de intenção.

Este guia monta um conftest.py para paho-mqtt 2.x com broker escolhido por variáveis de ambiente, uma factory de clientes conectados com teardown automático, um topic único por teste, um coletor de mensagens com timeouts e um limpador de mensagens retidas.

O que um bom conjunto de fixtures MQTT oferece

Fixture	Escopo	Responsabilidade
`broker`	session	Host, porta, credenciais e TLS a partir de variáveis de ambiente; skip se inacessível
`run_id`	session	ID aleatório que separa esta execução de qualquer outra no mesmo broker
`topic`	function	Um prefixo de topic único por teste, derivado do nome do teste
`mqtt_client`	function	Factory que devolve clientes conectados e os desconecta depois
`subscriber`	function	Factory: assina qualquer filtro e devolve um coletor
`collector`	function	Um subscriber no topic do próprio teste, com `wait_for()` e timeout
`retained`	function	Registra topics retidos e os limpa no teardown

Três propriedades importam: isolamento (nenhum teste vê as mensagens de outro), nada de sleeps fixos (espere eventos, com prazo) e limpeza garantida (o teardown roda mesmo quando o teste falha).

O conftest.py

Instale as dependências: pip install pytest paho-mqtt. Depois coloque isto em tests/conftest.py:

tests/conftest.py

import os, re, socket, threading, uuid
from dataclasses import dataclass

import paho.mqtt.client as mqtt
import pytest


@dataclass(frozen=True)
class BrokerConfig:
    host: str
    port: int
    username: str | None
    password: str | None
    tls: bool


@pytest.fixture(scope="session")
def broker():
    cfg = BrokerConfig(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        username=os.getenv("MQTT_USER"),
        password=os.getenv("MQTT_PASS"),
        tls=os.getenv("MQTT_TLS") == "1",
    )
    try:
        socket.create_connection((cfg.host, cfg.port), timeout=3).close()
    except OSError as exc:
        pytest.skip(f"MQTT broker {cfg.host}:{cfg.port} unreachable: {exc}")
    return cfg


@pytest.fixture(scope="session")
def run_id():
    return uuid.uuid4().hex[:8]


@pytest.fixture
def topic(request, run_id):
    name = re.sub(r"[^A-Za-z0-9_-]", "_", request.node.name)
    return f"testmqtt/pytest/{run_id}/{name}"


@pytest.fixture
def mqtt_client(broker):
    clients = []

    def make(client_id=None, protocol=mqtt.MQTTv311, **connect_kwargs):
        client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=client_id or f"pytest-{uuid.uuid4().hex[:10]}",
            protocol=protocol,
        )
        if broker.username:
            client.username_pw_set(broker.username, broker.password)
        if broker.tls:
            client.tls_set()
        clients.append(client)

        connected, result = threading.Event(), {}

        def on_connect(c, userdata, flags, reason_code, properties):
            result["rc"] = reason_code
            connected.set()

        client.on_connect = on_connect
        client.connect(broker.host, broker.port, keepalive=30, **connect_kwargs)
        client.loop_start()
        assert connected.wait(10), f"no CONNACK from {broker.host}:{broker.port}"
        assert not result["rc"].is_failure, f"connection refused: {result['rc']}"
        return client

    yield make

    for client in clients:
        client.disconnect()
        client.loop_stop()

Algumas escolhas deliberadas:

O broker vem do ambiente. Localmente, MQTT_HOST=localhost aponta para um container Mosquitto; no CI os mesmos testes rodam contra um broker do Testcontainers ou uma instância de staging. O padrão é o endpoint compartilhado broker.connectmq.com, suficiente para execuções rápidas.
Broker inacessível → skip, não falha. Quem não está com o Docker rodando recebe um motivo de skip claro em vez de trinta erros de conexão. Se quiser que o CI falhe de vez, use uma flag que transforma o skip em pytest.fail.
A factory espera o CONNACK. connect() só envia o pacote CONNECT; o Event espera a resposta do broker e a asserção mostra o reason code se o broker recusar.
O teardown desconecta antes de parar o loop, para que o pacote DISCONNECT seja de fato enviado e o broker não publique um Last Will de um cliente de teste.
Client IDs aleatórios. Dois testes com o mesmo ID derrubariam um ao outro do broker — veja boas práticas de client ID ou gere IDs legíveis com o gerador de client ID.

Esperando mensagens sem sleep

O coletor é a peça que tira o time.sleep() dos seus testes. Ele adiciona as mensagens recebidas sob uma variável de condição e deixa o teste esperar até que um certo número tenha chegado:

tests/conftest.py (continued)

class MessageCollector:
    def __init__(self):
        self.messages = []
        self._cond = threading.Condition()

    def __call__(self, client, userdata, msg):  # used as on_message
        with self._cond:
            self.messages.append(msg)
            self._cond.notify_all()

    def wait_for(self, count=1, timeout=5.0):
        with self._cond:
            if not self._cond.wait_for(lambda: len(self.messages) >= count, timeout):
                got = [m.topic for m in self.messages]
                raise AssertionError(f"expected {count} message(s) in {timeout}s, got {len(got)}: {got}")
            return list(self.messages[:count])

    def assert_silent(self, duration=1.0):
        with self._cond:
            self._cond.wait_for(lambda: len(self.messages) > 0, duration)
            assert not self.messages, f"unexpected messages: {[m.topic for m in self.messages]}"


def subscribe(client, topic_filter, qos=1, timeout=5.0):
    done, result = threading.Event(), {}

    def on_subscribe(c, userdata, mid, reason_code_list, properties):
        result["codes"] = reason_code_list
        done.set()

    client.on_subscribe = on_subscribe
    client.subscribe(topic_filter, qos=qos)
    assert done.wait(timeout), f"no SUBACK for {topic_filter}"
    assert not result["codes"][0].is_failure, f"subscription refused: {result['codes'][0]}"


@pytest.fixture
def subscriber(mqtt_client):
    def make(topic_filter, qos=1):
        messages = MessageCollector()
        sub = mqtt_client()
        sub.on_message = messages
        subscribe(sub, topic_filter, qos)
        return messages
    return make


@pytest.fixture
def collector(subscriber, topic):
    return subscriber(f"{topic}/#")


@pytest.fixture
def retained(mqtt_client):
    topics = []
    yield topics.append
    if topics:
        cleaner = mqtt_client()
        for t in topics:
            cleaner.publish(t, b"", qos=1, retain=True).wait_for_publish(5)

Esperar o SUBACK importa tanto quanto esperar as mensagens: se um teste publica antes de a assinatura estar ativa, o broker descarta a mensagem e o teste falha aleatoriamente. As fixtures subscriber e collector só retornam depois que o broker confirmou a assinatura.

assert_silent() cobre o caso negativo — por exemplo, que uma ACL nega um topic ou que uma mensagem com retain=False não é reenviada. Ela espera a duração inteira e falha assim que algo chega.

Testes que se leem como especificações

Com as fixtures prontas, o corpo dos testes se reduz à intenção:

tests/test_telemetry.py

import json
import pytest


@pytest.mark.parametrize("qos", [0, 1, 2])
def test_roundtrip(mqtt_client, collector, topic, qos):
    pub = mqtt_client()
    pub.publish(f"{topic}/temp", "21.5", qos=qos).wait_for_publish(5)

    [msg] = collector.wait_for(1)
    assert msg.topic == f"{topic}/temp"
    assert msg.payload == b"21.5"


def test_ordering_within_one_topic(mqtt_client, collector, topic):
    pub = mqtt_client()
    for i in range(20):
        pub.publish(f"{topic}/seq", json.dumps({"i": i}), qos=1)

    received = [json.loads(m.payload)["i"] for m in collector.wait_for(20, timeout=10)]
    assert received == list(range(20))


def test_retained_status_reaches_late_subscriber(mqtt_client, subscriber, topic, retained):
    status = f"{topic}/status"
    retained(status)
    mqtt_client().publish(status, "online", qos=1, retain=True).wait_for_publish(5)

    late = subscriber(status)          # subscribes after the publish
    [msg] = late.wait_for(1)
    assert msg.payload == b"online" and msg.retain

Cada teste recebe seu próprio prefixo de topic, como testmqtt/pytest/3f9a1c2e/test_roundtrip_1_ (a partir de test_roundtrip[1]), então variantes parametrizadas nunca se enxergam. Durante a depuração, assine testmqtt/pytest/# no cliente MQTT online para ver o tráfego da suíte ao vivo.

O teste de retidas registra o topic antes de publicar. Mesmo que uma asserção falhe, o teardown de retained publica um payload retido vazio e a próxima execução começa limpa — veja testar mensagens retidas para entender por que estado retido antigo é uma fonte comum de resultados flaky.

Opções do MQTT 5 pela factory

Como argumentos nomeados extras vão direto para connect(), os recursos do MQTT 5 não precisam de uma fixture nova:

tests/test_sessions.py

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties


def test_mqtt5_connect_with_session_expiry(mqtt_client):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 300
    client = mqtt_client(protocol=mqtt.MQTTv5, clean_start=False, properties=props)
    assert client.is_connected()

Use o mesmo padrão para testes de user properties, request-response ou message expiry.

Rodando a suíte

pytest.ini

[pytest]
testpaths = tests
markers =
    mqtt: tests that need a live MQTT broker

Run locally and in CI

pytest -q                                            # default broker
MQTT_HOST=localhost pytest -q                        # local Mosquitto
MQTT_PORT=8883 MQTT_TLS=1 MQTT_USER=$MQTT_USER MQTT_PASS=$MQTT_PASS pytest -q
pip install pytest-xdist && pytest -n 4              # parallel workers

Adicione o pytest-timeout com um timeout = 30 global no pytest.ini como última linha de defesa: um teste preso esperando uma mensagem que nunca chega falha em vez de travar o job do CI. Para um pipeline completo, veja testes MQTT no GitHub Actions.

Armadilhas

Asserções dentro de callbacks. Um assert em on_message roda na thread de rede do paho; a falha é engolida e o teste passa. Colete no callback, verifique no teste.
Clientes com escopo de sessão. São mais rápidos, mas uma assinatura feita por um teste continua entregando mensagens no próximo. Se compartilhar um cliente, cancele as assinaturas e redefina os callbacks em um wrapper com escopo de função.
Nomes de topic com curingas. IDs de parametrize podem conter caracteres como + ou #; o re.sub() da fixture topic os remove.
Testes unitários que não precisam de broker. Lógica que só formata payloads ou escolhe topics pertence a testes unitários rápidos com um cliente falso — veja brokers mock para testes unitários.

Uma dúzia de linhas de fixtures se paga rapidinho: os testes param de dormir, param de interferir uns nos outros e limpam a própria bagunça. Mantenha o conftest.py no seu repositório, escolha o broker por variáveis de ambiente, e a mesma suíte roda no notebook, no CI e contra staging sem alterações.

Perguntas frequentes

As fixtures MQTT devem ter escopo de sessão ou de função?

Mantenha a configuração do broker com escopo de sessão e os clientes com escopo de função. Um cliente novo por teste custa um round trip de CONNECT, mas garante que nenhuma assinatura, callback ou mensagem enfileirada vaze de um teste para o próximo. Só compartilhe um cliente entre testes se o tempo de conexão for um problema medido.

Como esperar uma mensagem MQTT no pytest sem sleep?

Use um coletor que adiciona as mensagens sob um threading.Condition e expõe wait_for(count, timeout). O teste bloqueia só até a mensagem chegar e falha com uma mensagem clara no timeout, em vez de dormir um tempo fixo e torcer.

Testes MQTT podem rodar em paralelo com pytest-xdist?

Sim, se cada teste usar seu próprio topic e seu próprio client ID. Derivar o topic de um ID por execução mais o nome do nó de teste, e gerar client IDs aleatórios, impede que workers paralelos recebam as mensagens uns dos outros ou se derrubem do broker.

Fixtures do pytest para MQTT: testes limpos, rápidos e isolados