MQTT Entegrasyon Testi: Uygulamanızı Gerçek Bir Broker ile Test Edin

MQTT entegrasyon testleri uygulamanızı gerçek bir broker'a karşı çalıştırır: kodunuz bağlanır, abone olur ve publish eder; test de gerçekten ağdan geçen mesajlar üzerinde doğrulama yapar. Mock'lu birim testlerinin yakalayamadıklarını yakalarlar: yanlış topic filtreleri, eksik QoS onayları, retained durum, ACL retleri ve yeniden bağlanma hataları.

Bu rehber pytest (paho-mqtt 2.x) ve Jest (MQTT.js 5) için çalışan kalıpları ve bu testleri hızlı ve kararsızlıktan uzak tutan beş kuralı gösteriyor.

Birim, entegrasyon mu yoksa uçtan uca mı?

Mock tabanlı testleri saf mantık için tutun; bu katmanı birim testleri için mock broker'lar yazısı ele alıyor. Bir broker'ın nasıl davrandığına bağlı olan her şey (wildcard'lar, retained mesajlar, QoS 1 yeniden teslimi, oturum süresinin dolması) gerçek bir broker'a karşı yapılan entegrasyon testlerine aittir.

Test broker'ını seçmek

Üç pratik seçeneğiniz var:

• Yerel Mosquitto (kurulu veya Docker ile): hızlı ve özel, ancak her geliştiricinin ve her CI runner'ının onu başlatması gerekir.
• Her çalıştırma için geçici container: tamamen izole ve atılabilir; Testcontainers ile MQTT testleri yazısına bakın.
• Barındırılan broker: kurulum yok, gerçek TLS ve kimlik doğrulama, laptoplardan, CI'dan ve staging cihazlarından aynı endpoint.

Test kodunuzun hangisiyle konuştuğunu umursamaması gerekir. Host, port ve kimlik bilgilerini ortam değişkenlerinden okuyun ve doğrudan çalışan bir varsayılan belirleyin:

MQTT_HOST=broker.connectmq.com
MQTT_PORT=1883
# MQTT_PORT=8883 for TLS; set MQTT_USER / MQTT_PASS for a private broker

Kararlı MQTT testleri için beş kural

1. Her test için benzersiz topic öneki. Her topic'i testmqtt/it/<run-id>/<random-id> üzerinden oluşturun. Böylece paralel test worker'ları, CI job'ları ve geliştiriciler birbirlerinin mesajlarını asla görmez.
2. Her bağlantı için benzersiz client ID. Aynı ID'ye sahip iki bağlantı birbirini düşürür; "rastgele" mesaj kaybeden testlerin klasik kaynağıdır.
3. Olayları bekleyin, asla uyumayın. Abone olmadan önce CONNACK'i, publish etmeden önce SUBACK'i ve mesajı timeout ile bekleyin. Bir sleep(1) ya çok uzundur (yavaş test paketi) ya da çok kısa (kararsız test paketi).
4. Testten sonra kalanları temizleyin. Retained mesajlar ve kalıcı oturumlar bağlantı kopmalarından sağ çıkar. Retained topic'leri boş bir retained publish ile temizleyin ve oturumları test etmiyorsanız temiz oturum kullanın.
5. Zamanlamayı veya global sırayı değil, içeriği doğrulayın. MQTT sıralamayı yalnızca tek bir publisher'dan gelen, topic ve QoS bazında garanti eder. Mesajları payload içindeki bir ID ile eşleştirin.

pytest: yeniden kullanılabilir bir istemci fixture'ı

Aşağıdaki fixture her teste, onayları bekleyen ve mesajları thread-safe bir kuyrukta toplayan bağlı bir yardımcı verir. Bunu conftest.py dosyasına koyun:

import os, queue, threading, time, uuid
import pytest
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN_ID = os.getenv("GITHUB_RUN_ID", uuid.uuid4().hex[:8])


class MqttProbe:
    def __init__(self):
        self.messages = queue.Queue()
        self.retained = set()
        self._connected = threading.Event()
        self._subacks = {}
        self._cond = threading.Condition()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"it-{uuid.uuid4().hex[:10]}")
        self.client.on_connect = lambda c, u, f, rc, p: self._connected.set()
        self.client.on_subscribe = self._on_subscribe
        self.client.on_message = lambda c, u, msg: self.messages.put(msg)
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
        if PORT == 8883:
            self.client.tls_set()

    def _on_subscribe(self, client, userdata, mid, reason_codes, properties):
        with self._cond:
            self._subacks[mid] = reason_codes
            self._cond.notify_all()

    def connect(self, timeout=5):
        self.client.connect(HOST, PORT, keepalive=30)
        self.client.loop_start()
        assert self._connected.wait(timeout), "no CONNACK from broker"

    def subscribe(self, topic, qos=1, timeout=5):
        _, mid = self.client.subscribe(topic, qos)
        with self._cond:
            assert self._cond.wait_for(lambda: mid in self._subacks, timeout), f"no SUBACK for {topic}"
            code = self._subacks.pop(mid)[0]
        assert not code.is_failure, f"subscribe to {topic} refused: {code}"

    def publish(self, topic, payload, qos=1, retain=False, timeout=5):
        info = self.client.publish(topic, payload, qos=qos, retain=retain)
        info.wait_for_publish(timeout)
        assert info.is_published(), f"publish to {topic} not acknowledged"
        if retain:
            self.retained.add(topic)

    def expect(self, topic, timeout=5):
        deadline = time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                msg = self.messages.get(timeout=left)
            except queue.Empty:
                break
            if msg.topic == topic:
                return msg
        raise AssertionError(f"no message on {topic} within {timeout}s")

    def close(self):
        for topic in self.retained:  # clear retained state left by the test
            self.client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
        self.client.disconnect()
        self.client.loop_stop()


@pytest.fixture
def prefix():
    return f"testmqtt/it/{RUN_ID}/{uuid.uuid4().hex[:8]}"


@pytest.fixture
def mqtt_probe():
    probe = MqttProbe()
    probe.connect()
    yield probe
    probe.close()

Bir uygulama servisi için yazılan test — burada komutları dinleyip durumunu raporlayan bir termostat — artık bir spesifikasyon gibi okunur:

import json
from conftest import HOST, PORT
from myapp.thermostat import ThermostatService


def test_setpoint_command_updates_state(mqtt_probe, prefix):
    service = ThermostatService(host=HOST, port=PORT, prefix=prefix)
    service.start()  # returns once the service has its own SUBACK
    try:
        mqtt_probe.subscribe(f"{prefix}/state")
        mqtt_probe.publish(f"{prefix}/cmd/setpoint", json.dumps({"value": 21.5}))

        msg = mqtt_probe.expect(f"{prefix}/state")
        assert json.loads(msg.payload)["setpoint"] == 21.5
    finally:
        service.stop()

prefix uygulamaya aktarılır ve bu, test edilebilirlik için en faydalı tasarım kararıdır: servisiniz topic'lerini yapılandırılabilir bir kökten oluşturuyorsa her test kendi ad alanında çalışabilir. service.start() çağrısının servis abone olana kadar beklediğine dikkat edin; uygulamanız böyle bir hazır olma sinyali sunmuyorsa bir tane ekleyin (bir olay, bir callback veya retained bir online durum mesajı). Aksi halde testin komutu servis dinlemeye başlamadan önce gelebilir. MQTT için pytest fixture'ları oturum kapsamlı broker'lar dahil daha fazla fixture çeşidi gösteriyor.

Jest: MQTT.js 5 ile promise yardımcıları

MQTT.js 5; testleri doğrusal hale getiren connectAsync, subscribeAsync, publishAsync ve endAsync ile gelir. İhtiyacınız olan tek yardımcı "bir topic'teki bir sonraki mesajı beklemek"tir:

const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

const URL = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const opts = () => ({
  clientId: `it-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  reconnectPeriod: 0,
});

function nextMessage(client, topic, timeoutMs = 5000) {
  return new Promise((resolve, reject) => {
    const onMessage = (t, payload, packet) => {
      if (t !== topic) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ payload: payload.toString(), retain: packet.retain });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on ${topic} within ${timeoutMs} ms`));
    }, timeoutMs);
    client.on('message', onMessage);
  });
}

let publisher;
let prefix;

beforeAll(async () => {
  publisher = await mqtt.connectAsync(URL, opts());
});
afterAll(() => publisher.endAsync());
beforeEach(() => {
  prefix = `testmqtt/it/${process.env.GITHUB_RUN_ID ?? 'local'}/${randomUUID().slice(0, 8)}`;
});

test('late subscribers receive the retained device config', async () => {
  const topic = `${prefix}/config`;
  await publisher.publishAsync(topic, JSON.stringify({ interval: 30 }), { qos: 1, retain: true });

  const device = await mqtt.connectAsync(URL, opts());
  try {
    const received = nextMessage(device, topic); // listen before subscribing
    await device.subscribeAsync(topic, { qos: 1 });
    const msg = await received;
    expect(msg.retain).toBe(true);
    expect(JSON.parse(msg.payload)).toEqual({ interval: 30 });
  } finally {
    await device.endAsync();
    await publisher.publishAsync(topic, '', { qos: 1, retain: true }); // clear retained
  }
}, 15000);

İki ayrıntı en yaygın Jest sorunlarını önler. Listener subscribeAsync'ten önce kaydedilir, çünkü broker retained mesajları SUBACK'ten hemen sonra gönderir; bazen beklenen promise çözülmeden önce. Ve her istemci endAsync() ile kapatılır; böylece Jest "Jest did not exit one second after the test run has completed" uyarısıyla askıda kalmaz. Daha fazla Jest kalıbı için Jest ile MQTT testi yazısına bakın.

Hata yollarını da test edin

Mutlu yol, bir MQTT istemcisinin en az ilginç kısmıdır. Entegrasyon testleri, sahada önemli olan davranışı ucuza kanıtlayabileceğiniz yerdir:

• Bozuk payload'lar. Komut topic'inize geçersiz JSON publish edin ve servisin ağ döngüsünü çökertmek yerine bir hata loglayıp çalışmaya devam ettiğini doğrulayın.
• Yetkilendirme. Bir topic'e publish etme izni olmayan bir kullanıcıyla bağlanın ve kodunuzun reddi raporladığını doğrulayın. MQTT 5'te QoS 1 bir publish, PUBACK içinde bir reason code döndürür; MQTT 3.1.1'de broker genellikle mesajı düşürür veya bağlantıyı kapatır.
• Yeniden bağlanmalar. Uygulamanın istemcisinin bağlantısını kesin (veya yerel bir broker'ı yeniden başlatın) ve yeniden abone olup bir sonraki komutu işlediğini doğrulayın. Bunu nasıl betikleyeceğinizi MQTT yeniden bağlanma testi gösteriyor.

Bu durumların her biri yukarıdaki fixture'larla birkaç satır tutar ve aksi halde yalnızca production loglarında ortaya çıkacak hataları yakalar.

Kararsız test kontrol listesi

Bir abonelik reddedilirse SUBACK'teki reason code nedenini söyler; 0x87 Not authorized genellikle test kullanıcısının ACL'sinin öneki kapsamadığı anlamına gelir. Herhangi bir kodu MQTT reason code arama aracıyla çözümleyin.

Testleri test etmek

Yeni bir test paketine güvenmeden önce onu dışarıdan izleyin. Online MQTT istemcisinde testmqtt/it/# topic'ine abone olun ve testleri çalıştırın: her testin önekinin belirdiğini, mesajlarının aktığını ve sonunda retained topic'lerin temizlendiğini (boş bir retained mesaj) görmelisiniz. Ardından test paketini paralel olarak iki kez çalıştırın — örneğin pytest-xdist ile pytest -n 4 veya iki CI job'ı — ve hiçbir şeyin değişmediğini doğrulayın.

Entegrasyon testi trafiğini izleyin

Gerçek bir broker'a karşı yapılan entegrasyon testleri, MQTT kodunuzun production'daki gibi davrandığını kanıtlamanın en ucuz yoludur. Her testi topic ve client ID ile izole edin, uyumak yerine onayları bekleyin, retained durumu temizleyin; test paketi doğru nedenlerle yeşil kalır. Her push'ta çalıştırmaya hazır olduğunuzda GitHub Actions ile CI'da MQTT testi yazısıyla devam edin.

Sıkça sorulan sorular