Fixtures pytest pour MQTT : tests isolés avec un broker

La plupart des suites de tests MQTT commencent de la même façon : un test qui se connecte, dort deux secondes, publie, redort et vérifie une liste globale. Ça marche le premier jour et devient flaky vers le dixième. Les fixtures pytest règlent le problème en déplaçant connexion, nommage des topics, attente et nettoyage dans quelques briques réutilisables, si bien que le corps de chaque test tient en trois lignes d'intention.

Ce guide construit un conftest.py pour paho-mqtt 2.x avec un broker choisi par variables d'environnement, une factory de clients connectés avec teardown automatique, un topic unique par test, un collecteur de messages avec timeouts et un nettoyeur de messages retenus.

Ce qu'apporte un bon jeu de fixtures MQTT

Fixture	Scope	Rôle
`broker`	session	Hôte, port, identifiants et TLS depuis les variables d'environnement ; skip si injoignable
`run_id`	session	ID aléatoire qui distingue cette exécution de toute autre sur le même broker
`topic`	function	Un préfixe de topic unique par test, dérivé du nom du test
`mqtt_client`	function	Factory qui renvoie des clients connectés et les déconnecte ensuite
`subscriber`	function	Factory : s'abonne à n'importe quel filtre et renvoie un collecteur
`collector`	function	Un subscriber sur le topic du test, avec `wait_for()` et timeout
`retained`	function	Mémorise les topics retenus et les vide au teardown

Trois propriétés comptent : l'isolation (aucun test ne voit les messages d'un autre), pas de sleeps fixes (on attend des événements, avec une échéance) et un nettoyage garanti (le teardown s'exécute même si le test échoue).

Le conftest.py

Installez les dépendances : pip install pytest paho-mqtt. Placez ensuite ceci dans tests/conftest.py :

tests/conftest.py

import os, re, socket, threading, uuid
from dataclasses import dataclass

import paho.mqtt.client as mqtt
import pytest


@dataclass(frozen=True)
class BrokerConfig:
    host: str
    port: int
    username: str | None
    password: str | None
    tls: bool


@pytest.fixture(scope="session")
def broker():
    cfg = BrokerConfig(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        username=os.getenv("MQTT_USER"),
        password=os.getenv("MQTT_PASS"),
        tls=os.getenv("MQTT_TLS") == "1",
    )
    try:
        socket.create_connection((cfg.host, cfg.port), timeout=3).close()
    except OSError as exc:
        pytest.skip(f"MQTT broker {cfg.host}:{cfg.port} unreachable: {exc}")
    return cfg


@pytest.fixture(scope="session")
def run_id():
    return uuid.uuid4().hex[:8]


@pytest.fixture
def topic(request, run_id):
    name = re.sub(r"[^A-Za-z0-9_-]", "_", request.node.name)
    return f"testmqtt/pytest/{run_id}/{name}"


@pytest.fixture
def mqtt_client(broker):
    clients = []

    def make(client_id=None, protocol=mqtt.MQTTv311, **connect_kwargs):
        client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=client_id or f"pytest-{uuid.uuid4().hex[:10]}",
            protocol=protocol,
        )
        if broker.username:
            client.username_pw_set(broker.username, broker.password)
        if broker.tls:
            client.tls_set()
        clients.append(client)

        connected, result = threading.Event(), {}

        def on_connect(c, userdata, flags, reason_code, properties):
            result["rc"] = reason_code
            connected.set()

        client.on_connect = on_connect
        client.connect(broker.host, broker.port, keepalive=30, **connect_kwargs)
        client.loop_start()
        assert connected.wait(10), f"no CONNACK from {broker.host}:{broker.port}"
        assert not result["rc"].is_failure, f"connection refused: {result['rc']}"
        return client

    yield make

    for client in clients:
        client.disconnect()
        client.loop_stop()

Quelques choix délibérés :

Le broker vient de l'environnement. En local, MQTT_HOST=localhost pointe vers un conteneur Mosquitto ; en CI, les mêmes tests tournent contre un broker Testcontainers ou une instance de staging. Par défaut, c'est l'endpoint partagé broker.connectmq.com, suffisant pour des exécutions rapides.
Broker injoignable → skip, pas échec. Un développeur sans Docker lancé obtient une raison de skip claire au lieu de trente erreurs de connexion. Pour que la CI échoue franchement, ajoutez un flag qui transforme le skip en pytest.fail.
La factory attend le CONNACK. connect() n'envoie que le paquet CONNECT ; l'Event attend la réponse du broker et l'assertion affiche le reason code en cas de refus.
Le teardown se déconnecte avant d'arrêter la boucle, pour que le paquet DISCONNECT soit réellement envoyé et que le broker ne publie pas de Last Will pour un client de test.
Des client IDs aléatoires. Deux tests avec le même ID s'éjecteraient mutuellement du broker : voir les bonnes pratiques de client ID, ou générez des IDs lisibles avec le générateur de client ID.

Attendre des messages sans sleep

Le collecteur est la pièce qui fait disparaître time.sleep() de vos tests. Il ajoute les messages entrants sous une variable de condition et laisse le test attendre qu'un nombre donné soit arrivé :

tests/conftest.py (continued)

class MessageCollector:
    def __init__(self):
        self.messages = []
        self._cond = threading.Condition()

    def __call__(self, client, userdata, msg):  # used as on_message
        with self._cond:
            self.messages.append(msg)
            self._cond.notify_all()

    def wait_for(self, count=1, timeout=5.0):
        with self._cond:
            if not self._cond.wait_for(lambda: len(self.messages) >= count, timeout):
                got = [m.topic for m in self.messages]
                raise AssertionError(f"expected {count} message(s) in {timeout}s, got {len(got)}: {got}")
            return list(self.messages[:count])

    def assert_silent(self, duration=1.0):
        with self._cond:
            self._cond.wait_for(lambda: len(self.messages) > 0, duration)
            assert not self.messages, f"unexpected messages: {[m.topic for m in self.messages]}"


def subscribe(client, topic_filter, qos=1, timeout=5.0):
    done, result = threading.Event(), {}

    def on_subscribe(c, userdata, mid, reason_code_list, properties):
        result["codes"] = reason_code_list
        done.set()

    client.on_subscribe = on_subscribe
    client.subscribe(topic_filter, qos=qos)
    assert done.wait(timeout), f"no SUBACK for {topic_filter}"
    assert not result["codes"][0].is_failure, f"subscription refused: {result['codes'][0]}"


@pytest.fixture
def subscriber(mqtt_client):
    def make(topic_filter, qos=1):
        messages = MessageCollector()
        sub = mqtt_client()
        sub.on_message = messages
        subscribe(sub, topic_filter, qos)
        return messages
    return make


@pytest.fixture
def collector(subscriber, topic):
    return subscriber(f"{topic}/#")


@pytest.fixture
def retained(mqtt_client):
    topics = []
    yield topics.append
    if topics:
        cleaner = mqtt_client()
        for t in topics:
            cleaner.publish(t, b"", qos=1, retain=True).wait_for_publish(5)

Attendre le SUBACK compte autant qu'attendre les messages : si un test publie avant que l'abonnement soit actif, le broker jette le message et le test échoue au hasard. Les fixtures subscriber et collector ne rendent la main qu'une fois l'abonnement confirmé par le broker.

assert_silent() couvre le cas négatif — par exemple qu'une ACL refuse un topic ou qu'un message avec retain=False n'est pas rejoué. Elle attend toute la durée et échoue dès que quelque chose arrive.

Des tests qui se lisent comme des spécifications

Avec les fixtures en place, le corps des tests se réduit à l'intention :

tests/test_telemetry.py

import json
import pytest


@pytest.mark.parametrize("qos", [0, 1, 2])
def test_roundtrip(mqtt_client, collector, topic, qos):
    pub = mqtt_client()
    pub.publish(f"{topic}/temp", "21.5", qos=qos).wait_for_publish(5)

    [msg] = collector.wait_for(1)
    assert msg.topic == f"{topic}/temp"
    assert msg.payload == b"21.5"


def test_ordering_within_one_topic(mqtt_client, collector, topic):
    pub = mqtt_client()
    for i in range(20):
        pub.publish(f"{topic}/seq", json.dumps({"i": i}), qos=1)

    received = [json.loads(m.payload)["i"] for m in collector.wait_for(20, timeout=10)]
    assert received == list(range(20))


def test_retained_status_reaches_late_subscriber(mqtt_client, subscriber, topic, retained):
    status = f"{topic}/status"
    retained(status)
    mqtt_client().publish(status, "online", qos=1, retain=True).wait_for_publish(5)

    late = subscriber(status)          # subscribes after the publish
    [msg] = late.wait_for(1)
    assert msg.payload == b"online" and msg.retain

Chaque test reçoit son propre préfixe de topic, par exemple testmqtt/pytest/3f9a1c2e/test_roundtrip_1_ (issu de test_roundtrip[1]) : les variantes paramétrées ne se voient jamais. Pendant le débogage, abonnez-vous à testmqtt/pytest/# dans le client MQTT en ligne pour suivre le trafic de la suite en direct.

Le test des messages retenus enregistre son topic avant de publier. Même si une assertion échoue, le teardown de retained publie un payload retenu vide et l'exécution suivante démarre propre — voir tester les messages retenus pour comprendre pourquoi un état retenu périmé est une source fréquente de résultats flaky.

Options MQTT 5 via la factory

Comme les arguments nommés supplémentaires sont transmis tels quels à connect(), les fonctionnalités MQTT 5 ne demandent aucune nouvelle fixture :

tests/test_sessions.py

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties


def test_mqtt5_connect_with_session_expiry(mqtt_client):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 300
    client = mqtt_client(protocol=mqtt.MQTTv5, clean_start=False, properties=props)
    assert client.is_connected()

Le même schéma s'applique aux tests de user properties, de request-response ou de message expiry.

Exécuter la suite

pytest.ini

[pytest]
testpaths = tests
markers =
    mqtt: tests that need a live MQTT broker

Run locally and in CI

pytest -q                                            # default broker
MQTT_HOST=localhost pytest -q                        # local Mosquitto
MQTT_PORT=8883 MQTT_TLS=1 MQTT_USER=$MQTT_USER MQTT_PASS=$MQTT_PASS pytest -q
pip install pytest-xdist && pytest -n 4              # parallel workers

Ajoutez pytest-timeout avec un timeout = 30 global dans pytest.ini en dernière ligne de défense : un test bloqué sur un message qui n'arrive jamais échoue au lieu de figer le job de CI. Pour un pipeline complet, voir les tests MQTT dans GitHub Actions.

Pièges

Des assertions dans les callbacks. Un assert dans on_message s'exécute sur le thread réseau de paho ; l'échec est avalé et le test passe. Collectez dans le callback, vérifiez dans le test.
Des clients en scope session. Plus rapides, mais un abonnement créé par un test continue de livrer dans le suivant. Si vous partagez un client, désabonnez-le et réinitialisez ses callbacks dans un wrapper en scope function.
Des noms de topic avec wildcards. Les IDs de parametrize peuvent contenir + ou # ; le re.sub() de la fixture topic les supprime.
Des tests unitaires sans broker. La logique qui ne fait que formater des payloads ou choisir des topics relève de tests unitaires rapides avec un faux client — voir les brokers simulés pour tests unitaires.

Une douzaine de lignes de fixtures se rentabilisent vite : les tests ne dorment plus, ne se gênent plus et nettoient derrière eux. Gardez le conftest.py dans votre dépôt, choisissez le broker par variables d'environnement, et la même suite tourne sur un portable, en CI et contre le staging sans modification.

Questions fréquentes

Les fixtures MQTT doivent-elles avoir un scope session ou function ?

Gardez la configuration du broker en scope session et les clients en scope function. Un client neuf par test coûte un aller-retour CONNECT, mais garantit qu'aucun abonnement, callback ou message en file ne fuit d'un test vers le suivant. Ne partagez un client entre tests que si le temps de connexion est un problème mesuré.

Comment attendre un message MQTT dans pytest sans sleep ?

Utilisez un collecteur qui ajoute les messages sous un threading.Condition et expose wait_for(count, timeout). Le test ne bloque que jusqu'à l'arrivée du message et échoue avec un message clair au timeout, au lieu de dormir un temps fixe en espérant.

Les tests MQTT peuvent-ils tourner en parallèle avec pytest-xdist ?

Oui, si chaque test utilise son propre topic et son propre client ID. Dériver le topic d'un ID par exécution et du nom du nœud de test, et générer des client IDs aléatoires, évite que des workers parallèles reçoivent les messages des autres ou s'éjectent mutuellement du broker.

Fixtures pytest pour MQTT : des tests propres, rapides et isolés