Tests d'intégration MQTT : tester votre application avec un vrai broker

Les tests d'intégration MQTT exécutent votre application face à un vrai broker : votre code se connecte, s'abonne et publie, et le test vérifie les messages qui transitent réellement sur le réseau. Ils détectent ce que les tests unitaires avec mocks ne peuvent pas voir : filtres de topic erronés, acquittements QoS manquants, état retained, refus d'ACL et bugs de reconnexion.

Ce guide présente des modèles éprouvés pour pytest (paho-mqtt 2.x) et Jest (MQTT.js 5), ainsi que les cinq règles qui gardent ces tests rapides et stables.

Unitaire, intégration ou end-to-end ?

Gardez les tests avec mocks pour la logique pure — brokers mock pour tests unitaires couvre cette couche. Tout ce qui dépend du comportement d'un broker (wildcards, retained messages, redélivrance QoS 1, expiration de session) relève des tests d'intégration face à un vrai broker.

Choisir le broker de test

Vous avez trois options pratiques :

• Mosquitto local (installé ou via Docker) — rapide et privé, mais chaque développeur et chaque runner de CI doit le démarrer.
• Conteneur éphémère par exécution — totalement isolé et jetable ; voir tests MQTT avec Testcontainers.
• Broker hébergé — aucune installation, TLS et authentification réels, et le même endpoint depuis les postes de dev, la CI et les appareils de staging.

Votre code de test ne devrait pas se soucier de celui auquel il parle. Lisez l'hôte, le port et les identifiants depuis des variables d'environnement, avec une valeur par défaut qui fonctionne immédiatement :

MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
# MQTT_PORT=8883 for TLS; set MQTT_USER / MQTT_PASS for a private broker

Les cinq règles de tests MQTT stables

1. Un préfixe de topic unique par test. Construisez chaque topic à partir de testmqtt/it/<run-id>/<random-id>. Les workers parallèles, les jobs de CI et les développeurs ne voient alors jamais les messages des autres.
2. Un client ID unique par connexion. Deux connexions avec le même ID se déconnectent mutuellement — une source classique de tests qui perdent des messages « au hasard ».
3. Attendre des événements, jamais dormir. Attendez le CONNACK avant de vous abonner, le SUBACK avant de publier, et un message avec un timeout. Un sleep(1) est soit trop long (suite lente), soit trop court (suite instable).
4. Nettoyer ce qui survit au test. Les retained messages et les sessions persistantes survivent aux déconnexions. Effacez les topics retained avec une publication retained vide et utilisez des sessions propres, sauf si vous testez justement les sessions.
5. Vérifier le contenu, pas le timing ni l'ordre global. MQTT ne garantit l'ordre que par topic et par QoS depuis un même publisher. Corrélez les messages grâce à un ID dans le payload.

pytest : une fixture client réutilisable

La fixture ci-dessous fournit à chaque test un helper connecté qui attend les acquittements et collecte les messages dans une file thread-safe. Placez-la dans conftest.py :

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

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN_ID = os.getenv("GITHUB_RUN_ID", uuid.uuid4().hex[:8])


class MqttProbe:
    def __init__(self):
        self.messages = queue.Queue()
        self.retained = set()
        self._connected = threading.Event()
        self._subacks = {}
        self._cond = threading.Condition()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"it-{uuid.uuid4().hex[:10]}")
        self.client.on_connect = lambda c, u, f, rc, p: self._connected.set()
        self.client.on_subscribe = self._on_subscribe
        self.client.on_message = lambda c, u, msg: self.messages.put(msg)
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        if PORT == 8883:
            self.client.tls_set()

    def _on_subscribe(self, client, userdata, mid, reason_codes, properties):
        with self._cond:
            self._subacks[mid] = reason_codes
            self._cond.notify_all()

    def connect(self, timeout=5):
        self.client.connect(HOST, PORT, keepalive=30)
        self.client.loop_start()
        assert self._connected.wait(timeout), "no CONNACK from broker"

    def subscribe(self, topic, qos=1, timeout=5):
        _, mid = self.client.subscribe(topic, qos)
        with self._cond:
            assert self._cond.wait_for(lambda: mid in self._subacks, timeout), f"no SUBACK for {topic}"
            code = self._subacks.pop(mid)[0]
        assert not code.is_failure, f"subscribe to {topic} refused: {code}"

    def publish(self, topic, payload, qos=1, retain=False, timeout=5):
        info = self.client.publish(topic, payload, qos=qos, retain=retain)
        info.wait_for_publish(timeout)
        assert info.is_published(), f"publish to {topic} not acknowledged"
        if retain:
            self.retained.add(topic)

    def expect(self, topic, timeout=5):
        deadline = time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                msg = self.messages.get(timeout=left)
            except queue.Empty:
                break
            if msg.topic == topic:
                return msg
        raise AssertionError(f"no message on {topic} within {timeout}s")

    def close(self):
        for topic in self.retained:  # clear retained state left by the test
            self.client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
        self.client.disconnect()
        self.client.loop_stop()


@pytest.fixture
def prefix():
    return f"testmqtt/it/{RUN_ID}/{uuid.uuid4().hex[:8]}"


@pytest.fixture
def mqtt_probe():
    probe = MqttProbe()
    probe.connect()
    yield probe
    probe.close()

Un test pour un service applicatif — ici un thermostat qui écoute des commandes et publie son état — se lit alors comme une spécification :

import json
from conftest import HOST, PORT
from myapp.thermostat import ThermostatService


def test_setpoint_command_updates_state(mqtt_probe, prefix):
    service = ThermostatService(host=HOST, port=PORT, prefix=prefix)
    service.start()  # returns once the service has its own SUBACK
    try:
        mqtt_probe.subscribe(f"{prefix}/state")
        mqtt_probe.publish(f"{prefix}/cmd/setpoint", json.dumps({"value": 21.5}))

        msg = mqtt_probe.expect(f"{prefix}/state")
        assert json.loads(msg.payload)["setpoint"] == 21.5
    finally:
        service.stop()

Le prefix est transmis à l'application, et c'est la décision de conception la plus utile pour la testabilité : si votre service construit ses topics à partir d'une racine configurable, chaque test peut s'exécuter dans son propre espace de noms. Notez que service.start() bloque jusqu'à ce que le service soit abonné ; si votre application n'expose pas un tel signal de disponibilité, ajoutez-en un (un événement, un callback ou un message de statut online retained). Sinon, la commande du test peut arriver avant que le service n'écoute. Fixtures pytest pour MQTT présente d'autres variantes, dont des brokers à portée de session.

Jest : des helpers à base de promesses avec MQTT.js 5

MQTT.js 5 fournit connectAsync, subscribeAsync, publishAsync et endAsync, qui rendent les tests linéaires. Le seul helper nécessaire est « attendre le prochain message sur un topic » :

const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

const URL = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const opts = () => ({
  clientId: `it-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  reconnectPeriod: 0,
});

function nextMessage(client, topic, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const onMessage = (t, payload, packet) => {
      if (t !== topic) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ payload: payload.toString(), retain: packet.retain });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on ${topic} within ${timeoutMs} ms`));
    }, timeoutMs);
    client.on('message', onMessage);
  });
}

let publisher;
let prefix;

beforeAll(async () => {
  publisher = await mqtt.connectAsync(URL, opts());
});
afterAll(() => publisher.endAsync());
beforeEach(() => {
  prefix = `testmqtt/it/${process.env.GITHUB_RUN_ID ?? 'local'}/${randomUUID().slice(0, 8)}`;
});

test('late subscribers receive the retained device config', async () => {
  const topic = `${prefix}/config`;
  await publisher.publishAsync(topic, JSON.stringify({ interval: 30 }), { qos: 1, retain: true });

  const device = await mqtt.connectAsync(URL, opts());
  try {
    const received = nextMessage(device, topic); // listen before subscribing
    await device.subscribeAsync(topic, { qos: 1 });
    const msg = await received;
    expect(msg.retain).toBe(true);
    expect(JSON.parse(msg.payload)).toEqual({ interval: 30 });
  } finally {
    await device.endAsync();
    await publisher.publishAsync(topic, '', { qos: 1, retain: true }); // clear retained
  }
}, 15000);

Deux détails évitent les problèmes Jest les plus fréquents. Le listener est enregistré avant subscribeAsync, car le broker envoie les retained messages juste après le SUBACK — parfois avant que la promesse attendue ne soit résolue. Et chaque client est fermé avec endAsync(), pour que Jest ne reste pas bloqué avec « Jest did not exit one second after the test run has completed ». D'autres modèles Jest dans tester MQTT avec Jest.

Testez aussi les chemins d'échec

Le chemin nominal est la partie la moins intéressante d'un client MQTT. Les tests d'intégration permettent de prouver à moindre coût le comportement qui compte sur le terrain :

• Payloads malformés. Publiez du JSON invalide sur votre topic de commandes et vérifiez que le service journalise une erreur et continue de tourner au lieu de faire planter sa boucle réseau.
• Autorisation. Connectez-vous avec un utilisateur qui n'a pas le droit de publier sur un topic et vérifiez que votre code signale le refus. En MQTT 5, une publication QoS 1 renvoie un reason code dans le PUBACK ; en MQTT 3.1.1, le broker abandonne généralement le message ou ferme la connexion.
• Reconnexions. Déconnectez le client de l'application (ou redémarrez un broker local) et vérifiez qu'il se réabonne et traite la commande suivante. Tests de reconnexion MQTT montre comment l'automatiser.

Chacun de ces cas tient en quelques lignes avec les fixtures ci-dessus et détecte des bugs qui n'apparaîtraient sinon que dans les logs de production.

Checklist des tests instables

Si un abonnement est refusé, le reason code du SUBACK vous en donne la raison — 0x87 Not authorized signifie généralement que l'ACL de l'utilisateur de test ne couvre pas le préfixe. Décodez n'importe quel code avec la recherche de reason codes MQTT.

Tester les tests

Avant de faire confiance à une nouvelle suite, observez-la de l'extérieur. Abonnez-vous à testmqtt/it/# dans le client MQTT en ligne et lancez les tests : vous devez voir apparaître le préfixe de chaque test, ses messages circuler, puis les topics retained effacés (un retained message vide) à la fin. Lancez ensuite la suite deux fois en parallèle — par exemple pytest -n 4 avec pytest-xdist, ou deux jobs de CI — et vérifiez que rien ne change.

Observer le trafic des tests d'intégration

Les tests d'intégration face à un vrai broker sont le moyen le moins coûteux de prouver que votre code MQTT se comporte comme en production. Isolez chaque test par topic et par client ID, attendez les acquittements au lieu de dormir, nettoyez l'état retained, et la suite restera verte pour les bonnes raisons. Quand vous voudrez l'exécuter à chaque push, poursuivez avec tester MQTT en CI avec GitHub Actions.

Questions fréquentes