Testes hardware-in-the-loop com MQTT

Simuladores e brokers mock pegam bugs de lógica, mas nunca reproduzem uma queda de tensão durante um handshake TLS nem uma pilha Wi-Fi que desiste depois da terceira reconexão. Os testes hardware-in-the-loop (HIL) com MQTT fecham essa lacuna: placas reais com firmware real se conectam a um broker de teste, e um script em um PC as controla, escuta cada mensagem e decide se o teste passou ou falhou.

Este guia mostra como montar uma pequena bancada HIL, automatizá-la com pytest e paho-mqtt 2.x, injetar falhas como queda de energia e transformar os resultados em relatórios que o seu CI consiga exibir.

Como é uma bancada HIL com MQTT

Uma bancada não precisa ser cara. Uma prateleira com algumas placas, um hub USB e uma pequena máquina Linux bastam para começar.

O runner é ao mesmo tempo maestro e observador: liga o dispositivo, assina tudo o que ele publica, envia comandos como um backend faria e verifica tempos e conteúdo.

Escolha o broker da bancada

O broker define o quão realista é a sua bancada. Um broker local no runner tira a internet da equação, o que é bom para testar a lógica do protocolo de forma determinística. Um broker remoto com as configurações de produção também testa DNS, certificados TLS, autenticação e o caminho de rede real, que é de onde costumam vir as falhas em campo. Muitas equipes usam os dois: um broker local para feedback rápido do firmware e um remoto para as execuções noturnas.

Seja qual for a escolha, conecte-se uma vez manualmente com o cliente MQTT online e assine os topics do dispositivo antes de automatizar qualquer coisa. Se você não vê as mensagens do dispositivo manualmente, nenhum script vai ver.

Isole cada execução

Uma bancada é compartilhada ao longo do tempo: a execução de ontem deixou mensagens retidas, a bancada de um colega publica em topics parecidos e dois jobs de CI podem rodar ao mesmo tempo. Dê a cada bancada e a cada execução a sua própria raiz de topics, por exemplo testmqtt/hil/<bench>/<run-id>/..., e deixe o prefixo de topics do firmware configurável (pelo console serial, por um arquivo de provisionamento ou por uma flag de build) em vez de fixá-lo no código. O padrão completo está em como nomear topics de teste para que nunca colidam.

Se o prefixo não puder mudar em tempo de execução, use pelo menos um prefixo fixo por bancada e limpe as mensagens retidas sob ele antes de cada teste.

O harness: um gravador MQTT

O coração da bancada é uma pequena classe que assina o prefixo do dispositivo, mantém uma transcrição de cada mensagem e permite que os testes esperem por uma mensagem específica com timeout. Esperar por eventos em vez de usar sleep() é, de longe, a maior correção para testes de hardware instáveis.

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

class Recorder:
    def __init__(self, host, port, prefix):
        self.prefix = prefix
        self.inbox = queue.Queue()
        self.transcript = []
        self.ready = threading.Event()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"hil-runner-{uuid.uuid4().hex[:8]}")
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, codes, props: self.ready.set()
        self.client.on_message = self._on_message
        self.client.connect(host, port, keepalive=30)
        self.client.loop_start()
        if not self.ready.wait(10):
            raise RuntimeError("runner could not subscribe to the test broker")

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.subscribe(f"{self.prefix}/#", qos=1)

    def _on_message(self, client, userdata, msg):
        entry = (time.time(), msg.topic, msg.payload.decode(errors="replace"), msg.retain)
        self.transcript.append(entry)
        self.inbox.put(entry)

    def expect(self, suffix, timeout=10, match=lambda payload: True):
        want, deadline = f"{self.prefix}/{suffix}", time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                _, topic, payload, _ = self.inbox.get(timeout=left)
            except queue.Empty:
                break
            if topic == want and match(payload):
                return payload
        raise AssertionError(f"no matching message on {want} within {timeout}s")

    def publish(self, suffix, payload, qos=1):
        self.client.publish(f"{self.prefix}/{suffix}", payload, qos=qos).wait_for_publish(5)

    def drain(self):
        while not self.inbox.empty():
            self.inbox.get_nowait()

    def close(self):
        self.client.loop_stop()
        self.client.disconnect()

Controle de energia e serial

Todo teste deve começar de um boot a frio, então a bancada precisa de um jeito de cortar a energia. Para placas alimentadas por USB em um hub que permite chavear cada porta, o uhubctl basta; para dispositivos ligados na tomada, uma placa de relés ou uma PDU chaveável fazem o mesmo trabalho. Os logs seriais são capturados em uma thread em segundo plano para que um teste que falhou possa anexar exatamente o que o firmware imprimiu.

import os, subprocess, threading, time
import pytest, serial
from hil.recorder import Recorder

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
BENCH = os.getenv("HIL_BENCH", "bench1")
RUN_ID = os.getenv("GITHUB_RUN_ID", time.strftime("%Y%m%dT%H%M%S"))

def set_power(on: bool):
    # Troque pela chamada do seu relé ou PDU. Exemplo: porta 2 do hub 1-1.
    subprocess.run(["uhubctl", "-l", "1-1", "-p", "2", "-a", "on" if on else "off"], check=True)

class Rig:
    def __init__(self, mqtt):
        self.mqtt = mqtt
    def power_off(self):
        set_power(False)
    def power_cycle(self, off_seconds=2):
        set_power(False); time.sleep(off_seconds)
        self.mqtt.drain(); set_power(True)

def capture(port, log, stop):
    while not stop.is_set():
        log.write(port.read(512))

@pytest.fixture
def rig(tmp_path):
    prefix = f"testmqtt/hil/{BENCH}/{RUN_ID}/dut"
    rec = Recorder(HOST, 1883, prefix)
    log = open(tmp_path / "serial.log", "wb")
    port = serial.Serial(os.getenv("HIL_SERIAL", "/dev/ttyUSB0"), 115200, timeout=0.2)
    stop = threading.Event()
    threading.Thread(target=capture, args=(port, log, stop), daemon=True).start()
    yield Rig(rec)
    stop.set(); time.sleep(0.3); port.close(); log.close()
    rec.close()

O firmware deste exemplo publica online em <prefix>/status depois de conectar, define um Last Will offline no mesmo topic, escuta em <prefix>/cmd/# e responde em <prefix>/ack.

Escreva os testes

import json

def test_boot_reports_online(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")

def test_command_round_trip(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")
    rig.mqtt.publish("cmd/led", json.dumps({"id": "t-1", "state": "on"}))
    ack = json.loads(rig.mqtt.expect("ack", timeout=5))
    assert ack["id"] == "t-1" and ack["ok"] is True

def test_power_loss_publishes_last_will(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")
    rig.power_off()
    # keep alive 30 s -> o broker desiste após 1,5 x 30 s
    rig.mqtt.expect("status", timeout=60, match=lambda p: p == "offline")

O último teste prova algo que nenhum simulador prova de verdade: um corte abrupto de energia, um broker que só percebe quando o keep alive expira e um Last Will que chega ao backend. Veja como testar o Last Will para entender exatamente quando o broker o envia.

Injete falhas

O valor do HIL cresce a cada modo de falha que você consegue provocar sob demanda:

• Queda de energia no meio de um publish: corte a energia logo após um comando e verifique se o dispositivo se recupera e não reporta um estado desatualizado.
• Reinício do broker: reinicie o broker local e confira se o dispositivo reconecta com backoff, refaz as assinaturas e volta a publicar. Os cenários estão em testes de reconexão MQTT.
• Perda de rede sem perda de energia: desligue o access point Wi-Fi ou a porta do switch. O dispositivo continua rodando, a conexão TCP morre em silêncio e você descobre se a sua lógica de keep alive e reconexão funciona de fato.
• Credenciais erradas ou ACL revogada: troque a senha do dispositivo no broker de teste e confirme que o firmware registra o motivo do CONNACK em vez de ficar em loop silencioso. A consulta de reason codes ajuda a decifrar o que o dispositivo reporta.

Relatórios

Rode a suíte com pytest --junitxml=report.xml para que qualquer sistema de CI mostre o resultado de cada teste. Em caso de falha, anexe o log serial e a transcrição MQTT do gravador: juntos eles mostram se o dispositivo nunca enviou a mensagem, enviou para o topic errado ou se o broker a recusou. Um runner de CI self-hosted ligado à bancada permite que cada merge de firmware dispare uma execução no hardware; o lado do workflow é o mesmo de testes MQTT no GitHub Actions.

Um bom relatório HIL contém:

1. Versão do firmware e commit git do dispositivo sob teste.
2. Nome da bancada, endpoint do broker e ID da execução (o prefixo de topics).
3. Resultado e duração de cada teste, além dos tempos medidos.
4. Log serial e transcrição MQTT de cada teste que falhou.

Mantenha a bancada confiável

• Regrave o firmware no início de cada execução para que o teste rode sempre contra a build que você imagina.
• Limpe as mensagens retidas sob o prefixo da bancada antes da execução, senão um online antigo pode fazer um boot quebrado parecer normal.
• Mantenha uma imagem de firmware sabidamente boa; se ela começar a falhar, o problema é a bancada, não o código.
• Coloque testes instáveis em quarentena em vez de repeti-los às cegas e corrija a causa, geralmente uma espera que faltou ou um topic compartilhado.

Acompanhe o tráfego da bancada no navegador

Testes hardware-in-the-loop transformam "funcionou na minha mesa" em uma verificação repetível que roda a cada build. Comece com uma placa, um interruptor de energia e três testes (boot, ida e volta de um comando e queda de energia) e amplie o catálogo de falhas conforme o campo for mostrando novas.

Perguntas frequentes