Testes de contrato de mensagens MQTT com JSON Schema

Um broker MQTT roteia bytes; ele nunca verifica o que há dentro deles. Quando o time de firmware renomeia temp para temperature, a publicação dá certo, o broker entrega e o dashboard simplesmente para de mostrar dados, sem erro algum. Os testes de contrato de mensagens MQTT com JSON Schema tornam esse acordo explícito: o formato do payload fica em um arquivo de schema, os publishers provam nos testes que produzem mensagens válidas, os consumidores provam que as aceitam e o serviço de ingestão rejeita tudo o que não bate.

Este guia mostra como escrever um schema de payload, validá-lo em Python e Node.js, ligá-lo a testes de produtor e consumidor, validar em tempo de execução e evoluir o schema sem quebrar subscribers.

Por que contratos importam mais no MQTT

No HTTP, uma requisição ruim falha no servidor com um 400 e quem chamou vê o erro. No publish/subscribe, o publisher nunca fica sabendo que um subscriber engasgou. A falha aparece depois, em outro lugar, muitas vezes como dados faltando e não como erro.

Bons contratos começam com bons payloads. Veja boas práticas de design de payload MQTT para nomes de campos, unidades e timestamps.

Escreva o schema

Mantenha os schemas em um diretório (ou repositório) compartilhado do qual dependam tanto o código do dispositivo quanto o do backend. Um arquivo por tipo de mensagem e versão major:

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

Duas decisões de design importam aqui:

• additionalProperties: true permite que produtores adicionem campos opcionais sem quebrar consumidores antigos. Os consumidores devem ignorar o que não conhecem.
• O campo schema torna cada mensagem autodescritiva, então um consumidor consegue escolher o validador certo, e uma entrada de dead letter diz exatamente qual contrato foi violado.

Validar em Python

O pacote jsonschema suporta o draft 2020-12. Observe que as palavras-chave format só são verificadas quando você passa um format checker, e date-time precisa do pacote opcional rfc3339-validator (pip install "jsonschema[format]" 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)        # falha cedo se o schema estiver quebrado
    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 retorna todas as violações em vez de parar na primeira, o que deixa falhas de teste e logs de dead letter muito mais úteis.

Testes de produtor: prove que você envia mensagens válidas

O teste de produtor chama o código real que monta o payload (encoder do firmware, simulador, gateway) e valida a saída. Não precisa de broker:

import json
from contracts import contract_errors
from device.telemetry import build_payload   # seu 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) == []

Depois adicione um teste que passe pelo broker, para pegar também problemas de codificação na camada MQTT. Publique com propriedades do MQTT 5 que descrevem o conteúdo:

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

Em uma suíte real, espere o on_subscribe antes de publicar (ou use uma fixture como em testes de integração MQTT). Content type e user properties são recursos do MQTT 5; no MQTT 3.1.1, conte com o campo schema no payload.

Testes de consumidor: prove que você aceita mensagens válidas

Os testes de consumidor vão na direção oposta. Mantenha uma pasta de payloads de exemplo, validados contra o schema no CI, e passe-os ao handler do consumidor:

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

Os arquivos de exemplo viram as fixtures compartilhadas: quando o time produtor adiciona um campo, adiciona um exemplo, e a suíte de testes de cada consumidor roda contra ele.

Validar em tempo de execução com Ajv

Os testes protegem o código que você entrega. Os dispositivos em campo continuam rodando o firmware que tiverem. Valide na borda de ingestão e envie as violações para um topic de dead letter. Em Node.js com MQTT.js 5 e 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 é válido: armazene
});

Compile o schema uma única vez na inicialização; ajv.compile é a etapa cara, e a função compilada valida cada mensagem rapidamente.

Versionamento sem quebrar subscribers

Classifique cada mudança de schema antes do merge:

Depois teste a compatibilidade mecanicamente: valide todos os exemplos v1 contra o schema v2 (consumidores novos conseguem ler dispositivos antigos?) e todos os exemplos v2 contra o schema v1 (consumidores antigos toleram dispositivos novos?). As duas checagens são poucas linhas em cima dos testes parametrizados acima.

Para uma mudança incompatível, sinalize a versão no campo schema, em um nível do topic (.../telemetry/v2) ou em uma user property do MQTT 5, e rode os dois validadores até nenhum dispositivo publicar v1. Uma versão no nível do topic é fácil de observar com um wildcard: assine testmqtt/contracts/+/telemetry no cliente MQTT online e veja quais versões ainda estão ativas.

Inspecione o tráfego dos testes de contrato

Checklist

1. Um JSON Schema por tipo de mensagem e versão major, em um local compartilhado.
2. Um campo schema (ou user property) em toda mensagem.
3. Testes de produtor validam a saída real do encoder; testes de consumidor rodam sobre exemplos compartilhados.
4. Validação em tempo de execução na ingestão, com um topic de dead letter e o texto do erro.
5. Testes de compatibilidade entre versões no CI (exemplo com GitHub Actions).
6. Combine com fuzz testing para que entradas inválidas sejam rejeitadas, e não apenas as válidas aceitas.

Um arquivo de schema custa uma tarde. Ele troca o buraco silencioso no dashboard depois de uma versão de firmware por um teste falhando no pull request que o causou, e esse é o lugar mais barato para consertar um contrato quebrado.

Perguntas frequentes