MQTT Client Test Etme: Birim, Entegrasyon ve E2E Testleri

MQTT client'larını test etmek zahmetlidir, çünkü ilginç davranışlar ağ üzerinden, callback'lerin içinde ve başka birinin thread'inde gerçekleşir. Bu rehber işe yarayan katmanlı bir yaklaşımı gösteriyor: mock'lanmış bir client ile hızlı birim testleri, CI'da tek kullanımlık bir broker container'ına karşı entegrasyon testleri ve sürümden önce gerçek bir bulut broker'a karşı kısa bir uçtan uca duman testi.

MQTT Client Kodunu Test Etmek Neden Zor?

MQTT kodu, farklı şekillerde başarısız olan üç kaygıyı bir araya getirir:

İş mantığı: payload'ları ayrıştırmak, neyin publish edileceğine karar vermek, topic'leri cihazlarla eşlemek.
Protokol davranışı: QoS, retained mesajlar, yeniden bağlanma, yeniden subscribe olma, last will.
Ortam: TLS sertifikaları, kimlik bilgileri, portlar, firewall'lar, broker kotaları.

Kararsız (flaky) MQTT test paketlerinin çoğu, bu üçünü bir broker'a bağlanıp bekleyen tek bir testte kapsamaya çalışır. Bunları ayırmak, her katmanın kendi hatalarını yakalayabilecek en ucuz aracı kullanmasını sağlar.

Katman	Neyi yakalar	Broker gerekir mi	Hız	Ne zaman çalışır
Birim	Ayrıştırma, topic eşleme, karar mantığı	Hayır (mock)	Milisaniyeler	Her kayıtta
Entegrasyon	Subscribe/publish bağlantıları, QoS, retained, yeniden bağlanma	Yerel container	Saniyeler	Her pull request'te
Uçtan uca duman testi	TLS, kimlik bilgileri, portlar, gerçek ağ	Gerçek bulut broker	Saniyeler	Sürümden önce, zamanlanmış olarak

Örnekler paho-mqtt 2.x ve pytest ile Python kullanır, ancak yapı her dile taşınabilir.

Adım 1: Test Edilebilir Kod Tasarlayın

En faydalı tek değişiklik, MQTT client'ını kenarda tutmaktır. Kararları, bağımlılık olarak bir publish fonksiyonu alan düz fonksiyonlara veya sınıflara koyun ve bunları paho'ya ince bir adaptörün bağlamasına izin verin.

thermostat.py:

import json
from dataclasses import dataclass
from typing import Callable

Publish = Callable[[str, str, int, bool], None]


@dataclass
class Thermostat:
    prefix: str
    publish: Publish
    setpoint: float = 21.0
    hysteresis: float = 0.5
    heating: bool = False

    def on_temperature(self, topic: str, payload: bytes) -> None:
        try:
            value = float(json.loads(payload)["value"])
        except (ValueError, KeyError, TypeError, json.JSONDecodeError):
            return  # bozuk mesajları yok say

        room = topic.split("/")[-2]
        if not self.heating and value < self.setpoint - self.hysteresis:
            self.heating = True
        elif self.heating and value > self.setpoint + self.hysteresis:
            self.heating = False
        else:
            return

        self.publish(f"{self.prefix}/{room}/heater/set",
                     "ON" if self.heating else "OFF", 1, True)

adapter.py:

import threading

import paho.mqtt.client as mqtt
from thermostat import Thermostat


def build_client(host, port, client_id, prefix, username=None, password=None, tls=False):
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    if username:
        client.username_pw_set(username, password)
    if tls:
        client.tls_set()

    thermostat = Thermostat(
        prefix=prefix,
        publish=lambda t, p, q, r: client.publish(t, p, qos=q, retain=r),
    )

    def on_connect(client, userdata, flags, reason_code, properties):
        if not reason_code.is_failure:
            client.subscribe(f"{prefix}/+/temperature", qos=1)

    ready = threading.Event()

    def on_subscribe(client, userdata, mid, reason_codes, properties):
        ready.set()

    def on_message(client, userdata, msg):
        thermostat.on_temperature(msg.topic, msg.payload)

    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    return client, thermostat, ready

on_connect içinde subscribe olmak bilinçli bir tercihtir: her yeniden bağlanmadan sonra subscription'ları geri yükler; döndürülen ready olayı da testlerin sleep() ile tahmin yürütmek yerine subscription etkin olana kadar beklemesini sağlar. Entegrasyon testlerinin doğrulayacağı bağlantı mantığı tam olarak budur.

Adım 2: Mock'lanmış Client ile Birim Testleri

Thermostat, publish'i bir parametre olarak aldığı için bir Mock ağı tamamen ortadan kaldırır.

test_thermostat.py:

import json
from unittest.mock import Mock

from thermostat import Thermostat

PREFIX = "home/test"


def make():
    pub = Mock()
    return Thermostat(prefix=PREFIX, publish=pub), pub


def msg(value):
    return json.dumps({"value": value}).encode()


def test_turns_heater_on_below_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    pub.assert_called_once_with(f"{PREFIX}/kitchen/heater/set", "ON", 1, True)


def test_no_publish_inside_hysteresis_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.2))
    pub.assert_not_called()


def test_turns_heater_off_above_band():
    t, pub = make()
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(20.0))
    t.on_temperature(f"{PREFIX}/kitchen/temperature", msg(21.6))
    assert pub.call_args.args[1] == "OFF"


def test_ignores_garbage():
    t, pub = make()
    for bad in [b"", b"not json", b'{"temp": 3}', b'{"value": "hot"}']:
        t.on_temperature(f"{PREFIX}/kitchen/temperature", bad)
    pub.assert_not_called()

Bunlar milisaniyeler içinde çalışır ve en sık hata barındıran mantığı kapsar. paho'nun kendisini ayrıntılı biçimde mock'lama isteğine direnin (client.subscribe'ın tam olarak hangi argümanlarla çağrıldığını doğrulamak gibi): bu tür testler uygulamayı aynen yansıtır, yeniden düzenlemelerde bozulur ve yine de broker'ın gönderdiğiniz şeyi kabul ettiğini kanıtlamaz. Bu, entegrasyon katmanının işidir.

Adım 3: Broker Container'ına Karşı Entegrasyon Testleri

Container içindeki gerçek bir broker, paylaşılan durum ve internet bağımlılığı olmadan dürüst bir protokol davranışı sunar. Resmi Eclipse Mosquitto imajı, anonim bağlantılara izin veren bir yapılandırma dosyası içerir; bu da testler için idealdir:

docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
  mosquitto -c /mosquitto-no-auth.conf

Şimdi gerçek adaptörü çalıştıran testler yazın. Bir yardımcı, ikinci bir client üzerinde mesajları toplar ve sabit bir süre beklemek yerine bir koşulu bekler.

test_integration.py:

import os
import queue
import uuid

import paho.mqtt.client as mqtt
import pytest

from adapter import build_client

HOST = os.getenv("MQTT_TEST_HOST", "localhost")
PORT = int(os.getenv("MQTT_TEST_PORT", "1883"))


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


@pytest.fixture
def probe(prefix):
    received = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"probe-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: received.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/#", qos=1)
    c.loop_start()
    yield c, received
    c.loop_stop()
    c.disconnect()


def wait_for(q, predicate, timeout=5.0):
    while True:
        topic, payload, retain = q.get(timeout=timeout)  # zaman aşımında queue.Empty fırlatır
        if predicate(topic, payload):
            return topic, payload, retain


def test_temperature_drives_heater(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/office/temperature", '{"value": 18.5}', qos=1)
        topic, payload, _ = wait_for(received, lambda t, p: t.endswith("/heater/set"))
        assert topic == f"{prefix}/office/heater/set"
        assert payload == "ON"
    finally:
        client.loop_stop()
        client.disconnect()


def test_heater_command_is_retained(prefix, probe):
    pc, received = probe
    client, _, ready = build_client(HOST, PORT, f"sut-{uuid.uuid4().hex[:8]}", prefix)
    client.loop_start()
    try:
        assert ready.wait(5), "SUT did not subscribe in time"
        pc.publish(f"{prefix}/hall/temperature", '{"value": 15}', qos=1)
        wait_for(received, lambda t, p: t.endswith("/heater/set"))
    finally:
        client.loop_stop()
        client.disconnect()

    # Geç gelen bir subscriber retained komutu hemen almalı
    late = queue.Queue()
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"late-{uuid.uuid4().hex[:8]}")
    c.on_message = lambda cl, ud, m: late.put((m.topic, m.payload.decode(), m.retain))
    c.connect(HOST, PORT)
    c.subscribe(f"{prefix}/hall/heater/set", qos=1)
    c.loop_start()
    try:
        _, payload, retain = late.get(timeout=5)
        assert payload == "ON" and retain is True
    finally:
        c.loop_stop()
        c.disconnect()

Entegrasyon testlerini kararlı tutan üç alışkanlık:

Her test için benzersiz topic öneki. Bir testteki retained mesajlar bir sonrakine sızamaz.
sleep() yerine koşulları bekleyin. Zaman aşımlı bir kuyruk hızlı başarısız olur ve asla gereğinden uzun beklemez.
Döngüleri her zaman finally içinde durdurun ve bağlantıyı kesin. Sızan ağ thread'leri sonraki testlerin askıda kalmasına yol açar.

Yeniden bağlanma mantığını test etmek için testin ortasında container'ı durdurup başlatın (docker restart test-broker) ve client'ın yeniden subscribe olduğunu ve yeniden başlatmadan sonra publish edilen bir mesajı aldığını doğrulayın.

Broker'ı CI'da Çalıştırma

GitHub Actions, container'ı testlerden önce bir adım olarak başlatabilir:

name: tests
on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Start MQTT broker
        run: |
          docker run -d --name test-broker -p 1883:1883 eclipse-mosquitto:2 \
            mosquitto -c /mosquitto-no-auth.conf
      - run: pip install "paho-mqtt>=2,<3" pytest
      - run: pytest -q

Birim ve entegrasyon testleri tek bir pytest çalıştırmasını paylaşabilir; yavaş olanları yerelde atlamak istiyorsanız @pytest.mark.integration ile işaretleyin.

Adım 4: Gerçek Bir Broker'a Karşı Uçtan Uca Duman Testleri

Bir container, TLS ayarlarınızın, CA paketinizin, firewall kurallarınızın veya üretim kimlik bilgilerinizin çalıştığını söyleyemez. Gerçek uç noktaya karşı kısa bir duman testi söyleyebilir. Açık kaynak projeler veya hızlı kontroller için broker.connectmq.com adresindeki public MQTT test broker'ı yeterlidir; 8883 portunda TLS üzerinden anonim client'ları kabul eder. Public alan herkes tarafından okunabildiği için payload'ları yapay tutun.

smoke_test.py:

import os
import sys
import threading
import uuid

import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
USER = os.getenv("MQTT_USERNAME")
PASSWORD = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"smoke-{uuid.uuid4().hex[:8]}"
TOPIC = f"{os.getenv('MQTT_PREFIX', 'connectmq/demo')}/smoke-{uuid.uuid4().hex[:8]}/ping"

got = threading.Event()
nonce = uuid.uuid4().hex
errors = []


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        errors.append(f"connect failed: {reason_code}")
        got.set()  # ana thread'i uyandır; callback'ler süreci sonlandıramaz
        return
    client.subscribe(TOPIC, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    client.publish(TOPIC, nonce, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == nonce:
        got.set()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                     protocol=mqtt.MQTTv5)
if USER:
    client.username_pw_set(USER, PASSWORD)
client.tls_set()
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
client.connect(HOST, 8883, keepalive=30)
client.loop_start()
ok = got.wait(timeout=15) and not errors
for e in errors:
    print(e)
client.loop_stop()
client.disconnect()
print("smoke test passed" if ok else "smoke test FAILED: no echo within 15 s")
sys.exit(0 if ok else 1)

Üretim için MQTT_USERNAME, MQTT_PASSWORD, MQTT_CLIENT_ID ve MQTT_PREFIX değerlerini CI secret'larından ayarlayarak aynı betiği özel broker'ınıza yöneltin. ConnectMQ gibi tam yönetilen bir MQTT servisinde client ID'ler verilen kimlik bilgilerine bağlıdır; bu yüzden bir cihazın kimliğini ödünç almak yerine CI için ayrı bir kimlik bilgisi seti oluşturun; aksi halde duman testi o cihazın bağlantısını keser.

Duman testlerini her commit'te değil, sürümlerden önce ve bir zamanlamaya göre (örneğin saatlik) çalıştırın. Paylaşımlı broker'larda kotalar geçerlidir: public broker client başına saniyede 10 mesaja ve client başına 10 subscription'a izin verir; ücretsiz ConnectMQ Serverless planı ise eşzamanlı bağlantıları 25 ile sınırlar.

Yaygın Anti-Desenler

Bazı alışkanlıklar MQTT test paketlerini yavaş ve kararsız hale getirir. Kod incelemesinde bunlara dikkat edin:

Sabit beklemeler. Publish'ten sonraki time.sleep(2) ya çok uzundur ve paket genelinde dakikalar harcatır ya da yoğun bir CI runner'ında çok kısadır. Bunun yerine zaman aşımlı bir kuyruk veya olay bekleyin.
Testler arasında paylaşılan topic'ler. Bir testten kalan retained mesajlar ve subscription'lar bir sonrakine sızar. Her test için benzersiz bir önek üretin.
Yutulan callback hataları. MQTT callback'leri içindeki exception'lar ağ thread'inde çalışır, bu yüzden testi doğrudan başarısız kılmaz. Bunları bir listede toplayın ve teardown sırasında listenin boş olduğunu doğrulayın.
CI'da cihaz kimlik bilgileri. Testler için gerçek bir cihazın kimliğini ödünç almak o cihazın bağlantısını keser. Otomasyon için ayrı kimlik bilgileri oluşturun.
Kendi kodunuz yerine kütüphaneyi test etmek. paho'nun QoS 1 mesajlarını teslim ettiğini kanıtlamanıza gerek yok; kodunuzun doğru topic'lere subscribe olduğunu ve doğru tepki verdiğini kanıtlamanız gerekir.
Tek bir dev uçtan uca test. Başarısız olduğunda hangi katmanın bozulduğunu kimse bilmez. Uçtan uca testleri az ve dar tutun, ayrıntıları birim testlerine indirin.

Test Kontrol Listesi

İş mantığı MQTT callback'lerinin dışında yaşar ve mock bir publish ile birim testine tabi tutulur.
Bozuk, boş ve aşırı büyük payload'lar test edilir ve güvenle yok sayılır.
Entegrasyon testleri taze bir broker container'ı ve test başına benzersiz bir topic öneki kullanır.
Yeniden bağlanma ve yeniden subscribe olma davranışı broker yeniden başlatılarak test edilir.
Retained mesajlar ve last will davranışı geç gelen bir subscriber ile doğrulanır.
Gerçek kimlik bilgileriyle bir TLS duman testi her sürümden önce çalışır.
CI kimlik bilgileri cihaz kimlik bilgilerinden ayrıdır.

Dile özgü ayrıntılar için PHP MQTT client eğitimi, Paho MQTT C client eğitimi ve daha kapsamlı MQTT client kütüphaneleri rehberine bakın.

ConnectMQ ile Başlayın

Test pipeline'ınıza konuşabileceği gerçek bir broker verin. app.connectmq.com üzerinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun, CI'a özel bir kimlik bilgisi seti ekleyin ve her sürümde duman testini ona karşı çalıştırın. Portalın canlı topic explorer'ı ve tarayıcı içi client'ı, testlerinizin tam olarak neyi publish ettiğini görmeyi kolaylaştırır. Broker'ı ConnectMQ çalıştırır, testler sizde kalır.

Sık Sorulan Sorular

Birim testlerinde MQTT kütüphanesini mock'lamalı mıyım?

Kütüphanenin iç yapısını mock'lamak yerine, enjekte ettiğiniz bir publish fonksiyonu gibi kendi sınırınızda mock'layın. Gerçek protokol davranışını bir broker container'ına karşı entegrasyon testiyle doğrulayın.

CI'daki MQTT entegrasyon testleri için hangi broker'ı kullanmalıyım?

Tek kullanımlık bir Eclipse Mosquitto container'ı en basit seçimdir: yaklaşık bir saniyede başlar, /mosquitto-no-auth.conf ile kimlik bilgisi gerektirmez ve çalıştırmalar arasında geride durum bırakmaz.

broker.connectmq.com'u otomatik testlerde kullanabilir miyim?

Yapay veri ve benzersiz topic'lerle ara sıra yapılan duman testleri için evet; client başına saniyede 10 mesaj ve 10 subscription limitleri dahilinde. Yoğun test paketleri için yerel bir container veya kendi özel broker'ınızı kullanın.

MQTT yeniden bağlanma mantığını nasıl test ederim?

Test sırasında broker container'ını yeniden başlatın ve client'ın yeniden bağlandığını, yeniden subscribe olduğunu ve yeniden başlatmadan sonra publish edilen bir mesajı aldığını doğrulayın.

MQTT Client'larını Test Etme: Birim, Entegrasyon ve Uçtan Uca Testler