MQTT ile IoT Uçtan Uca Test: Cihazdan Arayüze

Birim testleri her servisin çalıştığını kanıtlar. Müşterilerinizi ise bir sensörün gönderdiği ölçümün dashboard'da görünüp görünmediği ve bastıkları butonun cihaza ulaşıp ulaşmadığı ilgilendirir. MQTT üzerinden IoT sistemleri için uçtan uca test tam olarak bunu kontrol eder: simüle edilmiş bir cihaz gerçek broker üzerinden publish eder ve test, ingestion servisi ile veritabanından API ve UI'a kadar her adımı doğrular; ardından geri bir komut gönderir ve cihazın onu aldığını kontrol eder.

Bu rehber bu testi çalıştırılabilir pytest ve Playwright koduyla nasıl yapılandıracağınızı ve her merge'de çalışacak kadar hızlı ve güvenilir tutmayı gösteriyor.

Bir olayın izlediği yol

MQTT tabanlı tipik bir IoT sisteminin iki yönü vardır ve uçtan uca bir test paketi ikisini de kapsamalıdır:

Yön	Adımlar	Neler bozulabilir
Uplink (telemetri)	Cihaz → broker → ingestion servisi → veritabanı → API → UI	Topic veya payload değişiklikleri, ACL'ler, parse, saat dilimleri, cache
Downlink (komutlar)	UI → API → backend → broker → cihaz → ack	Yanlış topic, eksik izinler, QoS, çevrimdışı cihazlar
Varlık (presence)	Cihaz bağlantısının kopması → Last Will → backend → UI'daki durum	Keep alive ayarları, LWT topic'i, eskimiş durum

Her adımın zaten kendi testleri vardır (broker sınırı için MQTT entegrasyon testleri yazısına bakın). Uçtan uca test paketi küçüktür, birkaç senaryodan ibarettir; ama her biri tüm sınırları aynı anda geçer.

Firmware gibi davranan bir simülatör yazın

Simülatör en önemli parçadır. Gerçek firmware ile aynı client ID şemasını, topic'leri, QoS'u, payload formatını ve Last Will'i kullanmalıdır; aksi hâlde başka bir sistemi test etmiş olursunuz. paho-mqtt 2.x ile minimal bir Python sürümü:

sim_device.py

import json, queue, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

class SimDevice:
    def __init__(self, device_id, host, user, password, prefix="testmqtt/e2e"):
        self.id = device_id
        self.base = f"{prefix}/{device_id}"
        self.commands = queue.Queue()
        self.subscribed = queue.Queue()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=device_id,
                                  protocol=mqtt.MQTTv5)
        self.client.username_pw_set(user, password)
        self.client.tls_set()
        self.client.will_set(f"{self.base}/status", "offline", qos=1, retain=True)
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, rc, p: self.subscribed.put(mid)
        self.client.on_message = self._on_message
        self.host = host
        self.crashed = False

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.publish(f"{self.base}/status", "online", qos=1, retain=True)
        client.subscribe(f"{self.base}/cmd", qos=1)

    def _on_message(self, client, userdata, msg):
        cmd = json.loads(msg.payload)
        self.commands.put(cmd)
        client.publish(f"{self.base}/cmd/ack", json.dumps({"id": cmd["id"], "ok": True}), qos=1)

    def start(self):
        self.client.connect(self.host, 8883, keepalive=30)
        self.client.loop_start()
        self.subscribed.get(timeout=10)        # SUBACK alındı: komutlara hazır

    def send_reading(self, temp_c):
        payload = {"schema": "telemetry/v1", "device": self.id, "temp_c": temp_c,
                   "ts": datetime.now(timezone.utc).isoformat()}
        self.client.publish(f"{self.base}/telemetry", json.dumps(payload), qos=1).wait_for_publish()

    def crash(self):
        """TCP bağlantısını DISCONNECT olmadan kes; broker Last Will mesajını göndersin."""
        self.crashed = True
        self.client.loop_stop()
        self.client.socket().close()

    def stop(self):
        if not self.crashed:
            # broker temiz kalsın diye retained durumu temizle
            self.client.publish(f"{self.base}/status", "", qos=1, retain=True).wait_for_publish()
            self.client.disconnect()
        self.client.loop_stop()

Zaten bir filo simülatörünüz varsa onu yeniden kullanın; aynı fikri çok sayıda cihaza ölçeklemek için MQTT cihaz simülatörü oluşturma yazısına bakın. Firmware ile simülatör arasında payload'ın ayrışmasını önlemek için ortak bir JSON Schema ve sözleşme testleri ikisini de hizalı tutar.

Sleep ile değil, koşulla bekleyin

Asenkron pipeline'ların gecikmesi değişkendir. Sabit bir time.sleep(2), yoğun bir CI runner'ında ya çok kısa kalır ya da diğer her durumda zaman kaybettirir. Bir koşulu, sağlanana ya da süre dolana kadar yoklayın:

eventually.py

import time

def eventually(check, timeout=15.0, interval=0.25):
    """check() AssertionError fırlatmayı bırakana kadar çağır, timeout sonrası yeniden fırlat."""
    deadline = time.monotonic() + timeout
    while True:
        try:
            return check()
        except AssertionError:
            if time.monotonic() > deadline:
                raise
            time.sleep(interval)

Uplink testi: cihazdan veritabanına ve API'a

Her çalıştırma benzersiz bir cihaz ID'si alır; böylece paralel çalıştırmalar ve önceki çalıştırmalardan kalanlar asla birbirine karışmaz:

test_e2e_uplink.py

import os, uuid
import httpx, psycopg, pytest
from eventually import eventually
from sim_device import SimDevice

HOST = "broker.connectmq.com"
API = os.environ.get("API_URL", "http://localhost:8000")
DSN = os.environ.get("DATABASE_URL", "postgresql://app:app@localhost:5432/app")

@pytest.fixture
def device():
    dev = SimDevice(f"e2e-{uuid.uuid4().hex[:8]}", HOST,
                    os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    dev.start()
    yield dev
    dev.stop()
    with psycopg.connect(DSN) as db:
        db.execute("DELETE FROM readings WHERE device_id = %s", (dev.id,))

def test_reading_reaches_database_and_api(device):
    device.send_reading(21.5)

    def stored():
        with psycopg.connect(DSN) as db:
            row = db.execute(
                "SELECT temp_c FROM readings WHERE device_id = %s ORDER BY ts DESC LIMIT 1",
                (device.id,),
            ).fetchone()
        assert row is not None, "reading not stored yet"
        assert row[0] == pytest.approx(21.5)
    eventually(stored)

    def served():
        r = httpx.get(f"{API}/devices/{device.id}/latest", timeout=5)
        assert r.status_code == 200
        assert r.json()["temp_c"] == pytest.approx(21.5)
    eventually(served)

Veritabanı kontrolü ingestion'ın çalıştığını, API kontrolü ise sorguların, serileştirmenin ve cache'in bunun üzerinde çalıştığını gösterir. Test başarısız olduğunda ilk başarısız assertion bozuk adımın adını verir; yalnızca UI yerine her katmanı doğrulamanın asıl değeri budur. MQTT'den PostgreSQL'e rehberi buna uygun bir ingestion servisi gösterir.

Playwright ile UI ve downlink testi

Tarayıcı testi kullanıcının gördüğünü kontrol eder ve komut yolunu cihaza kadar yürütür. Python için Playwright ile:

test_e2e_ui.py

import os
from playwright.sync_api import Page, expect
from eventually import eventually

APP = os.environ.get("APP_URL", "http://localhost:3000")

def test_dashboard_shows_reading_and_sends_command(page: Page, device):
    device.send_reading(23.0)

    page.goto(f"{APP}/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)
    expect(page.get_by_test_id("temp-value")).to_have_text("23.0 °C", timeout=15_000)

    page.get_by_role("button", name="Reboot").click()

    cmd = device.commands.get(timeout=15)        # simülatör komutu MQTT üzerinden aldı
    assert cmd["action"] == "reboot"
    expect(page.get_by_test_id("last-command")).to_contain_text("acknowledged", timeout=15_000)

Bu test device fixture'ını yeniden kullanır (onu conftest.py dosyasına taşıyın). Playwright'ın expect'i zaten timeout'a kadar yeniden dener; bu yüzden tarayıcıda ek bir yoklamaya gerek yoktur. CSS sınıfları yerine data-testid attribute'ları kullanın; yeniden tasarımlardan sağ çıkarlar.

Varlık: Last Will yolunu test edin

Çevrimdışı cihaz tespiti, kullanıcıların güvendiği ama nadiren test edilen bir özelliktir. Temiz bir bağlantı kesme değil, bir çökme simüle edin; çünkü normal bir DISCONNECT, Last Will'i bastırır:

test_e2e_presence.py

from playwright.sync_api import Page, expect

def test_crashed_device_shows_offline(page: Page, device):
    page.goto(f"http://localhost:3000/devices/{device.id}")
    expect(page.get_by_test_id("device-status")).to_have_text("online", timeout=15_000)

    device.crash()

    # kapanan socket Will'i hemen tetikler; sessizce kopan bağlantı 1,5 x keepalive'a kadar sürer
    expect(page.get_by_test_id("device-status")).to_have_text("offline", timeout=60_000)

Kapatılan bir TCP socket'i Will'i hemen tetikler; ancak gücünü veya kapsama alanını kaybeden bir cihaz hiçbir şey göndermez ve broker bunu ancak keep alive aralığının yaklaşık 1,5 katı sonra fark eder; bu nedenle timeout'u bunun üzerinde tutun. Zamanlama kuralları için Last Will testi yazısına bakın. Yerel bir çalıştırma sırasında online MQTT istemcisinde testmqtt/e2e/# topic'ini izlemek durum, telemetri, komut ve ack mesajlarını sırayla gösterir; başarısız bir senaryoyu debug etmenin en hızlı yolu budur.

Uçtan uca test trafiğini izleyin

Ortam ve CI

Backend'i, veritabanını ve UI'ı Docker Compose ile çalıştırın ve bunları, simülatör gibi, aynı broker'a yönlendirin:

Test paketini yerelde çalıştır

export MQTT_USER=username MQTT_PASS=password
docker compose up -d --wait          # api, ingest, postgres, web
pip install pytest pytest-playwright paho-mqtt httpx "psycopg[binary]"
playwright install chromium
pytest -q tests/e2e --maxfail=1

CI'da aynı komutlar bir job içinde çalışır; GitHub Actions'ta MQTT testleri yazısına bakın. Broker yapılandırmasını (TLS, ACL'ler, MQTT sürümü) production ile birebir aynı tutun; çünkü izin ve TLS hataları, uçtan uca testlerin yakalaması gereken şeylerin ta kendisidir.

Test paketini güvenilir tutmak

Her çalıştırmada benzersiz ID'ler. Cihaz ID'si, topic öneki ve veritabanı satırları tek bir rastgele sonekten türetilir.
SUBACK'i bekleyin. Simülatörün start() metodu komut aboneliği onaylanana kadar bekler; bu olmadan ilk komut, henüz kimse dinlemeden gelebilir.
Retained mesajları temizleyin. stop() içinde durum topic'ine boş bir retained payload publish etmek broker'ı düzenli tutar; bkz. retained mesajlar.
Az senaryo, tam derinlik. Her adımı geçen beş uçtan uca senaryo, birim testlerini tekrarlayan elli senaryodan daha değerlidir.
Adımın adıyla başarısız olun. "reading not stored yet" gibi assertion mesajları yazın ki rapor zincirin nerede koptuğunu söylesin.

Bir MQTT sistemi için uçtan uca test paketi; bir simülatör, bir yoklama yardımcısı ve bir olayı cihazdan ekrana ve geri takip eden birkaç testten oluşur. Dakikalar içinde çalışır, birim testlerinin göremediği entegrasyon hatalarını yakalar ve her sürümden önce "çalışıyor mu?" sorusuna somut bir yanıt verir.

Sıkça sorulan sorular

IoT'de entegrasyon testi ile uçtan uca test arasındaki fark nedir?

Entegrasyon testi tek bir sınırı kontrol eder; örneğin ingestion servisinizin broker'dan bir mesaj alıp almadığını. Uçtan uca test ise tek bir olayı tüm bileşenler boyunca takip eder: cihaz, broker, backend, veritabanı, API ve UI; ayrıca cihaza geri giden komut yolunu da.

IoT uçtan uca testleri için gerçek donanım gerekir mi?

Çoğu için gerekmez. Firmware ile aynı MQTT topic'lerini ve payload'larını kullanan bir yazılım cihaz simülatörü cloud tarafını güvenilir şekilde kapsar ve CI'da çalışır. Radyo, güç ve firmware'e özgü davranışlar için küçük bir hardware-in-the-loop test paketi tutun.

MQTT uçtan uca testlerinin kararsız (flaky) olmasını nasıl önlerim?

Her çalıştırmaya benzersiz bir cihaz ID'si ve topic öneki verin, sabit sleep yerine timeout'lu koşul beklemesi kullanın, publish etmeden önce SUBACK'i bekleyin ve testin oluşturduğu satırları ve retained mesajları temizleyin.

MQTT Üzerinden IoT Sistemleri için Uçtan Uca Test