MQTT ile Hardware-in-the-Loop Testi

Simülatörler ve mock broker'lar mantık hatalarını yakalar, ama TLS el sıkışması sırasında yaşanan bir voltaj düşüşünü ya da üçüncü yeniden bağlanmadan sonra pes eden bir Wi-Fi yığınını asla yeniden üretemez. MQTT ile hardware-in-the-loop (HIL) testi bu boşluğu kapatır: gerçek firmware çalıştıran gerçek kartlar bir test broker'ına bağlanır, bir PC'deki script onları yönetir, her mesajı dinler ve testin geçip geçmediğine karar verir.

Bu rehber küçük bir HIL düzeneğini nasıl kuracağınızı, pytest ve paho-mqtt 2.x ile nasıl otomatikleştireceğinizi, güç kesintisi gibi hataları nasıl enjekte edeceğinizi ve sonuçları CI'ınızın gösterebileceği raporlara nasıl dönüştüreceğinizi anlatıyor.

Bir MQTT HIL düzeneği neye benzer

Bir düzeneğin pahalı olması gerekmez. Birkaç kart, bir USB hub ve küçük bir Linux makinesi bulunan bir raf başlamak için yeterlidir.

Runner hem orkestra şefi hem gözlemcidir: cihazı açar, cihazın publish ettiği her şeye subscribe olur, bir backend'in yapacağı gibi komut gönderir, zamanlamayı ve içeriği kontrol eder.

Düzenek için broker seçimi

Düzeneğinizin ne kadar gerçekçi olacağını broker belirler. Runner üzerindeki yerel bir broker interneti denklemden çıkarır; bu da protokol mantığını deterministik biçimde test etmek için iyidir. Üretim ayarlarınıza sahip uzak bir broker ise DNS'i, TLS sertifikalarını, kimlik doğrulamayı ve gerçek ağ yolunu da test eder; sahadaki hatalar genellikle buradan çıkar. Pek çok ekip ikisini birden kullanır: hızlı firmware geri bildirimi için yerel bir broker, gece çalıştırmaları için uzak bir broker.

Hangisini seçerseniz seçin, bir şeyi otomatikleştirmeden önce online MQTT client ile bir kez elle bağlanın ve cihazın topic'lerine subscribe olun. Cihazın mesajlarını elle göremiyorsanız, hiçbir test scripti de göremez.

Her çalıştırmayı izole edin

Bir düzenek zaman içinde paylaşılır: dünkü çalıştırma retained mesajlar bırakmıştır, bir iş arkadaşınızın düzeneği benzer topic'lere publish eder ve iki CI job'ı aynı anda çalışabilir. Her düzeneğe ve her çalıştırmaya kendi topic kökünü verin, örneğin testmqtt/hil/<bench>/<run-id>/..., ve firmware'in topic önekini koda gömmek yerine yapılandırılabilir yapın (seri konsol, bir provisioning dosyası ya da bir build flag'i ile). Desenin tamamı testlerin asla çakışmaması için test topic'lerini adlandırma yazısında anlatılıyor.

Önek çalışma zamanında değiştirilemiyorsa, düzenek başına en az bir sabit önek kullanın ve her testten önce bu önek altındaki retained mesajları temizleyin.

Test altyapısı: bir MQTT kaydedici

Düzeneğin kalbi; cihazın önekine subscribe olan, her mesajın dökümünü tutan ve testlerin belirli bir mesajı timeout ile beklemesini sağlayan küçük bir sınıftır. sleep() yerine olayları beklemek, kararsız donanım testlerine karşı açık ara en etkili çözümdür.

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

class Recorder:
    def __init__(self, host, port, prefix):
        self.prefix = prefix
        self.inbox = queue.Queue()
        self.transcript = []
        self.ready = threading.Event()
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                  client_id=f"hil-runner-{uuid.uuid4().hex[:8]}")
        if os.getenv("MQTT_USER"):
            self.client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
        self.client.on_connect = self._on_connect
        self.client.on_subscribe = lambda c, u, mid, codes, props: self.ready.set()
        self.client.on_message = self._on_message
        self.client.connect(host, port, keepalive=30)
        self.client.loop_start()
        if not self.ready.wait(10):
            raise RuntimeError("runner could not subscribe to the test broker")

    def _on_connect(self, client, userdata, flags, reason_code, properties):
        client.subscribe(f"{self.prefix}/#", qos=1)

    def _on_message(self, client, userdata, msg):
        entry = (time.time(), msg.topic, msg.payload.decode(errors="replace"), msg.retain)
        self.transcript.append(entry)
        self.inbox.put(entry)

    def expect(self, suffix, timeout=10, match=lambda payload: True):
        want, deadline = f"{self.prefix}/{suffix}", time.monotonic() + timeout
        while (left := deadline - time.monotonic()) > 0:
            try:
                _, topic, payload, _ = self.inbox.get(timeout=left)
            except queue.Empty:
                break
            if topic == want and match(payload):
                return payload
        raise AssertionError(f"no matching message on {want} within {timeout}s")

    def publish(self, suffix, payload, qos=1):
        self.client.publish(f"{self.prefix}/{suffix}", payload, qos=qos).wait_for_publish(5)

    def drain(self):
        while not self.inbox.empty():
            self.inbox.get_nowait()

    def close(self):
        self.client.loop_stop()
        self.client.disconnect()

Güç ve seri port kontrolü

Her test soğuk açılışla başlamalıdır; bu yüzden düzeneğin gücü kesebilmesi gerekir. Port bazında güç anahtarlamayı destekleyen bir hub'a bağlı USB beslemeli kartlar için uhubctl yeterlidir; şebekeden beslenen cihazlarda bir röle kartı ya da anahtarlanabilir bir PDU aynı işi görür. Seri loglar arka plandaki bir thread'de kaydedilir, böylece başarısız bir test firmware'in tam olarak ne yazdırdığını rapora ekleyebilir.

import os, subprocess, threading, time
import pytest, serial
from hil.recorder import Recorder

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
BENCH = os.getenv("HIL_BENCH", "bench1")
RUN_ID = os.getenv("GITHUB_RUN_ID", time.strftime("%Y%m%dT%H%M%S"))

def set_power(on: bool):
    # Kendi röle veya PDU çağrınızla değiştirin. Örnek: hub 1-1'in 2. portu.
    subprocess.run(["uhubctl", "-l", "1-1", "-p", "2", "-a", "on" if on else "off"], check=True)

class Rig:
    def __init__(self, mqtt):
        self.mqtt = mqtt
    def power_off(self):
        set_power(False)
    def power_cycle(self, off_seconds=2):
        set_power(False); time.sleep(off_seconds)
        self.mqtt.drain(); set_power(True)

def capture(port, log, stop):
    while not stop.is_set():
        log.write(port.read(512))

@pytest.fixture
def rig(tmp_path):
    prefix = f"testmqtt/hil/{BENCH}/{RUN_ID}/dut"
    rec = Recorder(HOST, 1883, prefix)
    log = open(tmp_path / "serial.log", "wb")
    port = serial.Serial(os.getenv("HIL_SERIAL", "/dev/ttyUSB0"), 115200, timeout=0.2)
    stop = threading.Event()
    threading.Thread(target=capture, args=(port, log, stop), daemon=True).start()
    yield Rig(rec)
    stop.set(); time.sleep(0.3); port.close(); log.close()
    rec.close()

Bu örnekteki firmware bağlandıktan sonra <prefix>/status topic'ine online publish eder, aynı topic'te offline değerli bir Last Will tanımlar, <prefix>/cmd/# topic'ini dinler ve <prefix>/ack üzerinden yanıt verir.

Testleri yazın

import json

def test_boot_reports_online(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")

def test_command_round_trip(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")
    rig.mqtt.publish("cmd/led", json.dumps({"id": "t-1", "state": "on"}))
    ack = json.loads(rig.mqtt.expect("ack", timeout=5))
    assert ack["id"] == "t-1" and ack["ok"] is True

def test_power_loss_publishes_last_will(rig):
    rig.power_cycle()
    rig.mqtt.expect("status", timeout=30, match=lambda p: p == "online")
    rig.power_off()
    # keep alive 30 sn -> broker 1,5 x 30 sn sonra vazgeçer
    rig.mqtt.expect("status", timeout=60, match=lambda p: p == "offline")

Son test, hiçbir simülatörün gerçekten kanıtlayamadığı bir şeyi kanıtlar: ani bir güç kesintisi, durumu ancak keep alive süresi dolunca fark eden bir broker ve backend'e ulaşan bir Last Will. Broker'ın bu mesajı tam olarak ne zaman gönderdiği için Last Will testi yazısına bakın.

Hata enjekte edin

HIL'in değeri, isteğe bağlı tetikleyebildiğiniz her hata senaryosuyla artar:

• Publish ortasında güç kaybı: bir komuttan hemen sonra gücü kesin ve cihazın toparlandığını, eskimiş bir durum bildirmediğini kontrol edin.
• Broker yeniden başlatma: yerel broker'ı yeniden başlatın ve cihazın backoff ile yeniden bağlandığını, tekrar subscribe olduğunu ve publish etmeye devam ettiğini doğrulayın. Senaryolar MQTT yeniden bağlanma testi yazısında ele alınıyor.
• Güç kaybı olmadan ağ kaybı: Wi-Fi erişim noktasını ya da switch portunu kapatın. Cihaz çalışmaya devam eder, TCP bağlantısı sessizce ölür ve keep alive ile yeniden bağlanma mantığınızın gerçekten çalışıp çalışmadığını görürsünüz.
• Hatalı kimlik bilgileri veya iptal edilmiş ACL: test broker'ında cihazın parolasını değiştirin ve firmware'in sessizce döngüye girmek yerine CONNACK nedenini logladığını doğrulayın. Reason code aracı, cihazın bildirdiği kodu çözmenize yardımcı olur.

Raporlama

Her CI sisteminin test bazında sonucu gösterebilmesi için paketi pytest --junitxml=report.xml ile çalıştırın. Bir test başarısız olduğunda seri logu ve kaydedicinin MQTT dökümünü ekleyin: ikisi birlikte cihazın mesajı hiç göndermediğini mi, yanlış topic'e mi gönderdiğini yoksa broker'ın mı reddettiğini gösterir. Düzeneğe bağlı self-hosted bir CI runner, her firmware merge'ünün bir donanım çalıştırması başlatmasını sağlar; workflow tarafı GitHub Actions'ta MQTT testi ile aynıdır.

İyi bir HIL raporu şunları içerir:

1. Test edilen cihazın firmware sürümü ve git commit'i.
2. Düzenek adı, broker endpoint'i ve çalıştırma ID'si (topic öneki).
3. Her testin sonucu ve süresi, ayrıca ölçülen zamanlamalar.
4. Başarısız her test için seri log ve MQTT dökümü.

Düzeneği güvenilir tutun

• Testin her zaman düşündüğünüz build üzerinde çalışması için her çalıştırmanın başında cihazı yeniden flash'layın.
• Çalıştırmadan önce düzenek öneki altındaki retained mesajları temizleyin; aksi halde eski bir online mesajı bozuk bir açılışı sorunsuz gösterebilir.
• Çalıştığını bildiğiniz bir firmware imajı saklayın; o da başarısız olmaya başlarsa sorun kodda değil düzenektedir.
• Kararsız testleri körü körüne tekrar çalıştırmak yerine karantinaya alın ve nedenini düzeltin; bu genellikle eksik bir bekleme ya da paylaşılan bir topic'tir.

Düzeneğinizin trafiğini tarayıcıda izleyin

Hardware-in-the-loop testi, "benim masamda çalışıyordu" cümlesini her build'de çalışan tekrarlanabilir bir kontrole dönüştürür. Bir kart, bir güç anahtarı ve üç testle (açılış, komut gidiş-dönüşü ve güç kaybı) başlayın; saha size yeni hatalar öğrettikçe hata kataloğunu genişletin.

Sıkça sorulan sorular