Tester les clients MQTT : unitaires, intégration et E2E

Q: Dois-je mocker la bibliothèque MQTT dans les tests unitaires ?

Mockez à votre propre frontière, par exemple une fonction publish que vous injectez, plutôt que les mécanismes internes de la bibliothèque. Vérifiez le comportement réel du protocole avec un test d'intégration contre un broker en conteneur.

Tester les clients MQTT est délicat, car le comportement intéressant se produit à travers un réseau, dans des callbacks, sur le thread de quelqu'un d'autre. Ce guide présente une approche en couches qui fonctionne : des tests unitaires rapides avec un client mocké, des tests d'intégration contre un broker éphémère en conteneur dans la CI, et un court smoke test de bout en bout contre un vrai broker cloud avant chaque release.

Pourquoi le code client MQTT est difficile à tester

Le code MQTT mêle trois préoccupations qui échouent de manières différentes :

La logique métier : analyse des payloads, décision de ce qu'il faut publier, association des topics aux appareils.
Le comportement du protocole : QoS, retained messages, reconnexions, réabonnements, last will.
L'environnement : certificats TLS, identifiants, ports, pare-feu, quotas du broker.

La plupart des suites de tests MQTT instables essaient de couvrir les trois dans un seul test qui se connecte à un broker et attend avec sleep. Les séparer permet à chaque couche d'utiliser l'outil le moins coûteux capable de détecter ses bugs.

Couche	Ce qu'elle détecte	Broker nécessaire	Vitesse	Exécution
Unitaire	Analyse, association des topics, logique de décision	Non (mocké)	Millisecondes	À chaque sauvegarde
Intégration	Câblage subscribe/publish, QoS, retained, reconnexion	Conteneur local	Secondes	À chaque pull request
Smoke test de bout en bout	TLS, identifiants, ports, réseau réel	Vrai broker cloud	Secondes	Avant release, de façon planifiée

Les exemples utilisent Python avec paho-mqtt 2.x et pytest, mais la structure se transpose à n'importe quel langage.

Étape 1 : concevoir du code testable

Le changement le plus utile consiste à garder le client MQTT en périphérie. Placez les décisions dans de simples fonctions ou classes qui reçoivent une fonction de publication en dépendance, et laissez un adaptateur léger les relier à paho.

thermostat.py :

import json
from dataclasses import dataclass
from typing import Callable

Publish = Callable[[str, str, int, bool], None]


@dataclass
class Thermostat:
    prefix: str
    publish: Publish
    setpoint: float = 21.0
    hysteresis: float = 0.5
    heating: bool = False

    def on_temperature(self, topic: str, payload: bytes) -> None:
        try:
            value = float(json.loads(payload)["value"])
        except (ValueError, KeyError, TypeError, json.JSONDecodeError):
            return  # ignorer les messages malformés

        room = topic.split("/")[-2]
        if not self.heating and value < self.setpoint - self.hysteresis:
            self.heating = True
        elif self.heating and value > self.setpoint + self.hysteresis:
            self.heating = False
        else:
            return

        self.publish(f"{self.prefix}/{room}/heater/set",
                     "ON" if self.heating else "OFF", 1, True)

adapter.py :

import threading

import paho.mqtt.client as mqtt
from thermostat import Thermostat


def build_client(host, port, client_id, prefix, username=None, password=None, tls=False):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if username:
        client.username_pw_set(username, password)
    if tls:
        client.tls_set()

    thermostat = Thermostat(
        prefix=prefix,
        publish=lambda t, p, q, r: client.publish(t, p, qos=q, retain=r),
    )

    def on_connect(client, userdata, flags, reason_code, properties):
        if not reason_code.is_failure:
            client.subscribe(f"{prefix}/+/temperature", qos=1)

    ready = threading.Event()

    def on_subscribe(client, userdata, mid, reason_codes, properties):
        ready.set()

    def on_message(client, userdata, msg):
        thermostat.on_temperature(msg.topic, msg.payload)

    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    return client, thermostat, ready

S'abonner dans on_connect est délibéré : cela rétablit les abonnements après chaque reconnexion, et l'événement ready renvoyé permet aux tests d'attendre que l'abonnement soit actif au lieu de deviner avec sleep(). C'est exactement le type de câblage que les tests d'intégration vont vérifier.

Étape 2 : tests unitaires avec un client mocké

Comme Thermostat reçoit publish en paramètre, un Mock remplace entièrement le réseau.

test_thermostat.py :

import json
from unittest.mock import Mock

from thermostat import Thermostat

PREFIX = "home/test"


def make():
    pub = Mock()
    return Thermostat(prefix=PREFIX, publish=pub), pub


def msg(value):
    return json.dumps({"value": value}).encode()


def test_turns_heater_on_below_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    pub.assert_called_once_with(f"{PREFIX}/kitchen/heater/set", "ON", 1, True)


def test_no_publish_inside_hysteresis_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.2))
    pub.assert_not_called()


def test_turns_heater_off_above_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.6))
    assert pub.call_args.args[1] == "OFF"


def test_ignores_garbage():
    t, pub = make()
    for bad in [b"", b"not json", b'{"temp": 3}', b'{"value": "hot"}']:
        t.on_temperature(f"{PREFIX}/kitchen/temperature", bad)
    pub.assert_not_called()

Ces tests s'exécutent en quelques millisecondes et couvrent la logique qui contient le plus souvent des bugs. Résistez à la tentation de mocker paho lui-même en détail (vérifier que client.subscribe a été appelé avec des arguments exacts) : de tels tests reflètent l'implémentation, cassent lors des refactorisations et ne prouvent toujours pas que le broker accepte ce que vous envoyez. C'est le rôle de la couche d'intégration.

Étape 3 : tests d'intégration contre un broker en conteneur

Un vrai broker dans un conteneur offre un comportement de protocole authentique, sans état partagé ni dépendance à Internet. L'image officielle Eclipse Mosquitto inclut un fichier de configuration qui autorise les connexions anonymes, idéal pour les tests :

docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf

Écrivez maintenant des tests qui exercent le véritable adaptateur. Un utilitaire collecte les messages sur un second client et attend une condition au lieu de dormir pendant une durée fixe.

test_integration.py :

import os
import queue
import uuid

import paho.mqtt.client as mqtt
import pytest

from adapter import build_client

HOST = os.getenv("MQTT_TEST_HOST", "localhost")
PORT = int(os.getenv("MQTT_TEST_PORT", "1883"))


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


@pytest.fixture
def probe(prefix):
    received = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: received.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/#", qos=1)
    c.loop_start()
    yield c, received
    c.loop_stop()
    c.disconnect()


def wait_for(q, predicate, timeout=5.0):
    while True:
        topic, payload, retain = q.get(timeout=timeout)  # lève queue.Empty à l'expiration du délai
        if predicate(topic, payload):
            return topic, payload, retain


def test_temperature_drives_heater(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/office/temperature", '{"value": 18.5}', qos=1)
        topic, payload, _ = wait_for(received, lambda t, p: t.endswith("/heater/set"))
        assert topic == f"{prefix}/office/heater/set"
        assert payload == "ON"
    finally:
        client.loop_stop()
        client.disconnect()


def test_heater_command_is_retained(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/hall/temperature", '{"value": 15}', qos=1)
        wait_for(received, lambda t, p: t.endswith("/heater/set"))
    finally:
        client.loop_stop()
        client.disconnect()

    # Un subscriber tardif doit recevoir immédiatement la commande retenue
    late = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"late-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: late.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/hall/heater/set", qos=1)
    c.loop_start()
    try:
        _, payload, retain = late.get(timeout=5)
        assert payload == "ON" and retain is True
    finally:
        c.loop_stop()
        c.disconnect()

Trois habitudes rendent les tests d'intégration stables :

Un préfixe de topic unique par test. Les retained messages d'un test ne peuvent pas fuir dans le suivant.
Attendre des conditions, pas sleep(). Une file avec délai d'expiration échoue rapidement et n'attend jamais plus que nécessaire.
Toujours arrêter les boucles et se déconnecter dans finally. Des threads réseau orphelins font bloquer les tests suivants.

Pour tester la logique de reconnexion, arrêtez puis redémarrez le conteneur en cours de test (docker restart test-broker) et vérifiez que le client se réabonne et reçoit un message publié après le redémarrage.

Exécuter le broker en CI

GitHub Actions peut démarrer le conteneur dans une étape précédant les tests :

name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Start MQTT broker
        run: |
          docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
            mosquitto -c /mosquitto-no-auth.conf
      - run: pip install "paho-mqtt>=2,<3" pytest
      - run: pytest -q

Les tests unitaires et d'intégration peuvent partager une même exécution pytest ; marquez les plus lents avec @pytest.mark.integration si vous voulez les ignorer en local.

Étape 4 : smoke tests de bout en bout contre un vrai broker

Un conteneur ne peut pas vous dire si vos paramètres TLS, votre bundle de CA, vos règles de pare-feu ou vos identifiants de production fonctionnent. Un court smoke test contre le véritable endpoint, si. Pour des projets open source ou des vérifications rapides, le serveur MQTT public gratuit à l'adresse broker.connectmq.com suffit ; il accepte les clients anonymes en TLS sur le port 8883. Gardez des payloads synthétiques, car la zone publique est lisible par tout le monde.

smoke_test.py :

import os
import sys
import threading
import uuid

import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
USER = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"smoke-{uuid.uuid4().hex[:8]}"
TOPIC = f"{os.getenv('MQTT_PREFIX', 'connectmq/demo')}/smoke-{uuid.uuid4().hex[:8]}/ping"

got = threading.Event()
nonce = uuid.uuid4().hex
errors = []


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        errors.append(f"connect failed: {reason_code}")
        got.set()  # réveiller le thread principal ; les callbacks ne peuvent pas terminer le processus
        return
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    client.publish(TOPIC, nonce, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == nonce:
        got.set()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USER:
    client.username_pw_set(USER, PASSWORD)
client.tls_set()
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 8883, keepalive=30)
client.loop_start()
ok = got.wait(timeout=15) and not errors
for e in errors:
    print(e)
client.loop_stop()
client.disconnect()
print("smoke test passed" if ok else "smoke test FAILED: no echo within 15 s")
sys.exit(0 if ok else 1)

En production, pointez le même script vers votre broker privé en définissant MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID et MQTT_PREFIX à partir des secrets de la CI. Sur un broker MQTT managé comme ConnectMQ, les client IDs sont liés aux identifiants émis : créez donc un jeu d'identifiants dédié à la CI plutôt que d'emprunter l'identité d'un appareil, sinon le smoke test déconnectera cet appareil.

Exécutez les smoke tests avant les releases et de façon planifiée (par exemple toutes les heures), pas à chaque commit. Des quotas s'appliquent sur les brokers partagés : le serveur MQTT public gratuit autorise 10 messages par seconde et par client et 10 abonnements par client, et le plan ConnectMQ Serverless gratuit plafonne les connexions simultanées à 25.

Anti-modèles courants

Quelques habitudes rendent les suites de tests MQTT lentes et instables. Surveillez-les en revue de code :

Les attentes fixes. Un time.sleep(2) après une publication est soit trop long, ce qui gaspille des minutes sur l'ensemble de la suite, soit trop court sur un runner de CI chargé. Attendez plutôt une file ou un événement avec délai d'expiration.
Des topics partagés entre tests. Les retained messages et les abonnements résiduels d'un test fuient dans le suivant. Générez un préfixe unique par test.
Les erreurs de callback avalées. Les exceptions dans les callbacks MQTT s'exécutent sur le thread réseau ; elles ne font donc pas échouer le test directement. Capturez-les dans une liste et vérifiez qu'elle est vide lors du teardown.
Des identifiants d'appareils en CI. Emprunter l'identité d'un vrai appareil pour les tests déconnecte cet appareil. Créez des identifiants distincts pour l'automatisation.
Tester la bibliothèque plutôt que votre code. Vous n'avez pas à prouver que paho livre les messages QoS 1 ; vous devez prouver que votre code s'abonne aux bons topics et réagit correctement.
Un unique test géant de bout en bout. Quand il échoue, personne ne sait quelle couche a cassé. Gardez peu de tests de bout en bout, ciblés, et reportez le détail dans les tests unitaires.

Liste de contrôle des tests

La logique métier vit en dehors des callbacks MQTT et est testée unitairement avec un publish mocké.
Les payloads malformés, vides et surdimensionnés sont testés et ignorés en toute sécurité.
Les tests d'intégration utilisent un conteneur de broker neuf et un préfixe de topic unique par test.
Le comportement de reconnexion et de réabonnement est testé en redémarrant le broker.
Le comportement des retained messages et du last will est vérifié avec un subscriber tardif.
Un smoke test TLS avec de vrais identifiants s'exécute avant chaque release.
Les identifiants de la CI sont distincts de ceux des appareils.

Pour les détails propres à chaque langage, consultez le tutoriel du client MQTT PHP, le tutoriel du client Paho MQTT C et le guide plus général des bibliothèques clientes MQTT.

Démarrer avec ConnectMQ

Donnez à votre pipeline de test un vrai broker avec lequel dialoguer. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, ajoutez un jeu d'identifiants dédié à la CI et exécutez le smoke test contre lui à chaque release. L'explorateur de topics en direct et le client intégré au navigateur du portail permettent de voir facilement ce que vos tests ont publié. ConnectMQ exploite le broker, vous gardez les tests.

Questions fréquentes

Dois-je mocker la bibliothèque MQTT dans les tests unitaires ?

Mockez à votre propre frontière, par exemple une fonction publish que vous injectez, plutôt que les mécanismes internes de la bibliothèque. Vérifiez le comportement réel du protocole avec un test d'intégration contre un broker en conteneur.

Quel broker utiliser pour les tests d'intégration MQTT en CI ?

Un conteneur Eclipse Mosquitto éphémère est le choix le plus simple : il démarre en une seconde environ, ne nécessite aucun identifiant avec /mosquitto-no-auth.conf et ne laisse aucun état entre deux exécutions.

Puis-je utiliser broker.connectmq.com dans des tests automatisés ?

Oui pour des smoke tests occasionnels avec des données synthétiques et des topics uniques, dans la limite de 10 messages par seconde et de 10 abonnements par client. Pour des suites de tests intensives, utilisez un conteneur local ou votre propre broker privé.

Comment tester la logique de reconnexion MQTT ?

Redémarrez le conteneur du broker pendant le test et vérifiez que le client se reconnecte, se réabonne et reçoit un message publié après le redémarrage.

Tester les clients MQTT : tests unitaires, d'intégration et de bout en bout