Contract-Tests für MQTT-Nachrichten mit JSON Schema

Ein MQTT-Broker routet Bytes; er prüft nie, was darin steht. Wenn das Firmware-Team temp in temperature umbenennt, gelingt der Publish, der Broker stellt zu, und das Dashboard zeigt stillschweigend nichts mehr an. Contract-Tests für MQTT-Nachrichten mit JSON Schema machen diese Vereinbarung explizit: Das Payload-Format liegt in einer Schema-Datei, Publisher beweisen in Tests, dass sie gültige Nachrichten erzeugen, Consumer beweisen, dass sie sie akzeptieren, und der Ingestion-Service lehnt alles ab, was nicht passt.

Dieser Guide zeigt, wie du ein Payload-Schema schreibst, es in Python und Node.js validierst, in Producer- und Consumer-Tests einbindest, zur Laufzeit validierst und das Schema weiterentwickelst, ohne Subscriber zu brechen.

Warum Verträge bei MQTT wichtiger sind

Bei HTTP scheitert ein fehlerhafter Request am Server mit einem 400, und der Aufrufer sieht es. Bei Publish/Subscribe erfährt der Publisher nie, dass ein Subscriber sich verschluckt hat. Der Fehler zeigt sich später, an anderer Stelle, oft als fehlende Daten statt als Fehlermeldung.

Gute Verträge beginnen mit guten Payloads. Unter Best Practices für MQTT-Payload-Design findest du Hinweise zu Feldnamen, Einheiten und Zeitstempeln.

Das Schema schreiben

Lege Schemas in ein gemeinsames Verzeichnis (oder Repository), von dem sowohl Geräte- als auch Backend-Code abhängen. Eine Datei pro Nachrichtentyp und Major-Version:

{
  "$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
}

Zwei Designentscheidungen sind hier wichtig:

• additionalProperties: true erlaubt Producern, optionale Felder hinzuzufügen, ohne alte Consumer zu brechen. Consumer sollten ignorieren, was sie nicht kennen.
• Das Feld schema macht jede Nachricht selbstbeschreibend. Ein Consumer kann so den richtigen Validator wählen, und ein Dead-Letter-Eintrag sagt dir genau, welcher Vertrag verletzt wurde.

Validierung in Python

Das Paket jsonschema unterstützt Draft 2020-12. Beachte, dass format-Keywords nur geprüft werden, wenn du einen Format-Checker übergibst, und dass date-time das optionale Paket rfc3339-validator braucht (pip install "jsonschema[format]" installiert es).

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)        # früh scheitern bei kaputtem Schema
    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 liefert alle Verstöße, statt beim ersten abzubrechen. Das macht fehlgeschlagene Tests und Dead-Letter-Logs deutlich aussagekräftiger.

Producer-Tests: beweisen, dass du gültige Nachrichten sendest

Der Producer-Test ruft den echten Code auf, der das Payload baut (Firmware-Encoder, Simulator, Gateway), und validiert die Ausgabe. Ein Broker ist nicht nötig:

import json
from contracts import contract_errors
from device.telemetry import build_payload   # dein echter Encoder

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) == []

Ergänze dann einen Test, der über den Broker läuft, damit du auch Kodierungsprobleme in der MQTT-Schicht erwischst. Publishe mit MQTT-5-Properties, die den Inhalt beschreiben:

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()

Warte in einer echten Suite vor dem Publish auf on_subscribe (oder nutze eine Fixture wie bei MQTT-Integrationstests). Content Type und User Properties sind MQTT-5-Features; bei MQTT 3.1.1 verlässt du dich auf das Feld schema im Payload.

Consumer-Tests: beweisen, dass du gültige Nachrichten akzeptierst

Consumer-Tests laufen in die entgegengesetzte Richtung. Pflege einen Ordner mit Beispiel-Payloads, die in der CI gegen das Schema validiert werden, und füttere sie in den Handler des Consumers:

import pathlib, pytest
from contracts import contract_errors
from backend.ingest import handle_telemetry   # dein echter 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

Die Beispieldateien werden zu gemeinsamen Fixtures: Wenn das Producer-Team ein Feld hinzufügt, fügt es ein Beispiel hinzu, und die Testsuite jedes Consumers läuft dagegen.

Laufzeitvalidierung mit Ajv

Tests schützen den Code, den du auslieferst. Geräte im Feld laufen weiter mit der Firmware, die sie gerade haben. Validiere am Ingestion-Rand und leite Verstöße auf einen Dead-Letter-Topic. In Node.js mit MQTT.js 5 und 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 ist gültig: speichern
});

Kompiliere das Schema einmal beim Start; ajv.compile ist der teure Schritt, und die kompilierte Funktion validiert jede Nachricht schnell.

Versionierung, ohne Subscriber zu brechen

Klassifiziere jede Schema-Änderung vor dem Merge:

Teste die Kompatibilität dann mechanisch: Validiere alle v1-Beispiele gegen das v2-Schema (können neue Consumer alte Geräte lesen?) und alle v2-Beispiele gegen das v1-Schema (tolerieren alte Consumer neue Geräte?). Beide Prüfungen sind nur ein paar Zeilen zusätzlich zu den parametrisierten Tests oben.

Bei einem Breaking Change kennzeichnest du die Version im Feld schema, in einer Topic-Ebene (.../telemetry/v2) oder in einer MQTT-5-User-Property und betreibst beide Validatoren, bis kein Gerät mehr v1 publisht. Eine Version auf Topic-Ebene lässt sich leicht per Wildcard beobachten: Abonniere testmqtt/contracts/+/telemetry im Online-MQTT-Client und prüfe, welche Versionen noch aktiv sind.

Traffic der Contract-Tests ansehen

Checkliste

1. Ein JSON Schema pro Nachrichtentyp und Major-Version, an einem gemeinsamen Ort.
2. Ein Feld schema (oder eine User Property) in jeder Nachricht.
3. Producer-Tests validieren die echte Encoder-Ausgabe; Consumer-Tests laufen gegen gemeinsame Beispiele.
4. Laufzeitvalidierung bei der Ingestion, mit Dead-Letter-Topic und Fehlertext.
5. Versionsübergreifende Kompatibilitätstests in der CI (Beispiel mit GitHub Actions).
6. Kombiniere das mit Fuzz-Tests, damit ungültige Eingaben abgelehnt und nicht nur gültige akzeptiert werden.

Eine Schema-Datei kostet einen Nachmittag. Sie ersetzt die stille Lücke im Dashboard nach einem Firmware-Release durch einen fehlschlagenden Test im verursachenden Pull Request, und dort lässt sich ein gebrochener Vertrag am günstigsten reparieren.

Häufig gestellte Fragen