JSON Schema ile MQTT Mesajlarının Sözleşme Testi

Bir MQTT broker'ı baytları yönlendirir; içlerinde ne olduğunu asla kontrol etmez. Firmware ekibi temp alanını temperature olarak yeniden adlandırdığında publish başarılı olur, broker mesajı teslim eder ve dashboard sessizce hiçbir şey göstermemeye başlar. JSON Schema ile MQTT mesajlarının sözleşme testi bu anlaşmayı açık hâle getirir: payload formatı bir şema dosyasında yaşar, publisher'lar testlerde geçerli mesaj ürettiklerini, consumer'lar bu mesajları kabul ettiklerini kanıtlar ve ingestion servisi uymayan her şeyi reddeder.

Bu rehber bir payload şeması yazmayı, onu Python ve Node.js'te doğrulamayı, üretici ve tüketici testlerine bağlamayı, çalışma zamanında doğrulamayı ve şemayı subscriber'ları bozmadan geliştirmeyi gösteriyor.

Sözleşmeler MQTT'de neden daha önemli

HTTP'de hatalı bir istek sunucuda 400 ile başarısız olur ve çağıran taraf bunu görür. Publish/subscribe'da ise publisher bir subscriber'ın tıkandığını asla öğrenmez. Hata daha sonra, başka bir yerde ve çoğu zaman bir hata mesajı olarak değil, eksik veri olarak ortaya çıkar.

İyi sözleşmeler iyi payload'larla başlar. Alan adlandırma, birimler ve zaman damgaları için MQTT payload tasarımı en iyi uygulamaları yazısına bakın.

Şemayı yazın

Şemaları hem cihaz hem de backend kodunun bağımlı olduğu ortak bir dizinde (veya repository'de) tutun. Her mesaj tipi ve major sürüm için bir dosya:

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

Burada iki tasarım kararı önemlidir:

• additionalProperties: true, üreticilerin eski consumer'ları bozmadan opsiyonel alanlar eklemesine izin verir. Consumer'lar bilmedikleri alanları yok saymalıdır.
• schema alanı her mesajı kendini tanımlayan hâle getirir; böylece bir consumer doğru validator'ı seçebilir ve bir dead-letter kaydı hangi sözleşmenin ihlal edildiğini tam olarak söyler.

Python'da doğrulama

jsonschema paketi draft 2020-12'yi destekler. format anahtar kelimelerinin yalnızca bir format checker verdiğinizde kontrol edildiğini ve date-time için opsiyonel rfc3339-validator paketinin gerektiğini unutmayın (pip install "jsonschema[format]" bunu kurar).

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)        # bozuk şemada erkenden hata ver
    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 ilk ihlalde durmak yerine tüm ihlalleri döndürür; bu da test hatalarını ve dead-letter loglarını çok daha kullanışlı kılar.

Üretici testleri: geçerli mesaj gönderdiğinizi kanıtlayın

Üretici testi, payload'ı oluşturan gerçek kodu (firmware encoder'ı, simülatör, gateway) çağırır ve çıktıyı doğrular. Broker gerekmez:

import json
from contracts import contract_errors
from device.telemetry import build_payload   # gerçek encoder'ınız

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

Ardından MQTT katmanındaki kodlama sorunlarını da yakalamak için broker üzerinden geçen bir test ekleyin. İçeriği tanımlayan MQTT 5 özellikleriyle publish edin:

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

Gerçek bir test paketinde publish etmeden önce on_subscribe'ı bekleyin (veya MQTT entegrasyon testleri yazısındaki gibi bir fixture kullanın). Content type ve user property'ler MQTT 5 özellikleridir; MQTT 3.1.1'de payload içindeki schema alanına güvenin.

Tüketici testleri: geçerli mesajları kabul ettiğinizi kanıtlayın

Tüketici testleri ters yönde çalışır. CI'da şemaya karşı doğrulanan örnek payload'lardan oluşan bir klasör tutun ve bunları consumer'ın handler'ına verin:

import pathlib, pytest
from contracts import contract_errors
from backend.ingest import handle_telemetry   # gerçek handler'ınız

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

Örnek dosyalar ortak fixture'lara dönüşür: üretici ekip bir alan eklediğinde bir örnek de ekler ve her consumer'ın test paketi bu örneğe karşı çalışır.

Ajv ile çalışma zamanında doğrulama

Testler, gönderdiğiniz kodu korur. Sahadaki cihazlar ise hangi firmware'e sahipse onu çalıştırmaya devam eder. Ingestion noktasında doğrulayın ve ihlalleri bir dead-letter topic'ine yönlendirin. MQTT.js 5 ve Ajv 8 ile Node.js'te:

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 geçerli: kaydet
});

Şemayı başlangıçta bir kez derleyin; maliyetli adım ajv.compile'dır ve derlenmiş fonksiyon her mesajı hızla doğrular.

Subscriber'ları bozmadan sürümleme

Merge etmeden önce her şema değişikliğini sınıflandırın:

Ardından uyumluluğu mekanik olarak test edin: tüm v1 örneklerini v2 şemasına karşı (yeni consumer'lar eski cihazları okuyabiliyor mu?) ve tüm v2 örneklerini v1 şemasına karşı (eski consumer'lar yeni cihazlara tolerans gösteriyor mu?) doğrulayın. Her iki kontrol de yukarıdaki parametrik testlerin üzerine birkaç satırdan ibarettir.

Uyumsuz bir değişiklikte sürümü schema alanında, bir topic seviyesinde (.../telemetry/v2) veya bir MQTT 5 user property'sinde belirtin ve hiçbir cihaz v1 publish etmeyene kadar iki validator'ı birlikte çalıştırın. Topic seviyesindeki bir sürümü wildcard ile izlemek kolaydır: online MQTT istemcisinde testmqtt/contracts/+/telemetry topic'ine abone olun ve hangi sürümlerin hâlâ aktif olduğunu kontrol edin.

Sözleşme testi trafiğini inceleyin

Kontrol listesi

1. Her mesaj tipi ve major sürüm için ortak bir konumda bir JSON Schema.
2. Her mesajda bir schema alanı (veya user property).
3. Üretici testleri gerçek encoder çıktısını doğrular; tüketici testleri ortak örnekler üzerinde çalışır.
4. Ingestion'da, dead-letter topic'i ve hata metniyle birlikte çalışma zamanı doğrulaması.
5. CI'da sürümler arası uyumluluk testleri (GitHub Actions örneği).
6. Yalnızca geçerli girdilerin kabul edilmesi değil, geçersiz girdilerin de reddedilmesi için fuzz testi ile birleştirin.

Bir şema dosyası bir öğleden sonraya mal olur. Bir firmware sürümünün ardından dashboard'da oluşan sessiz boşluğu, buna sebep olan pull request'te başarısız olan bir testle değiştirir; bozuk bir sözleşmeyi düzeltmenin en ucuz yeri de orasıdır.

Sıkça sorulan sorular