Tests de contrat des messages MQTT avec JSON Schema

Un broker MQTT route des octets ; il ne vérifie jamais leur contenu. Quand l'équipe firmware renomme temp en temperature, la publication réussit, le broker la distribue et le dashboard n'affiche plus rien, sans la moindre erreur. Les tests de contrat des messages MQTT avec JSON Schema rendent cet accord explicite : le format du payload vit dans un fichier de schéma, les publishers prouvent dans leurs tests qu'ils produisent des messages valides, les consommateurs prouvent qu'ils les acceptent, et le service d'ingestion rejette tout ce qui ne correspond pas.

Ce guide montre comment écrire un schéma de payload, le valider en Python et en Node.js, le brancher sur des tests producteur et consommateur, valider à l'exécution et faire évoluer le schéma sans casser les subscribers.

Pourquoi les contrats comptent davantage en MQTT

En HTTP, une mauvaise requête échoue côté serveur avec un 400 et l'appelant le voit. En publish/subscribe, le publisher n'apprend jamais qu'un subscriber s'est étranglé. L'échec apparaît plus tard, ailleurs, souvent sous forme de données manquantes plutôt que d'erreur.

Les bons contrats commencent par de bons payloads. Consultez les bonnes pratiques de conception de payload MQTT pour le nommage des champs, les unités et les timestamps.

Écrire le schéma

Rangez les schémas dans un dossier (ou un dépôt) partagé dont dépendent à la fois le code des appareils et celui du backend. Un fichier par type de message et par version majeure :

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$id": "https://example.com/contracts/telemetry.v1.json",
  "title": "Telemetry v1",
  "type": "object",
  "required": ["schema", "device", "ts", "temp_c"],
  "properties": {
    "schema": { "const": "telemetry/v1" },
    "device": { "type": "string", "pattern": "^[a-z0-9-]{1,64}$" },
    "ts": { "type": "string", "format": "date-time" },
    "temp_c": { "type": "number", "minimum": -60, "maximum": 150 },
    "battery": { "type": "integer", "minimum": 0, "maximum": 100 }
  },
  "additionalProperties": true
}

Deux choix de conception comptent ici :

• additionalProperties: true permet aux producteurs d'ajouter des champs optionnels sans casser les anciens consommateurs. Les consommateurs doivent ignorer ce qu'ils ne connaissent pas.
• Le champ schema rend chaque message autodescriptif : un consommateur peut choisir le bon validateur, et une entrée de dead letter indique exactement quel contrat a été enfreint.

Valider en Python

Le paquet jsonschema prend en charge le draft 2020-12. Notez que les mots-clés format ne sont vérifiés que si vous passez un format checker, et que date-time nécessite le paquet optionnel rfc3339-validator (pip install "jsonschema[format]" l'installe).

import json
import pathlib
from jsonschema import Draft202012Validator

CONTRACTS = pathlib.Path(__file__).parent / "contracts"

def load_validator(name: str) -> Draft202012Validator:
    schema = json.loads((CONTRACTS / f"{name}.json").read_text())
    Draft202012Validator.check_schema(schema)        # échouer tôt si le schéma est cassé
    return Draft202012Validator(schema, format_checker=Draft202012Validator.FORMAT_CHECKER)

VALIDATORS = {"telemetry/v1": load_validator("telemetry.v1")}

def contract_errors(payload: bytes) -> list[str]:
    try:
        doc = json.loads(payload)
    except ValueError as exc:
        return [f"not JSON: {exc}"]
    validator = VALIDATORS.get(doc.get("schema")) if isinstance(doc, dict) else None
    if validator is None:
        return ["unknown or missing schema field"]
    return [
        f"{'/'.join(map(str, e.absolute_path)) or '<root>'}: {e.message}"
        for e in validator.iter_errors(doc)
    ]

iter_errors renvoie toutes les violations au lieu de s'arrêter à la première, ce qui rend les échecs de test et les logs de dead letter bien plus utiles.

Tests producteur : prouver que vous envoyez des messages valides

Le test producteur appelle le vrai code qui construit le payload (encodeur du firmware, simulateur, passerelle) et valide la sortie. Aucun broker n'est nécessaire :

import json
from contracts import contract_errors
from device.telemetry import build_payload   # votre vrai encodeur

def test_telemetry_payload_matches_contract():
    payload = build_payload(device="sensor-01", temp_c=21.5, battery=88)
    assert contract_errors(payload) == []

def test_minimal_payload_matches_contract():
    payload = build_payload(device="sensor-01", temp_c=-4.0, battery=None)
    assert contract_errors(payload) == []

Ajoutez ensuite un test qui passe par le broker, pour détecter aussi les problèmes d'encodage dans la couche MQTT. Publiez avec des propriétés MQTT 5 qui décrivent le contenu :

import os, queue, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
from contracts import contract_errors
from device.telemetry import build_payload

HOST = "broker.connectmq.com"

def make_client():
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()
    c.connect(HOST, 8883)
    c.loop_start()
    return c

def test_published_message_matches_contract():
    topic = f"testmqtt/contracts/{uuid.uuid4().hex[:8]}/telemetry"
    received = queue.Queue()

    sub = make_client()
    sub.on_message = lambda c, u, m: received.put(m)
    sub.subscribe(topic, qos=1)

    props = Properties(PacketTypes.PUBLISH)
    props.ContentType = "application/json"
    props.UserProperty = [("schema", "telemetry/v1")]

    pub = make_client()
    pub.publish(topic, build_payload(device="sensor-01", temp_c=21.5, battery=88),
                qos=1, properties=props).wait_for_publish()

    msg = received.get(timeout=10)
    assert contract_errors(msg.payload) == []
    assert msg.properties.ContentType == "application/json"
    for c in (pub, sub):
        c.loop_stop(); c.disconnect()

Dans une vraie suite, attendez on_subscribe avant de publier (ou utilisez une fixture comme dans les tests d'intégration MQTT). Le content type et les user properties sont des fonctionnalités de MQTT 5 ; en MQTT 3.1.1, appuyez-vous sur le champ schema du payload.

Tests consommateur : prouver que vous acceptez les messages valides

Les tests consommateur vont dans le sens inverse. Gardez un dossier de payloads d'exemple, validés contre le schéma en CI, et passez-les au handler du consommateur :

import pathlib, pytest
from contracts import contract_errors
from backend.ingest import handle_telemetry   # votre vrai handler

EXAMPLES = sorted(pathlib.Path("contracts/examples/telemetry.v1").glob("*.json"))

@pytest.mark.parametrize("path", EXAMPLES, ids=lambda p: p.name)
def test_example_is_valid(path):
    assert contract_errors(path.read_bytes()) == []

@pytest.mark.parametrize("path", EXAMPLES, ids=lambda p: p.name)
def test_consumer_accepts_example(path):
    reading = handle_telemetry(path.read_bytes())
    assert reading.device

Les fichiers d'exemple deviennent les fixtures partagées : quand l'équipe productrice ajoute un champ, elle ajoute un exemple, et la suite de tests de chaque consommateur s'exécute dessus.

Valider à l'exécution avec Ajv

Les tests protègent le code que vous livrez. Les appareils sur le terrain exécutent toujours le firmware qu'ils ont. Validez au point d'ingestion et envoyez les violations sur un topic de dead letter. En Node.js avec MQTT.js 5 et Ajv 8 :

import { readFileSync } from "node:fs";
import mqtt from "mqtt";
import Ajv2020 from "ajv/dist/2020.js";
import addFormats from "ajv-formats";

const ajv = new Ajv2020({ allErrors: true });
addFormats(ajv);
const validate = ajv.compile(JSON.parse(readFileSync("contracts/telemetry.v1.json", "utf8")));

const client = mqtt.connect("mqtts://broker.connectmq.com:8883", {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
});

client.on("connect", () => client.subscribe("testmqtt/contracts/+/telemetry", { qos: 1 }));

client.on("message", (topic, payload) => {
  let doc: unknown;
  try {
    doc = JSON.parse(payload.toString("utf8"));
  } catch {
    doc = undefined;
  }
  if (doc === undefined || !validate(doc)) {
    const reason = doc === undefined ? "not JSON" : ajv.errorsText(validate.errors);
    client.publish("testmqtt/contracts/deadletter", payload, {
      qos: 1,
      properties: { userProperties: { reason, topic } },
    });
    return;
  }
  // doc est valide : on le stocke
});

Compilez le schéma une seule fois au démarrage ; ajv.compile est l'étape coûteuse, et la fonction compilée valide ensuite chaque message rapidement.

Versionner sans casser les subscribers

Classez chaque changement de schéma avant le merge :

Testez ensuite la compatibilité de façon mécanique : validez tous les exemples v1 contre le schéma v2 (les nouveaux consommateurs savent-ils lire les anciens appareils ?) et tous les exemples v2 contre le schéma v1 (les anciens consommateurs tolèrent-ils les nouveaux appareils ?). Ces deux contrôles ne demandent que quelques lignes de plus que les tests paramétrés ci-dessus.

Pour un changement incompatible, signalez la version dans le champ schema, dans un niveau de topic (.../telemetry/v2) ou dans une user property MQTT 5, et faites tourner les deux validateurs jusqu'à ce qu'aucun appareil ne publie plus en v1. Une version au niveau du topic se surveille facilement avec un wildcard : abonnez-vous à testmqtt/contracts/+/telemetry dans le client MQTT en ligne et vérifiez quelles versions sont encore actives.

Inspecter le trafic des tests de contrat

Checklist

1. Un JSON Schema par type de message et par version majeure, dans un emplacement partagé.
2. Un champ schema (ou une user property) sur chaque message.
3. Les tests producteur valident la sortie réelle de l'encodeur ; les tests consommateur tournent sur des exemples partagés.
4. Validation à l'exécution lors de l'ingestion, avec un topic de dead letter et le texte de l'erreur.
5. Tests de compatibilité entre versions en CI (exemple avec GitHub Actions).
6. Combinez avec le fuzz testing pour que les entrées invalides soient rejetées, et pas seulement les valides acceptées.

Un fichier de schéma coûte un après-midi. Il remplace le trou silencieux dans le dashboard après une version de firmware par un test en échec sur la pull request qui l'a provoqué, et c'est l'endroit le moins cher pour réparer un contrat cassé.

Questions fréquentes