Pruebas de contrato de mensajes MQTT con JSON Schema

Un broker MQTT enruta bytes; nunca comprueba lo que contienen. Cuando el equipo de firmware renombra temp a temperature, la publicación tiene éxito, el broker la entrega y el dashboard deja de mostrar datos sin decir nada. Las pruebas de contrato de mensajes MQTT con JSON Schema hacen explícito ese acuerdo: el formato del payload vive en un fichero de esquema, los publishers demuestran en sus tests que producen mensajes válidos, los consumidores demuestran que los aceptan y el servicio de ingesta rechaza todo lo que no encaje.

Esta guía muestra cómo escribir un esquema de payload, validarlo en Python y Node.js, conectarlo a tests de productor y consumidor, validar en tiempo de ejecución y evolucionar el esquema sin romper subscribers.

Por qué los contratos importan más en MQTT

En HTTP, una petición incorrecta falla en el servidor con un 400 y quien llama lo ve. En publish/subscribe, el publisher nunca se entera de que un subscriber se atragantó. El fallo aparece más tarde, en otro sitio, a menudo como datos que faltan y no como un error.

Los buenos contratos empiezan con buenos payloads. Consulta buenas prácticas de diseño de payload MQTT para nombres de campos, unidades y timestamps.

Escribe el esquema

Guarda los esquemas en un directorio (o repositorio) compartido del que dependan tanto el código del dispositivo como el del backend. Un fichero por tipo de mensaje y versión mayor:

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

Aquí importan dos decisiones de diseño:

• additionalProperties: true permite a los productores añadir campos opcionales sin romper a los consumidores antiguos. Los consumidores deben ignorar lo que no conocen.
• El campo schema hace que cada mensaje se describa a sí mismo, de modo que un consumidor puede elegir el validador correcto y una entrada de dead letter te dice exactamente qué contrato ha incumplido.

Validar en Python

El paquete jsonschema soporta el draft 2020-12. Ten en cuenta que las palabras clave format solo se comprueban si pasas un format checker, y date-time necesita el paquete opcional rfc3339-validator (pip install "jsonschema[format]" lo instala).

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)        # falla pronto si el esquema está roto
    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 devuelve todas las infracciones en lugar de detenerse en la primera, lo que hace mucho más útiles los fallos de test y los logs de dead letter.

Tests de productor: demuestra que envías mensajes válidos

El test de productor llama al código real que construye el payload (encoder del firmware, simulador, gateway) y valida la salida. No hace falta broker:

import json
from contracts import contract_errors
from device.telemetry import build_payload   # tu encoder real

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

Después añade un test que pase por el broker, para detectar también problemas de codificación en la capa MQTT. Publica con propiedades de MQTT 5 que describan el contenido:

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

En una suite real, espera a on_subscribe antes de publicar (o usa una fixture como en pruebas de integración MQTT). El content type y las user properties son funciones de MQTT 5; en MQTT 3.1.1, apóyate en el campo schema del payload.

Tests de consumidor: demuestra que aceptas mensajes válidos

Los tests de consumidor van en la dirección contraria. Mantén una carpeta de payloads de ejemplo, validados contra el esquema en CI, y pásalos al handler del consumidor:

import pathlib, pytest
from contracts import contract_errors
from backend.ingest import handle_telemetry   # tu handler real

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

Los ficheros de ejemplo se convierten en las fixtures compartidas: cuando el equipo productor añade un campo, añade un ejemplo, y la suite de tests de cada consumidor se ejecuta contra él.

Validar en tiempo de ejecución con Ajv

Los tests protegen el código que despliegas. Los dispositivos en campo siguen ejecutando el firmware que tengan. Valida en el punto de ingesta y envía las infracciones a un topic de dead letter. En Node.js con MQTT.js 5 y 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 es válido: guárdalo
});

Compila el esquema una sola vez al arrancar; ajv.compile es el paso costoso, y la función compilada valida cada mensaje rápidamente.

Versionado sin romper subscribers

Clasifica cada cambio de esquema antes de hacer merge:

Después prueba la compatibilidad de forma mecánica: valida todos los ejemplos v1 contra el esquema v2 (¿pueden los consumidores nuevos leer dispositivos antiguos?) y todos los ejemplos v2 contra el esquema v1 (¿toleran los consumidores antiguos a los dispositivos nuevos?). Ambas comprobaciones son unas pocas líneas sobre los tests parametrizados de arriba.

Para un cambio incompatible, señala la versión en el campo schema, en un nivel del topic (.../telemetry/v2) o en una user property de MQTT 5, y ejecuta ambos validadores hasta que ningún dispositivo publique ya v1. Una versión a nivel de topic es fácil de vigilar con un wildcard: suscríbete a testmqtt/contracts/+/telemetry en el cliente MQTT online y comprueba qué versiones siguen activas.

Inspecciona el tráfico de los tests de contrato

Checklist

1. Un JSON Schema por tipo de mensaje y versión mayor, en una ubicación compartida.
2. Un campo schema (o user property) en cada mensaje.
3. Los tests de productor validan la salida real del encoder; los de consumidor se ejecutan sobre ejemplos compartidos.
4. Validación en tiempo de ejecución en la ingesta, con un topic de dead letter y el texto del error.
5. Tests de compatibilidad entre versiones en CI (ejemplo con GitHub Actions).
6. Combínalo con fuzz testing para que la entrada inválida se rechace, no solo para que la válida se acepte.

Un fichero de esquema cuesta una tarde. Sustituye el hueco silencioso en el dashboard tras una versión de firmware por un test que falla en el pull request que lo provocó, y ese es el lugar más barato para arreglar un contrato roto.

Preguntas frecuentes