Mocker MQTT en tests unitaires : fakes et brokers en mémoire

Pour tester unitairement du code qui parle MQTT, ne mockez pas le broker : mockez la frontière que votre code maîtrise. Placez le client derrière une petite interface, remplacez-le par un mock ou un fake en mémoire dans les tests unitaires, et gardez un vrai broker pour une couche plus fine de tests d'intégration. Vos tests unitaires s'exécutent alors en millisecondes, n'échouent jamais à cause du réseau et prouvent malgré tout que le bon topic et le bon payload partent et que les messages entrants sont correctement traités.

Ce guide présente les trois doublures de test qui fonctionnent en pratique — un client mocké, un fake écrit à la main et un broker embarqué en mémoire — avec des exemples en Python (paho-mqtt 2.x) et JavaScript (MQTT.js 5), et explique où chacune cesse d'être utile.

Choisir la bonne doublure de test

Doublure	Ce que c'est	Idéal pour	Angles morts
Client mock / stub	`unittest.mock`, `jest.fn()` à la place du client de la bibliothèque	Vérifier les appels : topic, payload, QoS, flag retain	Wildcards, QoS, messages retenus, reconnexions : rien du protocole n'est réel
Fake écrit à la main	Une petite classe pub/sub en mémoire avec la même interface	Tester des flux entre plusieurs composants, le routage par topic	Aussi correct que votre fake, pas plus ; pas de réseau, pas d'état de session
Broker embarqué	aedes (Node.js) démarré dans le processus de test	De vrais paquets MQTT sans Docker, très rapide	Ce n'est pas votre broker de production ; ACL, limites et clustering diffèrent
Vrai broker	Mosquitto/EMQX dans un conteneur, ou une instance de test hébergée	Tests d'intégration et end-to-end	Plus lent, demande mise en place et nettoyage

Les deux premières sont des tests unitaires. Les deux dernières sont des tests d'intégration, même quand elles semblent rapides. Garder cette distinction claire vous évite d'écrire des « tests unitaires » qui dépendent en secret d'un broker en marche — et de mocker précisément le comportement que vous vouliez vérifier.

Concevoir d'abord la jointure

Le mock devient indolore quand votre code métier n'appelle jamais directement paho.mqtt.client ou mqtt.connect. Injectez plutôt le client (ou un wrapper fin) :

telemetry.py

import json

class TelemetryPublisher:
    def __init__(self, client, device_id: str):
        self.client = client
        self.topic = f"testmqtt/devices/{device_id}/telemetry"

    def report(self, temperature: float) -> None:
        if not -40 <= temperature <= 125:
            raise ValueError("sensor out of range")
        payload = json.dumps({"t": round(temperature, 2)})
        self.client.publish(self.topic, payload, qos=1, retain=False)

def handle_command(client, userdata, msg):
    """paho-mqtt on_message callback: toggle a relay on 'on'/'off' commands."""
    command = msg.payload.decode().strip().lower()
    if command not in ("on", "off"):
        return
    userdata["relay"].set(command == "on")
    client.publish(msg.topic.replace("/cmd", "/state"), command, qos=1, retain=True)

Rien ici n'ouvre de socket, donc rien ici n'a besoin d'un broker pour être testé.

Python : mocker paho-mqtt avec unittest.mock

Utilisez create_autospec pour que le mock ait la vraie signature de Client : une faute de frappe comme client.publsh(...) fait alors échouer le test au lieu de passer en silence.

test_telemetry.py

import json
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, create_autospec

import paho.mqtt.client as mqtt
from telemetry import TelemetryPublisher, handle_command

class TelemetryTests(unittest.TestCase):
    def setUp(self):
        self.client = create_autospec(mqtt.Client, instance=True)

    def test_report_publishes_json_at_qos1(self):
        TelemetryPublisher(self.client, "dev-42").report(21.456)
        self.client.publish.assert_called_once_with(
            "testmqtt/devices/dev-42/telemetry", json.dumps({"t": 21.46}), qos=1, retain=False
        )

    def test_out_of_range_is_not_published(self):
        with self.assertRaises(ValueError):
            TelemetryPublisher(self.client, "dev-42").report(500)
        self.client.publish.assert_not_called()

    def test_command_sets_relay_and_reports_state(self):
        relay = MagicMock()
        msg = SimpleNamespace(topic="testmqtt/devices/dev-42/cmd", payload=b" ON ", qos=1, retain=False)
        handle_command(self.client, {"relay": relay}, msg)
        relay.set.assert_called_once_with(True)
        self.client.publish.assert_called_once_with(
            "testmqtt/devices/dev-42/state", "on", qos=1, retain=True
        )

if __name__ == "__main__":
    unittest.main()

Deux techniques font l'essentiel du travail : vérifier les appels à publish pour le trafic sortant, et appeler vous-même le callback avec un faux message pour le trafic entrant. Un SimpleNamespace avec topic, payload, qos et retain suffit, puisque votre handler ne lit que ces attributs. Si votre code appelle client.connect() à l'import, utilisez unittest.mock.patch("paho.mqtt.client.Client") — mais voyez-y le signe qu'il faut déplacer la connexion dans une factory injectable.

JavaScript : mocker MQTT.js avec Jest

Un client MQTT.js est un EventEmitter : le mock le plus fidèle est donc un vrai EventEmitter auquel on attache des méthodes jest.fn(). Votre code enregistre ses handlers on('message') comme d'habitude et le test émet les événements.

alarm.test.js

import { EventEmitter } from 'node:events';
import { connect } from 'mqtt';
import { startAlarm } from './alarm.js'; // subscribes to sensors, publishes alarms

jest.mock('mqtt', () => ({ connect: jest.fn() }));

function fakeClient() {
  const client = new EventEmitter();
  client.subscribe = jest.fn((topic, opts, cb) => cb?.(null, [{ topic, qos: opts?.qos ?? 0 }]));
  client.publish = jest.fn((topic, payload, opts, cb) => cb?.());
  client.end = jest.fn();
  return client;
}

test('publishes an alarm when temperature exceeds the threshold', () => {
  const client = fakeClient();
  connect.mockReturnValue(client);

  startAlarm({ url: 'mqtts://broker.connectmq.com:8883', threshold: 30 });
  client.emit('connect');
  expect(client.subscribe).toHaveBeenCalledWith('testmqtt/sensors/+/temp', { qos: 1 }, expect.any(Function));

  client.emit('message', 'testmqtt/sensors/kitchen/temp', Buffer.from('31.5'), { retain: false });
  expect(client.publish).toHaveBeenCalledWith(
    'testmqtt/alarms/kitchen', expect.stringContaining('"temp":31.5'), { qos: 1 }, expect.any(Function)
  );
});

L'événement message transmet (topic, payload, packet) avec payload sous forme de Buffer : conservez cette forme dans le mock, sinon votre test passera avec des strings alors que la production reçoit des Buffers. Pour d'autres patterns (faux timers pour la logique de reconnexion, helpers asynchrones), voir tester MQTT avec Jest.

Un fake en mémoire écrit à la main

Quand plusieurs composants communiquent via des topics, vérifier chaque appel à publish devient fragile. Un petit fake qui route réellement les messages est souvent plus lisible. paho-mqtt fournit topic_matches_sub, le matching des wildcards est donc correct sans effort :

fake_mqtt.py

from types import SimpleNamespace
from paho.mqtt.client import topic_matches_sub

class FakeMqtt:
    """In-memory pub/sub with the subset of the paho API our code uses."""
    def __init__(self):
        self.subscriptions, self.retained, self.published = [], {}, []
        self.on_message = None

    def subscribe(self, topic, qos=0):
        self.subscriptions.append(topic)
        for t, payload in self.retained.items():
            if topic_matches_sub(topic, t):
                self._deliver(t, payload, retain=True)

    def publish(self, topic, payload, qos=0, retain=False):
        data = payload.encode() if isinstance(payload, str) else payload
        self.published.append((topic, data, qos, retain))
        if retain and data:
            self.retained[topic] = data
        elif retain:
            self.retained.pop(topic, None)   # empty retained payload clears
        if any(topic_matches_sub(s, topic) for s in self.subscriptions):
            self._deliver(topic, data, retain=False)

    def _deliver(self, topic, payload, retain):
        if self.on_message:
            self.on_message(self, None, SimpleNamespace(topic=topic, payload=payload, qos=0, retain=retain))

Gardez le fake minuscule et testez-le une fois contre un vrai broker (le même scénario par les deux chemins), sinon vous finirez par tester le fake. Vous pouvez vérifier vos attentes sur les wildcards avec le topic matcher.

Node.js : aedes comme broker en mémoire

Aedes est un vrai broker MQTT que vous pouvez démarrer dans un processus Jest ou Mocha. Il parle le vrai protocole sur un port TCP local, votre code MQTT.js tourne donc sans modification — sans aucun mock :

broker.test.js

import { createServer } from 'node:net';
import { Aedes } from 'aedes';
import mqtt from 'mqtt';

let broker, server, url;

beforeAll(async () => {
  broker = await Aedes.createBroker();      // aedes 0.x: require('aedes')()
  server = createServer(broker.handle);
  await new Promise((resolve) => server.listen(0, resolve)); // random free port
  url = `mqtt://127.0.0.1:${server.address().port}`;
});

afterAll(async () => {
  await new Promise((resolve) => server.close(resolve));
  await new Promise((resolve) => broker.close(resolve));
});

test('retained state is delivered to late subscribers', async () => {
  const pub = await mqtt.connectAsync(url);
  await pub.publishAsync('testmqtt/unit/state', 'ready', { qos: 1, retain: true });

  const sub = await mqtt.connectAsync(url);
  const received = new Promise((resolve) => sub.on('message', (t, p, pkt) => resolve([t, p.toString(), pkt.retain])));
  await sub.subscribeAsync('testmqtt/unit/#', { qos: 1 });

  expect(await received).toEqual(['testmqtt/unit/state', 'ready', true]);
  await Promise.all([pub.endAsync(), sub.endAsync()]);
});

Écouter sur le port 0 donne à chaque fichier de test son propre port libre, ce qui évite les collisions entre workers Jest parallèles. Python n'a pas de broker embarqué aussi répandu ; la plupart des équipes utilisent plutôt un conteneur Mosquitto — voir Testcontainers pour MQTT et fixtures pytest.

Où s'arrêtent les tests unitaires

Les mocks excellent à vérifier vos décisions. Ils ne peuvent pas vérifier celles du broker. Passez un test dans la couche d'intégration lorsqu'il dépend :

de la QoS et des sessions — renvoi après une déconnexion, sessions persistantes, gestion des doublons ;
des messages retenus et du Last Will — ce sont des fonctionnalités du broker ; un fake ne reproduit que vos hypothèses ;
de l'authentification et des ACL — un mock ne renvoie jamais Not authorized ; un vrai broker si, parfois en silence ;
des limites de payload, des topic aliases, des propriétés MQTT 5 — les limites sont fixées par la configuration du broker ;
de TLS, des WebSockets, du timing de reconnexion — un comportement réseau a besoin d'un réseau.

Une pyramide saine comporte beaucoup de tests rapides avec mocks, quelques dizaines de tests d'intégration contre un vrai broker (voir tests d'intégration MQTT) et une poignée de vérifications end-to-end sur l'environnement que vous livrez.

Confronter le mock à la réalité

Avant de faire confiance à un test mocké, vérifiez une fois que ses hypothèses correspondent à un vrai broker. Publiez le même message depuis le client MQTT en ligne vers broker.connectmq.com et comparez : le type du payload, le flag retain pour les abonnés tardifs et les chaînes exactes des topics. S'ils diffèrent de ce que renvoie votre mock, corrigez d'abord le mock.

Comparez votre mock à un vrai broker

Mockez le client pour tester votre code rapidement, utilisez un petit fake ou un broker embarqué quand des composants interagissent, et laissez les garanties du protocole aux tests d'intégration contre un vrai broker. Ce découpage garde la suite rapide sans que les tests mentent sur ce que fait réellement MQTT.

Questions fréquentes

Faut-il mocker le client MQTT ou utiliser un vrai broker dans les tests ?

Les deux, à des niveaux différents. Mockez le client pour tester votre propre logique (quel topic, quel payload, comment vous réagissez à un message) en quelques millisecondes. Utilisez un vrai broker dans les tests d'intégration pour vérifier le comportement du protocole : QoS, messages retenus, ACL et reconnexions.

Aedes est-il un mock ou un vrai broker MQTT ?

Aedes est un vrai broker MQTT écrit en Node.js. Embarqué dans le processus de test avec une persistance en mémoire, il se comporte comme un broker rapide et jetable : les tests qui l'utilisent sont donc de petits tests d'intégration, pas des tests unitaires purs.

Comment simuler un message MQTT entrant dans un test unitaire ?

Appelez directement le handler de messages avec un faux objet message qui possède les champs lus par votre code (topic, payload, qos, retain). Avec paho-mqtt, cela revient à appeler on_message(client, userdata, msg) ; avec MQTT.js, à émettre un événement 'message' sur un client EventEmitter simulé.

Mocker MQTT dans les tests unitaires : fakes, stubs et brokers en mémoire