Pruebas hardware-in-the-loop con MQTT

Los simuladores y los brokers mock detectan errores de lógica, pero nunca reproducen una caída de tensión durante un handshake TLS ni una pila Wi-Fi que se rinde tras la tercera reconexión. Las pruebas hardware-in-the-loop (HIL) con MQTT cierran esa brecha: placas reales con firmware real se conectan a un broker de pruebas, y un script en un PC las controla, escucha cada mensaje y decide si la prueba pasa o falla.

Esta guía muestra cómo montar un pequeño banco HIL, automatizarlo con pytest y paho-mqtt 2.x, inyectar fallos como cortes de energía y convertir los resultados en informes que tu CI pueda mostrar.

Cómo es un banco HIL con MQTT

Un banco no tiene por qué ser caro. Una estantería con unas cuantas placas, un hub USB y un pequeño equipo Linux bastan para empezar.

El runner es a la vez director y observador: enciende el dispositivo, se suscribe a todo lo que publica, envía comandos como lo haría un backend y comprueba tiempos y contenido.

Elige el broker del banco

El broker determina lo realista que es tu banco. Un broker local en el runner saca internet de la ecuación, lo cual es bueno para probar la lógica del protocolo de forma determinista. Un broker remoto con tu configuración de producción prueba además DNS, certificados TLS, autenticación y la ruta de red real, que es de donde suelen venir los fallos en campo. Muchos equipos usan ambos: un broker local para feedback rápido del firmware y uno remoto para las ejecuciones nocturnas.

Elijas lo que elijas, conéctate una vez a mano con el cliente MQTT online y suscríbete a los topics del dispositivo antes de automatizar nada. Si no ves los mensajes del dispositivo manualmente, ningún script los verá.

Aísla cada ejecución

Un banco se comparte a lo largo del tiempo: la ejecución de ayer dejó mensajes retenidos, el banco de un compañero publica en topics parecidos y dos jobs de CI pueden correr a la vez. Da a cada banco y a cada ejecución su propia raíz de topics, por ejemplo testmqtt/hil/<bench>/<run-id>/..., y haz que el prefijo de topics del firmware sea configurable (por consola serie, un fichero de aprovisionamiento o un flag de compilación) en lugar de dejarlo fijo en el código. El patrón completo se describe en nombrar topics de prueba para que nunca colisionen.

Si el prefijo no puede cambiar en tiempo de ejecución, usa al menos un prefijo fijo por banco y borra los mensajes retenidos bajo él antes de cada prueba.

El arnés: un grabador MQTT

El corazón del banco es una pequeña clase que se suscribe al prefijo del dispositivo, guarda una transcripción de cada mensaje y permite a las pruebas esperar un mensaje concreto con timeout. Esperar eventos en lugar de usar sleep() es, con diferencia, la mejor solución contra las pruebas de hardware inestables.

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()

Control de alimentación y puerto serie

Cada prueba debería empezar desde un arranque en frío, así que el banco necesita poder cortar la alimentación. Para placas alimentadas por USB en un hub que permita conmutar cada puerto, basta con uhubctl; para dispositivos conectados a la red eléctrica, una placa de relés o una PDU conmutable hacen el mismo trabajo. Los logs serie se capturan en un hilo en segundo plano para que una prueba fallida pueda adjuntar exactamente lo que imprimió el firmware.

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):
    # Sustitúyelo por la llamada a tu relé o PDU. Ejemplo: puerto 2 del 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()

El firmware de este ejemplo publica online en <prefix>/status tras conectarse, configura un Last Will offline en el mismo topic, escucha en <prefix>/cmd/# y responde en <prefix>/ack.

Escribe las pruebas

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 -> el broker se rinde tras 1,5 x 30 s
    rig.mqtt.expect("status", timeout=60, match=lambda p: p == "offline")

La última prueba es algo que ningún simulador demuestra de verdad: un corte de energía brusco, un broker que solo se entera al expirar el keep alive y un Last Will que llega al backend. Consulta probar el Last Will para ver en detalle cuándo lo envía el broker.

Inyecta fallos

El valor del HIL crece con cada modo de fallo que puedas provocar a voluntad:

• Corte de energía a mitad de una publicación: corta la alimentación justo después de un comando y comprueba que el dispositivo se recupera y no informa de un estado obsoleto.
• Reinicio del broker: reinicia el broker local y verifica que el dispositivo se reconecta con backoff, se vuelve a suscribir y sigue publicando. Los escenarios se explican en pruebas de reconexión MQTT.
• Pérdida de red sin pérdida de energía: apaga el punto de acceso Wi-Fi o el puerto del switch. El dispositivo sigue funcionando, su conexión TCP muere en silencio y descubres si tu lógica de keep alive y reconexión funciona de verdad.
• Credenciales erróneas o una ACL revocada: cambia la contraseña del dispositivo en el broker de pruebas y confirma que el firmware registra el motivo del CONNACK en lugar de entrar en un bucle silencioso. La consulta de reason codes ayuda a descifrar lo que informa el dispositivo.

Informes

Ejecuta la suite con pytest --junitxml=report.xml para que cualquier sistema de CI muestre el resultado de cada prueba. Cuando algo falle, adjunta el log serie y la transcripción MQTT del grabador: juntos te dicen si el dispositivo nunca envió el mensaje, lo envió al topic equivocado o el broker lo rechazó. Un runner de CI self-hosted conectado al banco permite que cada merge de firmware lance una ejecución en hardware; la parte del workflow es la misma que en pruebas MQTT en GitHub Actions.

Un buen informe HIL contiene:

1. Versión de firmware y commit de git del dispositivo bajo prueba.
2. Nombre del banco, endpoint del broker e ID de ejecución (el prefijo de topics).
3. Resultado y duración de cada prueba, además de los tiempos medidos.
4. Log serie y transcripción MQTT de cada prueba fallida.

Mantén el banco fiable

• Vuelve a flashear el dispositivo al inicio de cada ejecución para que la prueba corra siempre contra la build que crees.
• Borra los mensajes retenidos bajo el prefijo del banco antes de la ejecución, o un online antiguo puede hacer que un arranque roto parezca correcto.
• Conserva una imagen de firmware que sepas que funciona; si empieza a fallar, el problema está en el banco, no en el código.
• Pon en cuarentena las pruebas inestables en lugar de reintentarlas a ciegas, y corrige la causa: normalmente una espera que falta o un topic compartido.

Observa el tráfico de tu banco en el navegador

Las pruebas hardware-in-the-loop convierten «en mi mesa funcionaba» en una comprobación repetible que se ejecuta en cada build. Empieza con una placa, un interruptor de alimentación y tres pruebas (arranque, ida y vuelta de un comando y corte de energía) y amplía el catálogo de fallos a medida que el campo te enseñe otros nuevos.

Preguntas frecuentes