Tests de bout en bout IoT sur MQTT : de l'appareil à l'UI

Les tests unitaires prouvent que chaque service fonctionne. Vos clients, eux, veulent savoir si la mesure envoyée par un capteur apparaît sur leur dashboard et si le bouton qu'ils cliquent atteint l'appareil. Les tests de bout en bout pour les systèmes IoT sur MQTT vérifient exactement cela : un appareil simulé publie via le vrai broker, et le test vérifie chaque étape, du service d'ingestion et de la base de données jusqu'à l'API et l'UI, puis renvoie une commande et vérifie que l'appareil l'a reçue.

Ce guide montre comment structurer ce test, avec du code pytest et Playwright exécutable, et comment le garder assez rapide et fiable pour qu'il tourne à chaque merge.

Le chemin parcouru par un événement

Un système IoT typique basé sur MQTT a deux sens de circulation, et une suite de bout en bout doit couvrir les deux :

Sens	Étapes	Ce qui peut casser
Uplink (télémétrie)	Appareil → broker → service d'ingestion → base de données → API → UI	Changements de topic ou de payload, ACL, parsing, fuseaux horaires, cache
Downlink (commandes)	UI → API → backend → broker → appareil → ack	Mauvais topic, permissions manquantes, QoS, appareils hors ligne
Présence	Déconnexion de l'appareil → Last Will → backend → statut dans l'UI	Réglages de keep alive, topic du LWT, état obsolète

Chaque étape a déjà ses propres tests (voir les tests d'intégration MQTT pour la frontière avec le broker). La suite de bout en bout est petite, une poignée de scénarios, mais chacun traverse toutes les frontières à la fois.

Construire un simulateur qui se comporte comme le firmware

Le simulateur est la pièce la plus importante. Il doit utiliser le même schéma de client ID, les mêmes topics, QoS, format de payload et Last Will que le vrai firmware, sinon vous testez un autre système. Une version Python minimale avec paho-mqtt 2.x :

sim_device.py

import json, queue, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

class SimDevice:
    def __init__(self, device_id, host, user, password, prefix="testmqtt/e2e"):
        self.id = device_id
        self.base = f"{prefix}/{device_id}"
        self.commands = queue.Queue()
        self.subscribed = queue.Queue()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=device_id,
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(user, password)
        self.client.tls_set()
        self.client.will_set(f"{self.base}/status", "offline", qos=1, retain=True)
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, rc, p: self.subscribed.put(mid)
        self.client.on_message = self._on_message
        self.host = host
        self.crashed = False

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.publish(f"{self.base}/status", "online", qos=1, retain=True)
        client.subscribe(f"{self.base}/cmd", qos=1)

    def _on_message(self, client, userdata, msg):
        cmd = json.loads(msg.payload)
        self.commands.put(cmd)
        client.publish(f"{self.base}/cmd/ack", json.dumps({"id": cmd["id"], "ok": True}), qos=1)

    def start(self):
        self.client.connect(self.host, 8883, keepalive=30)
        self.client.loop_start()
        self.subscribed.get(timeout=10)        # SUBACK reçu : prêt pour les commandes

    def send_reading(self, temp_c):
        payload = {"schema": "telemetry/v1", "device": self.id, "temp_c": temp_c,
                   "ts": datetime.now(timezone.utc).isoformat()}
        self.client.publish(f"{self.base}/telemetry", json.dumps(payload), qos=1).wait_for_publish()

    def crash(self):
        """Coupe la connexion TCP sans DISCONNECT pour que le broker envoie le Last Will."""
        self.crashed = True
        self.client.loop_stop()
        self.client.socket().close()

    def stop(self):
        if not self.crashed:
            # efface le statut retenu pour garder le broker propre
            self.client.publish(f"{self.base}/status", "", qos=1, retain=True).wait_for_publish()
            self.client.disconnect()
        self.client.loop_stop()

Si vous avez déjà un simulateur de flotte, réutilisez-le ; voir construire un simulateur d'appareils MQTT pour étendre la même idée à de nombreux appareils. Pour éviter que le payload ne diverge entre firmware et simulateur, un JSON Schema partagé et des tests de contrat maintiennent les deux alignés.

Attendre des conditions, jamais un sleep

Les pipelines asynchrones ont une latence variable. Un time.sleep(2) fixe est trop court sur un runner de CI chargé ou fait perdre du temps partout ailleurs. Interrogez une condition jusqu'à ce qu'elle soit vraie ou qu'un délai expire :

eventually.py

import time

def eventually(check, timeout=15.0, interval=0.25):
    """Appelle check() jusqu'à ce qu'il ne lève plus AssertionError, sinon relance après le timeout."""
    deadline = time.monotonic() + timeout
    while True:
        try:
            return check()
        except AssertionError:
            if time.monotonic() > deadline:
                raise
            time.sleep(interval)

Le test d'uplink : de l'appareil à la base et à l'API

Chaque exécution reçoit un ID d'appareil unique, pour que les exécutions parallèles et les restes d'exécutions précédentes n'interfèrent jamais :

test_e2e_uplink.py

import os, uuid
import httpx, psycopg, pytest
from eventually import eventually
from sim_device import SimDevice

HOST = "broker.connectmq.com"
API = os.environ.get("API_URL", "http://localhost:8000")
DSN = os.environ.get("DATABASE_URL", "postgresql://app:app@localhost:5432/app")

@pytest.fixture
def device():
    dev = SimDevice(f"e2e-{uuid.uuid4().hex[:8]}", HOST,
                    os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    dev.start()
    yield dev
    dev.stop()
    with psycopg.connect(DSN) as db:
        db.execute("DELETE FROM readings WHERE device_id = %s", (dev.id,))

def test_reading_reaches_database_and_api(device):
    device.send_reading(21.5)

    def stored():
        with psycopg.connect(DSN) as db:
            row = db.execute(
                "SELECT temp_c FROM readings WHERE device_id = %s ORDER BY ts DESC LIMIT 1",
                (device.id,),
            ).fetchone()
        assert row is not None, "reading not stored yet"
        assert row[0] == pytest.approx(21.5)
    eventually(stored)

    def served():
        r = httpx.get(f"{API}/devices/{device.id}/latest", timeout=5)
        assert r.status_code == 200
        assert r.json()["temp_c"] == pytest.approx(21.5)
    eventually(served)

La vérification en base vous dit que l'ingestion fonctionne ; celle de l'API, que les requêtes, la sérialisation et le cache fonctionnent par-dessus. Quand le test échoue, la première assertion en échec nomme l'étape cassée : c'est le principal intérêt de vérifier chaque couche plutôt que la seule UI. Le guide MQTT vers PostgreSQL montre un service d'ingestion correspondant.

Le test UI et downlink avec Playwright

Le test navigateur vérifie ce que voit l'utilisateur et emprunte le chemin des commandes jusqu'à l'appareil. Avec Playwright pour Python :

test_e2e_ui.py

import os
from playwright.sync_api import Page, expect
from eventually import eventually

APP = os.environ.get("APP_URL", "http://localhost:3000")

def test_dashboard_shows_reading_and_sends_command(page: Page, device):
    device.send_reading(23.0)

    page.goto(f"{APP}/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)
    expect(page.get_by_test_id("temp-value")).to_have_text("23.0 °C", timeout=15_000)

    page.get_by_role("button", name="Reboot").click()

    cmd = device.commands.get(timeout=15)        # le simulateur l'a reçue via MQTT
    assert cmd["action"] == "reboot"
    expect(page.get_by_test_id("last-command")).to_contain_text("acknowledged", timeout=15_000)

Ce test réutilise la fixture device (déplacez-la dans conftest.py). L'expect de Playwright réessaie déjà jusqu'à son timeout, donc aucun polling supplémentaire n'est nécessaire dans le navigateur. Utilisez des attributs data-testid plutôt que des classes CSS : ils survivent aux refontes.

Présence : tester le chemin du Last Will

La détection des appareils hors ligne est une fonctionnalité sur laquelle les utilisateurs comptent et qui est rarement testée. Simulez un crash, pas une déconnexion propre, car un DISCONNECT normal supprime le Last Will :

test_e2e_presence.py

from playwright.sync_api import Page, expect

def test_crashed_device_shows_offline(page: Page, device):
    page.goto(f"http://localhost:3000/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)

    device.crash()

    # un socket fermé déclenche le Will aussitôt ; un lien mort en silence prend jusqu'à 1,5 x keepalive
    expect(page.get_by_test_id("device-status")).to_have_text("offline", timeout=60_000)

Un socket TCP fermé déclenche le Will immédiatement, mais un appareil qui perd son alimentation ou sa couverture n'envoie rien, et le broker ne s'en aperçoit qu'après environ 1,5 fois l'intervalle de keep alive : gardez donc un timeout supérieur. Voir tester le Last Will pour les règles de temporisation. Observer testmqtt/e2e/# dans le client MQTT en ligne pendant une exécution locale affiche dans l'ordre les messages de statut, de télémétrie, de commande et d'ack : c'est le moyen le plus rapide de déboguer un scénario en échec.

Observer le trafic du test de bout en bout

Environnement et CI

Lancez le backend, la base de données et l'UI avec Docker Compose et pointez-les, comme le simulateur, vers le même broker :

Lancer la suite en local

export MQTT_USER=username MQTT_PASS=password
docker compose up -d --wait          # api, ingest, postgres, web
pip install pytest pytest-playwright paho-mqtt httpx "psycopg[binary]"
playwright install chromium
pytest -q tests/e2e --maxfail=1

En CI, les mêmes commandes fonctionnent dans un job ; voir les tests MQTT dans GitHub Actions. Gardez une configuration du broker (TLS, ACL, version de MQTT) identique à la production, car les erreurs de permissions et de TLS sont précisément ce que les tests de bout en bout doivent détecter.

Garder la suite fiable

Des IDs uniques par exécution. L'ID d'appareil, le préfixe de topic et les lignes en base dérivent tous d'un même suffixe aléatoire.
Attendre le SUBACK. Le start() du simulateur bloque jusqu'à la confirmation de l'abonnement aux commandes ; sans cela, la première commande peut arriver avant que quiconque n'écoute.
Nettoyer les messages retenus. Publier un payload retenu vide sur le topic de statut dans stop() garde le broker propre ; voir les messages retenus.
Peu de scénarios, pleine profondeur. Cinq scénarios de bout en bout qui traversent chaque étape valent mieux que cinquante qui dupliquent des tests unitaires.
Échouer avec le nom de l'étape. Écrivez des messages d'assertion comme "reading not stored yet", pour que le rapport indique où la chaîne s'est rompue.

Une suite de bout en bout pour un système MQTT, c'est un simulateur, un helper de polling et quelques tests qui suivent un événement de l'appareil jusqu'à l'écran et retour. Elle tourne en quelques minutes, détecte les erreurs d'intégration que les tests unitaires ne voient pas et vous donne une réponse concrète à « est-ce que ça marche ? » avant chaque release.

Questions fréquentes

Quelle différence entre tests d'intégration et tests de bout en bout en IoT ?

Un test d'intégration vérifie une frontière, par exemple que votre service d'ingestion reçoit un message du broker. Un test de bout en bout suit un seul événement à travers tous les composants : appareil, broker, backend, base de données, API et UI, ainsi que le chemin des commandes vers l'appareil.

Faut-il du vrai matériel pour les tests de bout en bout IoT ?

Pas pour la plupart. Un simulateur logiciel qui utilise les mêmes topics et payloads MQTT que le firmware couvre de façon fiable la partie cloud et tourne en CI. Gardez une petite suite hardware-in-the-loop pour la radio, l'énergie et les comportements propres au firmware.

Comment éviter que les tests de bout en bout MQTT soient instables ?

Donnez à chaque exécution un ID d'appareil et un préfixe de topic uniques, attendez des conditions avec un timeout plutôt que des sleeps fixes, attendez le SUBACK avant de publier et nettoyez les lignes et les messages retenus créés par le test.

Tests de bout en bout pour les systèmes IoT sur MQTT