Testes de integração MQTT: teste sua aplicação com um broker real

Testes de integração MQTT rodam sua aplicação contra um broker real: seu código conecta, assina e publica, e o teste verifica as mensagens que realmente atravessam a rede. Eles pegam o que testes unitários com mocks não conseguem: filtros de topic errados, confirmações de QoS ausentes, estado retained, negações de ACL e bugs de reconexão.

Este guia mostra padrões que funcionam com pytest (paho-mqtt 2.x) e Jest (MQTT.js 5), e as cinco regras que mantêm esses testes rápidos e sem instabilidade.

Unitário, integração ou end-to-end?

Mantenha os testes com mocks para a lógica pura — brokers mock para testes unitários cobre essa camada. Tudo o que depende de como um broker se comporta (wildcards, retained messages, reentrega com QoS 1, expiração de sessão) pertence aos testes de integração contra um broker real.

Escolhendo o broker de testes

Você tem três opções práticas:

• Mosquitto local (instalado ou via Docker): rápido e privado, mas todo desenvolvedor e todo runner de CI precisa iniciá-lo.
• Container efêmero por execução: totalmente isolado e descartável; veja testes MQTT com Testcontainers.
• Broker hospedado: sem configuração, com TLS e autenticação reais e o mesmo endpoint em notebooks, CI e dispositivos de staging.

Seu código de teste não deveria se importar com qual deles está falando. Leia host, porta e credenciais de variáveis de ambiente e use como padrão algo que funcione de cara:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
# MQTT_PORT=8883 for TLS; set MQTT_USER / MQTT_PASS for a private broker

As cinco regras para testes MQTT estáveis

1. Prefixo de topic único por teste. Monte cada topic a partir de testmqtt/it/<run-id>/<random-id>. Assim workers paralelos, jobs de CI e desenvolvedores nunca veem as mensagens uns dos outros.
2. Client ID único por conexão. Duas conexões com o mesmo ID derrubam uma à outra — uma fonte clássica de testes que perdem mensagens "aleatoriamente".
3. Espere eventos, nunca durma. Espere o CONNACK antes de assinar, o SUBACK antes de publicar e a mensagem com um timeout. Um sleep(1) é longo demais (suíte lenta) ou curto demais (suíte instável).
4. Limpe o que sobrevive ao teste. Retained messages e sessões persistentes sobrevivem a desconexões. Limpe os topics retained com uma publicação retained vazia e use sessões limpas, a menos que esteja testando sessões.
5. Verifique conteúdo, não tempo nem ordem global. O MQTT só garante ordem por topic e QoS a partir de um mesmo publisher. Correlacione mensagens com um ID no payload.

pytest: uma fixture reutilizável de cliente

A fixture abaixo dá a cada teste um helper conectado que espera as confirmações e coleta mensagens em uma fila thread-safe. Coloque-a em conftest.py:

import os, queue, threading, time, uuid
import pytest
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN_ID = os.getenv("GITHUB_RUN_ID", uuid.uuid4().hex[:8])


class MqttProbe:
    def __init__(self):
        self.messages = queue.Queue()
        self.retained = set()
        self._connected = threading.Event()
        self._subacks = {}
        self._cond = threading.Condition()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"it-{uuid.uuid4().hex[:10]}")
        self.client.on_connect = lambda c, u, f, rc, p: self._connected.set()
        self.client.on_subscribe = self._on_subscribe
        self.client.on_message = lambda c, u, msg: self.messages.put(msg)
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        if PORT == 8883:
            self.client.tls_set()

    def _on_subscribe(self, client, userdata, mid, reason_codes, properties):
        with self._cond:
            self._subacks[mid] = reason_codes
            self._cond.notify_all()

    def connect(self, timeout=5):
        self.client.connect(HOST, PORT, keepalive=30)
        self.client.loop_start()
        assert self._connected.wait(timeout), "no CONNACK from broker"

    def subscribe(self, topic, qos=1, timeout=5):
        _, mid = self.client.subscribe(topic, qos)
        with self._cond:
            assert self._cond.wait_for(lambda: mid in self._subacks, timeout), f"no SUBACK for {topic}"
            code = self._subacks.pop(mid)[0]
        assert not code.is_failure, f"subscribe to {topic} refused: {code}"

    def publish(self, topic, payload, qos=1, retain=False, timeout=5):
        info = self.client.publish(topic, payload, qos=qos, retain=retain)
        info.wait_for_publish(timeout)
        assert info.is_published(), f"publish to {topic} not acknowledged"
        if retain:
            self.retained.add(topic)

    def expect(self, topic, timeout=5):
        deadline = time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                msg = self.messages.get(timeout=left)
            except queue.Empty:
                break
            if msg.topic == topic:
                return msg
        raise AssertionError(f"no message on {topic} within {timeout}s")

    def close(self):
        for topic in self.retained:  # clear retained state left by the test
            self.client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
        self.client.disconnect()
        self.client.loop_stop()


@pytest.fixture
def prefix():
    return f"testmqtt/it/{RUN_ID}/{uuid.uuid4().hex[:8]}"


@pytest.fixture
def mqtt_probe():
    probe = MqttProbe()
    probe.connect()
    yield probe
    probe.close()

Um teste para um serviço da aplicação — aqui, um termostato que escuta comandos e reporta seu estado — passa a ser lido como uma especificação:

import json
from conftest import HOST, PORT
from myapp.thermostat import ThermostatService


def test_setpoint_command_updates_state(mqtt_probe, prefix):
    service = ThermostatService(host=HOST, port=PORT, prefix=prefix)
    service.start()  # returns once the service has its own SUBACK
    try:
        mqtt_probe.subscribe(f"{prefix}/state")
        mqtt_probe.publish(f"{prefix}/cmd/setpoint", json.dumps({"value": 21.5}))

        msg = mqtt_probe.expect(f"{prefix}/state")
        assert json.loads(msg.payload)["setpoint"] == 21.5
    finally:
        service.stop()

O prefix é passado para a aplicação, e essa é a decisão de design mais útil para testabilidade: se o seu serviço monta os topics a partir de uma raiz configurável, cada teste pode rodar no próprio namespace. Repare que service.start() bloqueia até o serviço estar inscrito; se a sua aplicação não expõe um sinal de prontidão assim, adicione um (um evento, um callback ou uma mensagem de status online retained). Caso contrário, o comando do teste pode chegar antes de o serviço estar escutando. Fixtures do pytest para MQTT mostra mais variações de fixtures, incluindo brokers com escopo de sessão.

Jest: helpers com promises no MQTT.js 5

O MQTT.js 5 traz connectAsync, subscribeAsync, publishAsync e endAsync, que deixam os testes lineares. O único helper de que você precisa é "esperar a próxima mensagem em um topic":

const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

const URL = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const opts = () => ({
  clientId: `it-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  reconnectPeriod: 0,
});

function nextMessage(client, topic, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const onMessage = (t, payload, packet) => {
      if (t !== topic) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ payload: payload.toString(), retain: packet.retain });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on ${topic} within ${timeoutMs} ms`));
    }, timeoutMs);
    client.on('message', onMessage);
  });
}

let publisher;
let prefix;

beforeAll(async () => {
  publisher = await mqtt.connectAsync(URL, opts());
});
afterAll(() => publisher.endAsync());
beforeEach(() => {
  prefix = `testmqtt/it/${process.env.GITHUB_RUN_ID ?? 'local'}/${randomUUID().slice(0, 8)}`;
});

test('late subscribers receive the retained device config', async () => {
  const topic = `${prefix}/config`;
  await publisher.publishAsync(topic, JSON.stringify({ interval: 30 }), { qos: 1, retain: true });

  const device = await mqtt.connectAsync(URL, opts());
  try {
    const received = nextMessage(device, topic); // listen before subscribing
    await device.subscribeAsync(topic, { qos: 1 });
    const msg = await received;
    expect(msg.retain).toBe(true);
    expect(JSON.parse(msg.payload)).toEqual({ interval: 30 });
  } finally {
    await device.endAsync();
    await publisher.publishAsync(topic, '', { qos: 1, retain: true }); // clear retained
  }
}, 15000);

Dois detalhes evitam os problemas mais comuns com Jest. O listener é registrado antes de subscribeAsync, porque o broker envia as retained messages logo após o SUBACK — às vezes antes de a promise aguardada ser resolvida. E todo cliente é fechado com endAsync(), para que o Jest não trave com "Jest did not exit one second after the test run has completed". Mais padrões de Jest em testando MQTT com Jest.

Teste também os caminhos de falha

O caminho feliz é a parte menos interessante de um cliente MQTT. Os testes de integração são onde você pode provar, com pouco custo, o comportamento que importa em campo:

• Payloads malformados. Publique JSON inválido no seu topic de comandos e verifique que o serviço registra um erro e continua rodando, em vez de derrubar seu loop de rede.
• Autorização. Conecte com um usuário que não tem permissão para publicar em um topic e verifique que seu código reporta a recusa. Com MQTT 5, uma publicação QoS 1 retorna um reason code no PUBACK; com MQTT 3.1.1, o broker normalmente descarta a mensagem ou fecha a conexão.
• Reconexões. Desconecte o cliente da aplicação (ou reinicie um broker local) e verifique que ele reassina e processa o próximo comando. Testes de reconexão MQTT mostra como automatizar isso.

Cada um desses casos leva poucas linhas com as fixtures acima e pega bugs que, de outra forma, só aparecem nos logs de produção.

Checklist de testes instáveis

Se uma assinatura for recusada, o reason code do SUBACK diz o motivo — 0x87 Not authorized geralmente significa que a ACL do usuário de testes não cobre o prefixo. Decodifique qualquer código com a consulta de reason codes MQTT.

Testando os testes

Antes de confiar em uma suíte nova, observe-a de fora. Assine testmqtt/it/# no cliente MQTT online e rode os testes: você deve ver o prefixo de cada teste aparecer, as mensagens fluírem e, no final, os topics retained serem limpos (uma retained message vazia). Depois rode a suíte duas vezes em paralelo — por exemplo pytest -n 4 com pytest-xdist, ou dois jobs de CI — e confirme que nada muda.

Veja o tráfego dos testes de integração

Testes de integração contra um broker real são a forma mais barata de provar que seu código MQTT se comporta como em produção. Isole cada teste por topic e client ID, espere as confirmações em vez de dormir, limpe o estado retained e a suíte vai continuar verde pelos motivos certos. Quando quiser rodá-la a cada push, continue com testando MQTT em CI com GitHub Actions.

Perguntas frequentes