Como Testar Clientes MQTT: Testes Unitários, de Integração e End-to-End

Testar clientes MQTT é complicado porque o comportamento interessante acontece através de uma rede, dentro de callbacks, na thread de outra pessoa. Este guia mostra uma abordagem em camadas que funciona: testes unitários rápidos com um cliente simulado, testes de integração contra um broker descartável em contêiner no CI e um smoke test end-to-end curto contra um broker real na nuvem antes do release.

Por que é difícil testar código cliente MQTT

O código MQTT mistura três preocupações que falham de maneiras diferentes:

1. Lógica de negócio: fazer o parse de payloads, decidir o que publicar, mapear topics para dispositivos.
2. Comportamento do protocolo: QoS, retained messages, reconexões, novas assinaturas, last will.
3. Ambiente: certificados TLS, credenciais, portas, firewalls, cotas do broker.

A maioria das suítes de teste MQTT instáveis tenta cobrir as três em um único teste que se conecta a um broker e espera com sleep. Separá-las permite que cada camada use a ferramenta mais barata capaz de pegar seus bugs.

Os exemplos usam Python com paho-mqtt 2.x e pytest, mas a estrutura vale para qualquer linguagem.

Passo 1: projete código testável

A mudança mais útil de todas é manter o cliente MQTT na borda. Coloque as decisões em funções ou classes simples que recebem uma função de publicação como dependência, e deixe um adaptador fino ligá-las ao paho.

thermostat.py:

import json
from dataclasses import dataclass
from typing import Callable

Publish = Callable[[str, str, int, bool], None]


@dataclass
class Thermostat:
    prefix: str
    publish: Publish
    setpoint: float = 21.0
    hysteresis: float = 0.5
    heating: bool = False

    def on_temperature(self, topic: str, payload: bytes) -> None:
        try:
            value = float(json.loads(payload)["value"])
        except (ValueError, KeyError, TypeError, json.JSONDecodeError):
            return  # ignora mensagens malformadas

        room = topic.split("/")[-2]
        if not self.heating and value < self.setpoint - self.hysteresis:
            self.heating = True
        elif self.heating and value > self.setpoint + self.hysteresis:
            self.heating = False
        else:
            return

        self.publish(f"{self.prefix}/{room}/heater/set",
                     "ON" if self.heating else "OFF", 1, True)

adapter.py:

import threading

import paho.mqtt.client as mqtt
from thermostat import Thermostat


def build_client(host, port, client_id, prefix, username=None, password=None, tls=False):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if username:
        client.username_pw_set(username, password)
    if tls:
        client.tls_set()

    thermostat = Thermostat(
        prefix=prefix,
        publish=lambda t, p, q, r: client.publish(t, p, qos=q, retain=r),
    )

    def on_connect(client, userdata, flags, reason_code, properties):
        if not reason_code.is_failure:
            client.subscribe(f"{prefix}/+/temperature", qos=1)

    ready = threading.Event()

    def on_subscribe(client, userdata, mid, reason_codes, properties):
        ready.set()

    def on_message(client, userdata, msg):
        thermostat.on_temperature(msg.topic, msg.payload)

    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    return client, thermostat, ready

Assinar dentro de on_connect é proposital: isso restaura as assinaturas após cada reconexão, e o evento ready retornado permite que os testes esperem até a assinatura estar ativa em vez de chutar com sleep(). É exatamente esse tipo de ligação que os testes de integração vão verificar.

Passo 2: testes unitários com um cliente simulado

Como Thermostat recebe publish como parâmetro, um Mock substitui a rede por completo.

test_thermostat.py:

import json
from unittest.mock import Mock

from thermostat import Thermostat

PREFIX = "home/test"


def make():
    pub = Mock()
    return Thermostat(prefix=PREFIX, publish=pub), pub


def msg(value):
    return json.dumps({"value": value}).encode()


def test_turns_heater_on_below_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    pub.assert_called_once_with(f"{PREFIX}/kitchen/heater/set", "ON", 1, True)


def test_no_publish_inside_hysteresis_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.2))
    pub.assert_not_called()


def test_turns_heater_off_above_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.6))
    assert pub.call_args.args[1] == "OFF"


def test_ignores_garbage():
    t, pub = make()
    for bad in [b"", b"not json", b'{"temp": 3}', b'{"value": "hot"}']:
        t.on_temperature(f"{PREFIX}/kitchen/temperature", bad)
    pub.assert_not_called()

Esses testes rodam em milissegundos e cobrem a lógica que mais costuma ter bugs. Resista à tentação de simular o próprio paho em detalhes (verificando que client.subscribe foi chamado com argumentos exatos): esses testes espelham a implementação, quebram em refatorações e ainda assim não provam que o broker aceita o que você envia. Esse é o trabalho da camada de integração.

Passo 3: testes de integração contra um broker em contêiner

Um broker real em um contêiner oferece comportamento de protocolo honesto, sem estado compartilhado e sem depender da internet. A imagem oficial do Eclipse Mosquitto inclui um arquivo de configuração que permite conexões anônimas, ideal para testes:

docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf

Agora escreva testes que exercitam o adaptador real. Uma função auxiliar coleta mensagens em um segundo cliente e espera por uma condição em vez de dormir por um tempo fixo.

test_integration.py:

import os
import queue
import uuid

import paho.mqtt.client as mqtt
import pytest

from adapter import build_client

HOST = os.getenv("MQTT_TEST_HOST", "localhost")
PORT = int(os.getenv("MQTT_TEST_PORT", "1883"))


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


@pytest.fixture
def probe(prefix):
    received = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: received.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/#", qos=1)
    c.loop_start()
    yield c, received
    c.loop_stop()
    c.disconnect()


def wait_for(q, predicate, timeout=5.0):
    while True:
        topic, payload, retain = q.get(timeout=timeout)  # lança queue.Empty no timeout
        if predicate(topic, payload):
            return topic, payload, retain


def test_temperature_drives_heater(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/office/temperature", '{"value": 18.5}', qos=1)
        topic, payload, _ = wait_for(received, lambda t, p: t.endswith("/heater/set"))
        assert topic == f"{prefix}/office/heater/set"
        assert payload == "ON"
    finally:
        client.loop_stop()
        client.disconnect()


def test_heater_command_is_retained(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/hall/temperature", '{"value": 15}', qos=1)
        wait_for(received, lambda t, p: t.endswith("/heater/set"))
    finally:
        client.loop_stop()
        client.disconnect()

    # Um assinante tardio deve receber o comando retido imediatamente
    late = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"late-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: late.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/hall/heater/set", qos=1)
    c.loop_start()
    try:
        _, payload, retain = late.get(timeout=5)
        assert payload == "ON" and retain is True
    finally:
        c.loop_stop()
        c.disconnect()

Três hábitos mantêm os testes de integração estáveis:

• Prefixo de topic único por teste. Retained messages de um teste não vazam para o próximo.
• Espere por condições, não por sleep(). Uma fila com timeout falha rápido e nunca espera mais do que o necessário.
• Sempre pare os loops e desconecte no finally. Threads de rede esquecidas fazem os testes seguintes travarem.

Para testar a lógica de reconexão, pare e reinicie o contêiner no meio do teste (docker restart test-broker) e verifique se o cliente refaz as assinaturas e recebe uma mensagem publicada após o reinício.

Rodando o broker no CI

O GitHub Actions pode iniciar o contêiner como uma etapa antes dos testes:

name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Start MQTT broker
        run: |
          docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
            mosquitto -c /mosquitto-no-auth.conf
      - run: pip install "paho-mqtt>=2,<3" pytest
      - run: pytest -q

Testes unitários e de integração podem compartilhar uma mesma execução do pytest; marque os lentos com @pytest.mark.integration se quiser pulá-los localmente.

Passo 4: smoke tests end-to-end contra um broker real

Um contêiner não consegue dizer se as suas configurações de TLS, o bundle de CAs, as regras de firewall ou as credenciais de produção funcionam. Um smoke test curto contra o endpoint real consegue. Para projetos open source ou verificações rápidas, o servidor MQTT público em broker.connectmq.com é suficiente; ele aceita clientes anônimos via TLS na porta 8883. Mantenha os payloads sintéticos, porque a área pública pode ser lida por qualquer pessoa.

smoke_test.py:

import os
import sys
import threading
import uuid

import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
USER = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"smoke-{uuid.uuid4().hex[:8]}"
TOPIC = f"{os.getenv('MQTT_PREFIX', 'connectmq/demo')}/smoke-{uuid.uuid4().hex[:8]}/ping"

got = threading.Event()
nonce = uuid.uuid4().hex
errors = []


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        errors.append(f"connect failed: {reason_code}")
        got.set()  # acorda a thread principal; callbacks não conseguem encerrar o processo
        return
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    client.publish(TOPIC, nonce, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == nonce:
        got.set()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USER:
    client.username_pw_set(USER, PASSWORD)
client.tls_set()
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 8883, keepalive=30)
client.loop_start()
ok = got.wait(timeout=15) and not errors
for e in errors:
    print(e)
client.loop_stop()
client.disconnect()
print("smoke test passed" if ok else "smoke test FAILED: no echo within 15 s")
sys.exit(0 if ok else 1)

Para produção, aponte o mesmo script para o seu broker privado definindo MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID e MQTT_PREFIX a partir dos secrets do CI. Em um broker MQTT gerenciado como o ConnectMQ, os client IDs ficam vinculados às credenciais emitidas, então crie um conjunto de credenciais dedicado para o CI em vez de pegar emprestada a identidade de um dispositivo; caso contrário, o smoke test vai desconectar esse dispositivo.

Rode os smoke tests antes dos releases e de forma agendada (por exemplo, a cada hora), não a cada commit. Brokers compartilhados têm cotas: o servidor MQTT público permite 10 mensagens por segundo por cliente e 10 assinaturas por cliente, e o plano gratuito ConnectMQ Serverless limita as conexões simultâneas a 25.

Antipadrões comuns

Alguns hábitos deixam as suítes de teste MQTT lentas e instáveis. Fique atento a eles na revisão de código:

• Sleeps fixos. Um time.sleep(2) depois de publicar é longo demais, desperdiçando minutos ao longo da suíte, ou curto demais em um runner de CI sobrecarregado. Espere em uma fila ou evento com timeout.
• Topics compartilhados entre testes. Retained messages e assinaturas remanescentes de um teste vazam para o próximo. Gere um prefixo único por teste.
• Erros engolidos nos callbacks. Exceções dentro de callbacks MQTT rodam na thread de rede, então não fazem o teste falhar diretamente. Capture-as em uma lista e verifique que ela está vazia no teardown.
• Credenciais de dispositivos no CI. Pegar emprestada a identidade de um dispositivo real para testes desconecta esse dispositivo. Crie credenciais separadas para a automação.
• Testar a biblioteca em vez do seu código. Você não precisa provar que o paho entrega mensagens QoS 1; precisa provar que o seu código assina os topics certos e reage corretamente.
• Um único teste end-to-end gigante. Quando ele falha, ninguém sabe qual camada quebrou. Mantenha os testes end-to-end poucos e focados, e leve os detalhes para os testes unitários.

Checklist de testes

1. A lógica de negócio fica fora dos callbacks MQTT e é testada unitariamente com um publish simulado.
2. Payloads malformados, vazios e grandes demais são testados e ignorados com segurança.
3. Os testes de integração usam um contêiner de broker novo e um prefixo de topic único por teste.
4. O comportamento de reconexão e nova assinatura é testado reiniciando o broker.
5. O comportamento de retained messages e de last will é verificado com um assinante tardio.
6. Um smoke test TLS com credenciais reais roda antes de cada release.
7. As credenciais do CI são separadas das credenciais dos dispositivos.

Para detalhes específicos de cada linguagem, veja o tutorial de cliente MQTT em PHP, o tutorial do cliente Paho MQTT C e o guia de bibliotecas cliente MQTT, mais abrangente.

Comece a usar o ConnectMQ

Dê ao seu pipeline de testes um broker real com quem conversar. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, adicione um conjunto de credenciais dedicado ao CI e rode o smoke test contra ele a cada release. O explorador de topics ao vivo e o cliente no navegador do portal facilitam ver exatamente o que os seus testes publicaram. O ConnectMQ opera o broker; você cuida dos testes.

Perguntas frequentes

Devo simular a biblioteca MQTT nos testes unitários?

Simule na sua própria fronteira, como uma função publish que você injeta, em vez de simular os detalhes internos da biblioteca. Verifique o comportamento real do protocolo com um teste de integração contra um broker em contêiner.

Qual broker devo usar nos testes de integração MQTT no CI?

Um contêiner descartável do Eclipse Mosquitto é a opção mais simples: ele sobe em cerca de um segundo, não precisa de credenciais com /mosquitto-no-auth.conf e não deixa estado algum entre as execuções.

Posso usar o broker.connectmq.com em testes automatizados?

Sim, para smoke tests ocasionais com dados sintéticos e topics únicos, dentro dos limites de 10 mensagens por segundo e 10 assinaturas por cliente. Para suítes de teste pesadas, use um contêiner local ou o seu próprio broker privado.

Como testar a lógica de reconexão MQTT?

Reinicie o contêiner do broker durante o teste e verifique se o cliente reconecta, refaz as assinaturas e recebe uma mensagem publicada após o reinício.