Tests hardware-in-the-loop avec MQTT

Les simulateurs et les brokers mock attrapent les bugs de logique, mais ils ne reproduisent jamais une chute de tension pendant un handshake TLS ni une pile Wi-Fi qui abandonne après la troisième reconnexion. Les tests hardware-in-the-loop (HIL) avec MQTT comblent ce manque : de vraies cartes avec leur vrai firmware se connectent à un broker de test, et un script sur un PC les pilote, écoute chaque message et décide si le test réussit ou échoue.

Ce guide montre comment monter un petit banc HIL, l'automatiser avec pytest et paho-mqtt 2.x, injecter des pannes comme une coupure d'alimentation et transformer les résultats en rapports que votre CI sait afficher.

À quoi ressemble un banc HIL MQTT

Un banc n'a pas besoin d'être coûteux. Une étagère avec quelques cartes, un hub USB et une petite machine Linux suffisent pour commencer.

Le runner est à la fois chef d'orchestre et observateur : il allume l'appareil, s'abonne à tout ce qu'il publie, envoie des commandes comme le ferait un backend et vérifie le timing et le contenu.

Choisir le broker du banc

Le broker détermine le réalisme de votre banc. Un broker local sur le runner retire internet de l'équation, ce qui est idéal pour tester la logique du protocole de façon déterministe. Un broker distant avec vos réglages de production teste en plus le DNS, les certificats TLS, l'authentification et le vrai chemin réseau, d'où viennent généralement les pannes sur le terrain. Beaucoup d'équipes utilisent les deux : un broker local pour un retour rapide sur le firmware et un broker distant pour les exécutions nocturnes.

Quel que soit votre choix, connectez-vous une fois à la main avec le client MQTT en ligne et abonnez-vous aux topics de l'appareil avant d'automatiser quoi que ce soit. Si vous ne voyez pas les messages de l'appareil manuellement, aucun script ne les verra.

Isoler chaque exécution

Un banc est partagé dans le temps : l'exécution d'hier a laissé des messages retenus, le banc d'un collègue publie sur des topics similaires et deux jobs de CI peuvent tourner en même temps. Donnez à chaque banc et à chaque exécution sa propre racine de topics, par exemple testmqtt/hil/<bench>/<run-id>/..., et rendez le préfixe de topics du firmware configurable (via la console série, un fichier de provisioning ou un flag de build) au lieu de le coder en dur. Le schéma complet est décrit dans nommer les topics de test pour qu'ils n'entrent jamais en collision.

Si le préfixe ne peut pas changer à l'exécution, utilisez au moins un préfixe fixe par banc et effacez les messages retenus sous ce préfixe avant chaque test.

Le harnais : un enregistreur MQTT

Le cœur du banc est une petite classe qui s'abonne au préfixe de l'appareil, garde une transcription de chaque message et permet aux tests d'attendre un message précis avec un timeout. Attendre des événements plutôt que d'appeler sleep() est de loin la meilleure parade contre les tests matériels instables.

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

Contrôle de l'alimentation et du port série

Chaque test devrait partir d'un démarrage à froid ; le banc doit donc pouvoir couper l'alimentation. Pour les cartes alimentées en USB sur un hub qui permet de commuter chaque port, uhubctl suffit ; pour les appareils sur secteur, une carte relais ou une PDU commutable font le même travail. Les logs série sont capturés dans un thread d'arrière-plan pour qu'un test en échec puisse joindre exactement ce que le firmware a affiché.

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):
    # Remplacez par l'appel à votre relais ou PDU. Exemple : port 2 du 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()

Dans cet exemple, le firmware publie online sur <prefix>/status après la connexion, définit un Last Will offline sur le même topic, écoute <prefix>/cmd/# et répond sur <prefix>/ack.

Écrire les tests

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 -> le broker abandonne après 1,5 x 30 s
    rig.mqtt.expect("status", timeout=60, match=lambda p: p == "offline")

Le dernier test prouve ce qu'aucun simulateur ne prouve vraiment : une coupure brutale de l'alimentation, un broker qui ne s'en aperçoit qu'à l'expiration du keep alive et un Last Will qui arrive au backend. Voir tester le Last Will pour savoir précisément quand le broker l'envoie.

Injecter des pannes

La valeur du HIL augmente avec chaque mode de défaillance que vous pouvez déclencher à la demande :

• Coupure pendant une publication : coupez l'alimentation juste après une commande et vérifiez que l'appareil récupère et ne remonte pas un état périmé.
• Redémarrage du broker : redémarrez le broker local et vérifiez que l'appareil se reconnecte avec backoff, se réabonne et reprend ses publications. Les scénarios sont détaillés dans tests de reconnexion MQTT.
• Perte réseau sans perte d'alimentation : éteignez le point d'accès Wi-Fi ou le port du switch. L'appareil continue de tourner, sa connexion TCP meurt en silence, et vous découvrez si votre logique de keep alive et de reconnexion fonctionne vraiment.
• Mauvais identifiants ou ACL révoquée : changez le mot de passe de l'appareil sur le broker de test et vérifiez que le firmware journalise la raison du CONNACK au lieu de boucler en silence. La recherche de reason codes aide à décoder ce que remonte l'appareil.

Rapports

Lancez la suite avec pytest --junitxml=report.xml pour que n'importe quel système de CI affiche le résultat de chaque test. En cas d'échec, joignez le log série et la transcription MQTT de l'enregistreur : ensemble, ils indiquent si l'appareil n'a jamais envoyé le message, l'a envoyé sur le mauvais topic ou si le broker l'a refusé. Un runner de CI self-hosted relié au banc permet à chaque merge de firmware de lancer une exécution sur le matériel ; côté workflow, c'est la même chose que pour les tests MQTT dans GitHub Actions.

Un bon rapport HIL contient :

1. La version du firmware et le commit git de l'appareil sous test.
2. Le nom du banc, l'endpoint du broker et l'ID d'exécution (le préfixe de topics).
3. Le résultat et la durée de chaque test, ainsi que les temps mesurés.
4. Le log série et la transcription MQTT de chaque test en échec.

Garder un banc fiable

• Reflashez l'appareil au début de chaque exécution pour que le test tourne toujours sur la build que vous croyez.
• Effacez les messages retenus sous le préfixe du banc avant l'exécution, sinon un vieux online peut faire passer un démarrage raté pour un succès.
• Gardez une image de firmware réputée bonne ; si elle se met à échouer, c'est le banc qui est cassé, pas le code.
• Mettez les tests instables en quarantaine au lieu de les relancer à l'aveugle, et corrigez la cause, généralement une attente manquante ou un topic partagé.

Suivre le trafic du banc dans le navigateur

Les tests hardware-in-the-loop transforment « ça marchait sur mon bureau » en une vérification reproductible qui tourne à chaque build. Commencez avec une carte, un interrupteur d'alimentation et trois tests (démarrage, aller-retour d'une commande et coupure d'alimentation), puis enrichissez le catalogue de pannes au fil de ce que le terrain vous apprend.

Questions fréquentes