Test için MQTT Trafiğini Kaydetme ve Tekrar Oynatma

MQTT trafiğini kaydedip tekrar oynatmak, bir broker'dan gelen gerçek mesajları — topic, payload, QoS, retain bayrağı ve her birinin geldiği an — bir dosyaya yakalamak ve daha sonra bu dosyayı aynı zamanlamayla yeniden yayınlamak demektir. Böylece regresyon testlerinizi, birinin tahmin ettiği verilerle değil, cihazlarınızın gerçekten gönderdiği verilerle kurarsınız.

Bu rehberde Python (paho-mqtt 2.x) ile küçük bir kaydedici ve zamanlaması hassas bir oynatıcı yazıyor, ardından bir kaydı otomatik bir regresyon testine dönüştürmeyi gösteriyoruz.

Tekrar oynatma neden sentetik veriden iyidir

Üretilmiş test verisi temizdir. Gerçek cihaz filoları değildir: farklı firmware sürümleri biraz farklı JSON gönderir, sensörler yeniden bağlandıktan sonra 50 ms içinde on ölçüm basar, bir gateway aynı retained durumu her dakika yeniden yayınlar. Hatalar tam da bu ayrıntılarda saklanır.

Sentetik verinin de yeri var — henüz kimsenin kaydetmediği uç durumlar için bir test verisi üreteci kullanın — ama bir kayıt, dizüstü bilgisayarda çalıştırabileceğiniz production'a en yakın şeydir.

Neyi kaydedeceğinizi seçin

Kaydedici, bir aboneliği olan sıradan bir MQTT istemcisidir. Test etmek istediğiniz akışı kapsayan bir topic filtresi seçin; tüm broker üzerinde # değil. Küçük kayıtlar incelemesi daha kolaydır ve daha az veri sızdırır. Başlamadan önce filtreyi topic matcher ile kontrol edin.

Önceden karar vermeniz gerekenler:

• Topic filtresi — örneğin bir sahanın telemetrisi için testmqtt/replay/site1/#.
• Abonelik QoS'u — QoS 2 ile abone olun. Teslim QoS'u, publish QoS'u ile abonelik QoS'unun küçük olanıdır; dolayısıyla QoS 2 abonelik orijinal seviyeyi korur.
• Payload kodlaması — payload'ları base64 olarak saklayın ki ikili formatlar (Protobuf, CBOR) gidiş-dönüşte bozulmasın.
• Gizlilik — gerçek payload'lar konum, isim veya token içerebilir. Bir kaydı repository'ye commit etmeden önce bunları temizleyin.

30 satırlık bir kaydedici

Kaydedici her satıra bir JSON nesnesi yazar (JSONL). t alanı kaydedicinin başlamasından bu yana geçen saniyedir; sistem saatindeki ayarlamalar zamanlamayı bozmasın diye monotonik bir saatle ölçülür.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
FILTER = sys.argv[1] if len(sys.argv) > 1 else "testmqtt/replay/#"
out = open(sys.argv[2] if len(sys.argv) > 2 else "capture.jsonl", "w", buffering=1)
start = time.monotonic()

def on_connect(client, userdata, flags, reason_code, properties):
    print("bağlandı:", reason_code)
    client.subscribe(FILTER, qos=2)        # her yeniden bağlanmada tekrar abone olur

def on_message(client, userdata, msg):
    out.write(json.dumps({
        "t": round(time.monotonic() - start, 6),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": msg.retain,
        "payload": base64.b64encode(msg.payload).decode(),
    }) + "\n")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-recorder-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, PORT, keepalive=30)
try:
    client.loop_forever()
except KeyboardInterrupt:
    client.disconnect()
    out.close()

python record.py 'testmqtt/replay/#' capture.jsonl ile çalıştırın, biraz trafik üretin ve Ctrl+C'ye basın. Satır tamponlama (buffering=1) sayesinde süreç zorla sonlandırılsa bile dosya kullanılabilir kalır.

Kayıttaki retained mesajlar

Kaydedici abone olduğu anda broker, eşleşen tüm retained mesajları retain: true ile hemen gönderir. Bunlar canlı trafik değil, kayıt anındaki durumun bir fotoğrafıdır ve hepsi t ≈ 0 civarında gelir. Canlı iletilen mesajlar, publisher bayrağı set etmiş olsa bile retain: false ile gelir (MQTT 5 istemcileri bunu Retain As Published abonelik seçeneğiyle değiştirebilir). Aşağıdaki oynatıcı bu fotoğrafı varsayılan olarak atlar; testiniz başlangıç durumuna ihtiyaç duyuyorsa onu ayrıca yayınlayın.

Zamanlamayı koruyan bir oynatıcı

Saf yaklaşım — mesajlar arasında sleep(gap) — kayar, çünkü her publish çağrısı da zaman alır. Bunun yerine her mesajın oynatma başlangıcına göre planlanan zamanını hesaplayın ve yalnızca o ana kadar bekleyin. Bir hız çarpanı, bir saatlik trafiği dakikalara sıkıştırmanızı sağlar.

import base64, json, os, sys, time
import paho.mqtt.client as mqtt

FILE = sys.argv[1]
SPEED = float(sys.argv[2]) if len(sys.argv) > 2 else 1.0     # 2.0 = iki kat hızlı
SRC, DST = "testmqtt/replay/", "testmqtt/replay-run/"          # topic yeniden yazma

records = [json.loads(line) for line in open(FILE) if line.strip()]
records = [r for r in records if not r["retain"]]              # retained fotoğrafı atla
if not records:
    sys.exit("oynatılacak mesaj yok")
offset = records[0]["t"]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"testmqtt-replayer-{os.getpid()}")
if os.environ.get("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
client.connect("broker.connectmq.com", 1883)
client.loop_start()

t0 = time.monotonic()
late = 0.0
for r in records:
    due = t0 + (r["t"] - offset) / SPEED
    delay = due - time.monotonic()
    if delay > 0:
        time.sleep(delay)
    else:
        late = max(late, -delay)
    topic = r["topic"].replace(SRC, DST, 1)
    info = client.publish(topic, base64.b64decode(r["payload"]), qos=r["qos"])

info.wait_for_publish(timeout=30)
print(f"{len(records)} mesaj oynatıldı, en kötü gecikme {late * 1000:.1f} ms")
client.loop_stop()
client.disconnect()

Topic yeniden yazma güvenlik ağıdır: kayıt testmqtt/replay/... altından geldi, oynatma testmqtt/replay-run/... altına gidiyor; böylece orijinal topic'leri dinleyen hiçbir şey tepki vermez. Test edilen sistemi yeniden yazılmış öneke yönlendirin.

En kötü gecikme satırı oynatmanın sadık olup olmadığını söyler. Yüksek hız çarpanlarında bu değer büyüyorsa darboğaz oynatıcının kendisidir: istemcinin inflight penceresi onaylanmamış QoS 1/2 mesajlarıyla doludur. Pencereyi büyütmek için istemci performans ayarı yazısına bakın.

Tekrar oynatmayı regresyon testine dönüştürün

Tekrar oynatma, test edilen sistemin ürettiği çıktıyı doğruluğu bilinen bir sonuçla karşılaştırdığınızda teste dönüşür. Desen şöyle:

1. Girdileri (cihaz telemetrisi) bir kez kaydedin ve temizlenmiş kaydı fixture olarak commit edin.
2. Kaydediciyi sistemin çıktı topic'lerinde çalıştırın: alarmlar, toplamlar, komutlar.
3. Girdileri oynatın.
4. Çıktı kaydını, güvenilir bir build'den alınmış golden dosyayla karşılaştırın.

Zamanlama çalıştırmalar arasında asla aynı olmaz; zaman damgalarını değil içeriği karşılaştırın. Topic'e göre gruplayın ve her topic için payload dizisini karşılaştırın. MQTT sıralamayı yalnızca topic ve publisher bazında garanti eder; global sıra karşılaştırması kararsız (flaky) testler üretir.

import base64, json, sys
from collections import defaultdict

def load(path):
    per_topic = defaultdict(list)
    for line in open(path):
        if line.strip():
            r = json.loads(line)
            per_topic[r["topic"]].append(base64.b64decode(r["payload"]))
    return per_topic

golden, actual = load(sys.argv[1]), load(sys.argv[2])
ok = True
for topic in sorted(set(golden) | set(actual)):
    if golden[topic] != actual[topic]:
        ok = False
        print(f"DIFF {topic}: beklenen {len(golden[topic])} mesaj, gelen {len(actual[topic])}")
print("PASS" if ok else "FAIL")
sys.exit(0 if ok else 1)

Payload'lar zaman damgası veya üretilmiş ID içeriyorsa karşılaştırmadan önce normalleştirin (JSON'u parse edip bu alanları çıkarın). Üç betiği bir pytest fixture'ına veya CI job'una bağlayın — entegrasyon testi rehberi ve GitHub Actions'ta MQTT testi her çalıştırmada nasıl broker başlatılacağını gösterir.

Dikkat edilmesi gereken tuzaklar

• Komutları oynatmak. .../cmd/... topic'lerini içeren bir kayıt gerçek bir cihaza ulaşırsa gerçek röleleri açıp kapatır. Test özellikle komutlarla ilgili değilse komut topic'lerini girdi kayıtlarından filtreleyin.
• Yinelenen mesajlar. QoS 1, yeniden bağlanma sonrası bir mesajı iki kez teslim edebilir ve kayıt bu kopyayı da tutar. Bu gerçekçidir — tüketicileriniz bununla başa çıkmalıdır — ama mesaj sayarken bunu bilin.
• Eksik MQTT 5 meta verileri. Yukarıdaki betikler topic, payload, QoS ve retain bilgisini saklar. Sisteminiz user property'lere, content type'a veya response topic'lere dayanıyorsa kayda msg.properties alanlarını ekleyin ve publish ederken bir Properties nesnesi geçin.
• Saate bağlı tüketiciler. Gerçek saati kullanan mantık (örneğin "60 sn boyunca ölçüm yok") 10× hızda farklı davranır. Bu testleri 1× hızda oynatın.
• Broker limitleri. Büyük kayıtların hızlı oynatılması, paylaşımlı broker'larda mesaj hızı veya kuyruk limitlerine takılabilir. Hızlı bir kontrolün ötesindeki her şey için kendi instance'ınızı kullanın.

Oynatmayı gözle doğrulayın

Herhangi bir şeyi otomatikleştirmeden önce bir oynatmanın gelişini izleyin. Online MQTT istemcisini açın, güvenli WebSocket üzerinden broker.connectmq.com adresine bağlanın ve yeniden yazılmış öneke abone olun; ardından replay.py'yi başlatın. Mesajların orijinal desene uyan dalgalar halinde geldiğini görmelisiniz.

Oynatmayı tarayıcınızda izleyin

Bir kaydedici, bir oynatıcı ve bir karşılaştırma betiği toplamda yüz satırdan az Python'dur ve gerçek trafiği tekrarlanabilir testlere dönüştürür. Sahadan bir kez kaydedin, kaydı temizleyin, kodunuzun yanında saklayın; gelecekteki her değişiklik, cihazlarınızın gerçekten gönderdiği verilere karşı test edilmiş olur.

Sıkça sorulan sorular