Gravar e reproduzir tráfego MQTT para testes

Gravar e reproduzir tráfego MQTT significa capturar mensagens reais de um broker em um arquivo — topic, payload, QoS, flag retain e o momento em que cada uma chegou — e depois publicar esse arquivo de volta com a mesma temporização. Você ganha testes de regressão construídos a partir do que seus dispositivos realmente enviam, e não do que alguém supôs que eles enviam.

Este guia monta um pequeno gravador e um reprodutor com temporização precisa em Python (paho-mqtt 2.x) e depois mostra como transformar uma gravação em um teste de regressão automatizado.

Por que a reprodução vence dados sintéticos

Dados de teste gerados são limpos. Frotas reais não: versões de firmware diferentes mandam JSON ligeiramente diferente, sensores disparam dez leituras em 50 ms após uma reconexão, um gateway republica o mesmo status retido a cada minuto. Os bugs moram nesses detalhes.

Dados sintéticos continuam úteis — use um gerador de dados de teste para casos extremos que ninguém gravou ainda — mas uma gravação é o mais perto de produção que você consegue rodar no notebook.

Escolha o que gravar

Um gravador é um cliente MQTT comum com uma assinatura. Escolha um filtro de topic que cubra o fluxo que você quer testar, e não # no broker inteiro: capturas menores são mais fáceis de revisar e vazam menos dados. Confira o filtro antes de começar com o topic matcher.

Decida com antecedência:

• Filtro de topic — por exemplo testmqtt/replay/site1/# para a telemetria de um local.
• QoS da assinatura — assine com QoS 2. O QoS de entrega é o menor entre o QoS de publicação e o da assinatura, então uma assinatura QoS 2 preserva o nível original.
• Codificação do payload — armazene payloads em base64 para que formatos binários (Protobuf, CBOR) sobrevivam à ida e volta.
• Privacidade — payloads reais podem conter localizações, nomes ou tokens. Limpe-os antes de versionar uma captura em um repositório.

Um gravador em 30 linhas

O gravador escreve um objeto JSON por linha (JSONL). O campo t é o número de segundos desde que o gravador iniciou, medido com um relógio monotônico para que ajustes no relógio do sistema não distorçam a temporização.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = sys.argv[1] if len(sys.argv) > 1 else "testmqtt/replay/#"
out = open(sys.argv[2] if len(sys.argv) > 2 else "capture.jsonl", "w", buffering=1)
start = time.monotonic()

def on_connect(client, userdata, flags, reason_code, properties):
    print("conectado:", reason_code)
    client.subscribe(FILTER, qos=2)        # assina de novo após cada reconexão

def on_message(client, userdata, msg):
    out.write(json.dumps({
        "t": round(time.monotonic() - start, 6),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": msg.retain,
        "payload": base64.b64encode(msg.payload).decode(),
    }) + "\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-recorder-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, PORT, keepalive=30)
try:
    client.loop_forever()
except KeyboardInterrupt:
    client.disconnect()
    out.close()

Rode com python record.py 'testmqtt/replay/#' capture.jsonl, gere algum tráfego e pressione Ctrl+C. O buffering por linha (buffering=1) garante que o arquivo seja utilizável mesmo se o processo for encerrado à força.

Mensagens retidas em uma captura

Quando o gravador assina, o broker envia imediatamente todas as mensagens retidas correspondentes com retain: true. Elas são um retrato do estado no momento da gravação, não tráfego ao vivo, e chegam todas em t ≈ 0. Mensagens encaminhadas ao vivo chegam com retain: false, mesmo que o publicador tenha ligado o flag (clientes MQTT 5 podem mudar isso com a opção de assinatura Retain As Published). O reprodutor abaixo ignora esse retrato por padrão; publique-o separadamente se o seu teste precisar de estado inicial.

Um reprodutor que respeita a temporização

A abordagem ingênua — sleep(gap) entre mensagens — acumula desvio, porque cada chamada de publish também leva tempo. Em vez disso, calcule o horário previsto de cada mensagem a partir do início da reprodução e durma só até lá. Um fator de velocidade permite comprimir uma hora de tráfego em minutos.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

FILE = sys.argv[1]
SPEED = float(sys.argv[2]) if len(sys.argv) > 2 else 1.0     # 2.0 = duas vezes mais rápido
SRC, DST = "testmqtt/replay/", "testmqtt/replay-run/"          # reescrita de topic

records = [json.loads(line) for line in open(FILE) if line.strip()]
records = [r for r in records if not r["retain"]]              # ignora o retrato retido
if not records:
    sys.exit("nada para reproduzir")
offset = records[0]["t"]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-replayer-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.connect("broker.connectmq.com", 1883)
client.loop_start()

t0 = time.monotonic()
late = 0.0
for r in records:
    due = t0 + (r["t"] - offset) / SPEED
    delay = due - time.monotonic()
    if delay > 0:
        time.sleep(delay)
    else:
        late = max(late, -delay)
    topic = r["topic"].replace(SRC, DST, 1)
    info = client.publish(topic, base64.b64decode(r["payload"]), qos=r["qos"])

info.wait_for_publish(timeout=30)
print(f"{len(records)} mensagens reproduzidas, pior atraso {late * 1000:.1f} ms")
client.loop_stop()
client.disconnect()

A reescrita de topic é a rede de proteção: a gravação veio de testmqtt/replay/... e a reprodução vai para testmqtt/replay-run/..., então nada que escute os topics originais reage a ela. Aponte o sistema sob teste para o prefixo reescrito.

A linha de pior atraso indica se a reprodução foi fiel. Se ela cresce com fatores de velocidade altos, o próprio reprodutor é o gargalo: a janela inflight do cliente está cheia de mensagens QoS 1/2 sem confirmação. Veja o ajuste de desempenho do cliente para aumentá-la.

Transforme uma reprodução em teste de regressão

Uma reprodução vira teste quando você compara o que o sistema sob teste produziu com um resultado sabidamente correto. O padrão:

1. Grave uma vez as entradas (telemetria dos dispositivos) e versione a captura, já limpa, como fixture.
2. Rode o gravador nos topics de saída do sistema: alertas, agregados, comandos.
3. Reproduza as entradas.
4. Compare a captura de saída com um arquivo golden gravado a partir de um build confiável.

A temporização nunca é idêntica entre execuções, então compare conteúdo, não timestamps. Agrupe por topic e compare a sequência de payloads de cada topic; o MQTT só garante ordem por topic e por publicador, e uma comparação de ordem global gera testes instáveis.

import base64, json, sys
from collections import defaultdict

def load(path):
    per_topic = defaultdict(list)
    for line in open(path):
        if line.strip():
            r = json.loads(line)
            per_topic[r["topic"]].append(base64.b64decode(r["payload"]))
    return per_topic

golden, actual = load(sys.argv[1]), load(sys.argv[2])
ok = True
for topic in sorted(set(golden) | set(actual)):
    if golden[topic] != actual[topic]:
        ok = False
        print(f"DIFF {topic}: esperadas {len(golden[topic])} msgs, recebidas {len(actual[topic])}")
print("PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)

Se os payloads contêm timestamps ou IDs gerados, normalize-os antes de comparar (faça o parse do JSON e remova esses campos). Integre os três scripts em uma fixture do pytest ou em um job de CI — o guia de testes de integração e testes MQTT no GitHub Actions mostram como subir um broker a cada execução.

Armadilhas para observar

• Reproduzir comandos. Uma captura de topics .../cmd/... vai acionar relés reais se chegar a um dispositivo real. Filtre os topics de comando das capturas de entrada, a menos que o teste seja sobre eles.
• Mensagens duplicadas. QoS 1 pode entregar uma mensagem duas vezes após uma reconexão, e a gravação mantém a duplicata. Isso é realista — seus consumidores devem lidar com isso — mas saiba que ela existe antes de contar mensagens.
• Metadados do MQTT 5 ausentes. Os scripts acima armazenam topic, payload, QoS e retain. Se o seu sistema depende de user properties, content type ou response topics, adicione campos de msg.properties ao registro e passe um objeto Properties ao publicar.
• Consumidores dependentes do relógio. Lógica que usa o horário real (por exemplo "nenhuma leitura em 60 s") se comporta de outro jeito a 10×. Reproduza esses testes a 1×.
• Limites do broker. Reproduções rápidas de capturas grandes podem esbarrar em limites de taxa de mensagens ou de filas em brokers compartilhados. Use sua própria instância para qualquer coisa além de uma verificação rápida.

Confira a reprodução a olho nu

Antes de automatizar qualquer coisa, acompanhe uma reprodução chegando. Abra o cliente MQTT online, conecte-se a broker.connectmq.com por WebSockets seguros e assine o prefixo reescrito; depois inicie o replay.py. Você deve ver as mensagens chegarem em rajadas que seguem o padrão original.

Veja a reprodução no navegador

Um gravador, um reprodutor e um script de comparação somam menos de cem linhas de Python e transformam tráfego real em testes repetíveis. Grave uma vez em campo, limpe a captura, guarde-a junto ao seu código, e toda mudança futura será testada contra o que seus dispositivos realmente enviam.

Perguntas frequentes