MQTT konuşan kodu birim testleriyle sınamak istiyorsanız broker'ı değil, kodunuzun sahip olduğu sınırı mock'layın. Client'ı küçük bir arayüzün arkasına alın, birim testlerinde onu bir mock ya da bellek içi bir fake ile değiştirin ve gerçek broker'ı daha ince bir entegrasyon testi katmanına bırakın. Böylece birim testleriniz milisaniyeler içinde çalışır, ağ yüzünden asla kırılmaz ve yine de doğru topic ile payload'ın gönderildiğini, gelen mesajların doğru işlendiğini kanıtlar.
Bu rehber, pratikte işe yarayan üç test double'ını — mock'lanmış bir client, elle yazılmış bir fake ve gömülü bir bellek içi broker — Python (paho-mqtt 2.x) ve JavaScript (MQTT.js 5) örnekleriyle gösteriyor ve her birinin nerede işe yaramaz hâle geldiğini açıklıyor.
Doğru test double'ını seçin
| Double | Nedir | Ne için iyi | Kör noktalar |
|---|---|---|---|
| Mock / stub client | Kütüphane client'ı yerine unittest.mock, jest.fn() | Çağrıları doğrulamak: topic, payload, QoS, retain bayrağı | Wildcard'lar, QoS, retained mesajlar, yeniden bağlanma — protokol düzeyinde hiçbir şey gerçek değil |
| Elle yazılmış fake | Aynı arayüze sahip küçük, bellek içi bir pub/sub sınıfı | Birden fazla bileşen arasındaki akışlar, topic yönlendirme | Ancak fake'iniz kadar doğru; ağ yok, oturum durumu yok |
| Gömülü broker | Test sürecinin içinde başlatılan aedes (Node.js) | Docker olmadan gerçek MQTT paketleri, çok hızlı | Üretim broker'ınız değil; ACL'ler, limitler ve cluster yapısı farklı |
| Gerçek broker | Container içinde Mosquitto/EMQX ya da barındırılan bir test instance'ı | Entegrasyon ve uçtan uca testler | Daha yavaş, kurulum ve temizlik gerektirir |
İlk ikisi birim testidir. Son ikisi, hızlı hissettirse bile entegrasyon testidir. Bu ayrımı net tutmak, gizlice çalışan bir broker'a bağımlı "birim testleri" yazmanızı ve doğrulamak istediğiniz davranışı tam da mock'layarak yok etmenizi engeller.
Önce dikiş yerini tasarlayın
İş mantığı kodunuz paho.mqtt.client ya da mqtt.connect'i hiçbir zaman doğrudan çağırmıyorsa mock'lamak zahmetsizdir. Bunun yerine client'ı (veya ince bir wrapper'ı) enjekte edin:
import json
class TelemetryPublisher:
def __init__(self, client, device_id: str):
self.client = client
self.topic = f"testmqtt/devices/{device_id}/telemetry"
def report(self, temperature: float) -> None:
if not -40 <= temperature <= 125:
raise ValueError("sensor out of range")
payload = json.dumps({"t": round(temperature, 2)})
self.client.publish(self.topic, payload, qos=1, retain=False)
def handle_command(client, userdata, msg):
"""paho-mqtt on_message callback: toggle a relay on 'on'/'off' commands."""
command = msg.payload.decode().strip().lower()
if command not in ("on", "off"):
return
userdata["relay"].set(command == "on")
client.publish(msg.topic.replace("/cmd", "/state"), command, qos=1, retain=True)Burada hiçbir şey socket açmıyor; dolayısıyla burada hiçbir şeyin test edilmek için broker'a ihtiyacı yok.
Python: unittest.mock ile paho-mqtt mock'lama
Mock'un gerçek Client imzasına sahip olması için create_autospec kullanın — böylece client.publsh(...) gibi bir yazım hatası sessizce geçmek yerine testi başarısız kılar.
import json
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock, create_autospec
import paho.mqtt.client as mqtt
from telemetry import TelemetryPublisher, handle_command
class TelemetryTests(unittest.TestCase):
def setUp(self):
self.client = create_autospec(mqtt.Client, instance=True)
def test_report_publishes_json_at_qos1(self):
TelemetryPublisher(self.client, "dev-42").report(21.456)
self.client.publish.assert_called_once_with(
"testmqtt/devices/dev-42/telemetry", json.dumps({"t": 21.46}), qos=1, retain=False
)
def test_out_of_range_is_not_published(self):
with self.assertRaises(ValueError):
TelemetryPublisher(self.client, "dev-42").report(500)
self.client.publish.assert_not_called()
def test_command_sets_relay_and_reports_state(self):
relay = MagicMock()
msg = SimpleNamespace(topic="testmqtt/devices/dev-42/cmd", payload=b" ON ", qos=1, retain=False)
handle_command(self.client, {"relay": relay}, msg)
relay.set.assert_called_once_with(True)
self.client.publish.assert_called_once_with(
"testmqtt/devices/dev-42/state", "on", qos=1, retain=True
)
if __name__ == "__main__":
unittest.main()İşin çoğunu iki teknik yapar: giden trafik için publish çağrılarını doğrulayın, gelen trafik için callback'i sahte bir mesajla kendiniz çağırın. Handler'ınız yalnızca topic, payload, qos ve retain özniteliklerini okuduğu için bunları içeren bir SimpleNamespace yeterlidir. Kodunuz import sırasında client.connect() çağırıyorsa unittest.mock.patch("paho.mqtt.client.Client") kullanın — ama bunu bağlantıyı enjekte edilebilir bir factory'ye taşımanız gerektiğinin işareti olarak görün.
JavaScript: Jest ile MQTT.js mock'lama
Bir MQTT.js client'ı bir EventEmitter'dır; bu yüzden en gerçekçi mock, jest.fn() metotları eklenmiş gerçek bir EventEmitter'dır. Kodunuz on('message') handler'larını her zamanki gibi kaydeder, test de olayları yayar.
import { EventEmitter } from 'node:events';
import { connect } from 'mqtt';
import { startAlarm } from './alarm.js'; // subscribes to sensors, publishes alarms
jest.mock('mqtt', () => ({ connect: jest.fn() }));
function fakeClient() {
const client = new EventEmitter();
client.subscribe = jest.fn((topic, opts, cb) => cb?.(null, [{ topic, qos: opts?.qos ?? 0 }]));
client.publish = jest.fn((topic, payload, opts, cb) => cb?.());
client.end = jest.fn();
return client;
}
test('publishes an alarm when temperature exceeds the threshold', () => {
const client = fakeClient();
connect.mockReturnValue(client);
startAlarm({ url: 'mqtts://broker.connectmq.com:8883', threshold: 30 });
client.emit('connect');
expect(client.subscribe).toHaveBeenCalledWith('testmqtt/sensors/+/temp', { qos: 1 }, expect.any(Function));
client.emit('message', 'testmqtt/sensors/kitchen/temp', Buffer.from('31.5'), { retain: false });
expect(client.publish).toHaveBeenCalledWith(
'testmqtt/alarms/kitchen', expect.stringContaining('"temp":31.5'), { qos: 1 }, expect.any(Function)
);
});message olayı (topic, payload, packet) iletir ve payload bir Buffer'dır — mock'ta bu biçimi koruyun; aksi hâlde testiniz string'lerle geçerken üretim Buffer alır. Daha fazla kalıp (yeniden bağlanma mantığı için sahte zamanlayıcılar, async yardımcılar) için Jest ile MQTT testi yazısına bakın.
Elle yazılmış bellek içi bir fake
Birden fazla bileşen topic'ler üzerinden konuştuğunda tek tek publish çağrılarını doğrulamak kırılganlaşır. Mesajları gerçekten yönlendiren küçük bir fake çoğu zaman daha anlaşılırdır. paho-mqtt topic_matches_sub ile gelir, yani wildcard eşleştirmesi zahmetsizce doğru olur:
from types import SimpleNamespace
from paho.mqtt.client import topic_matches_sub
class FakeMqtt:
"""In-memory pub/sub with the subset of the paho API our code uses."""
def __init__(self):
self.subscriptions, self.retained, self.published = [], {}, []
self.on_message = None
def subscribe(self, topic, qos=0):
self.subscriptions.append(topic)
for t, payload in self.retained.items():
if topic_matches_sub(topic, t):
self._deliver(t, payload, retain=True)
def publish(self, topic, payload, qos=0, retain=False):
data = payload.encode() if isinstance(payload, str) else payload
self.published.append((topic, data, qos, retain))
if retain and data:
self.retained[topic] = data
elif retain:
self.retained.pop(topic, None) # empty retained payload clears
if any(topic_matches_sub(s, topic) for s in self.subscriptions):
self._deliver(topic, data, retain=False)
def _deliver(self, topic, payload, retain):
if self.on_message:
self.on_message(self, None, SimpleNamespace(topic=topic, payload=payload, qos=0, retain=retain))Fake'i küçücük tutun ve bir kez gerçek bir broker'a karşı test edin (aynı senaryoyu iki yoldan da geçirin); yoksa sonunda fake'i test ediyor olursunuz. Wildcard beklentilerinizi topic matcher ile kontrol edebilirsiniz.
Node.js: bellek içi broker olarak aedes
Aedes, bir Jest veya Mocha süreci içinde başlatabileceğiniz gerçek bir MQTT broker'dır. Yerel bir TCP portu üzerinden gerçek protokolü konuşur; MQTT.js kodunuz değişmeden çalışır — hiç mock yok:
import { createServer } from 'node:net';
import { Aedes } from 'aedes';
import mqtt from 'mqtt';
let broker, server, url;
beforeAll(async () => {
broker = await Aedes.createBroker(); // aedes 0.x: require('aedes')()
server = createServer(broker.handle);
await new Promise((resolve) => server.listen(0, resolve)); // random free port
url = `mqtt://127.0.0.1:${server.address().port}`;
});
afterAll(async () => {
await new Promise((resolve) => server.close(resolve));
await new Promise((resolve) => broker.close(resolve));
});
test('retained state is delivered to late subscribers', async () => {
const pub = await mqtt.connectAsync(url);
await pub.publishAsync('testmqtt/unit/state', 'ready', { qos: 1, retain: true });
const sub = await mqtt.connectAsync(url);
const received = new Promise((resolve) => sub.on('message', (t, p, pkt) => resolve([t, p.toString(), pkt.retain])));
await sub.subscribeAsync('testmqtt/unit/#', { qos: 1 });
expect(await received).toEqual(['testmqtt/unit/state', 'ready', true]);
await Promise.all([pub.endAsync(), sub.endAsync()]);
});0 portunda dinlemek her test dosyasına kendi boş portunu verir; böylece paralel Jest worker'ları çakışmaz. Python'da bu kadar yaygın bir gömülü broker yok; çoğu ekip bunun yerine bir Mosquitto container'ı kullanır — bkz. MQTT için Testcontainers ve pytest fixture'ları.
Birim testlerinin bittiği yer
Mock'lar sizin kararlarınızı doğrulamakta mükemmeldir. Broker'ın kararlarını doğrulayamazlar. Bir test şunlara bağlıysa onu entegrasyon katmanına taşıyın:
- QoS ve oturumlar — bağlantı kopmasından sonra yeniden teslim, kalıcı oturumlar, kopya mesajların ele alınması.
- Retained mesajlar ve Last Will — bunlar broker özellikleridir; bir fake yalnızca sizin varsayımlarınızı yeniden üretir.
- Kimlik doğrulama ve ACL'ler — bir mock asla
Not authorizeddöndürmez; gerçek bir broker döndürür, bazen sessizce. - Payload limitleri, topic alias'ları, MQTT 5 özellikleri — limitleri broker yapılandırması belirler.
- TLS, WebSocket'ler, yeniden bağlanma zamanlaması — ağ davranışı bir ağ gerektirir.
Sağlıklı bir piramitte çok sayıda hızlı mock'lu test, gerçek bir broker'a karşı birkaç düzine entegrasyon testi (bkz. MQTT entegrasyon testleri) ve yayına aldığınız ortama karşı bir avuç uçtan uca kontrol bulunur.
Mock'u gerçeklikle karşılaştırın
Mock'lu bir teste güvenmeden önce, varsayımlarının gerçek bir broker'la örtüştüğünü bir kez doğrulayın. Aynı mesajı online MQTT client üzerinden broker.connectmq.com'a yayınlayın ve karşılaştırın: payload türü, sonradan abone olanlarda retain bayrağı ve topic string'lerinin birebir hâli. Mock'unuzun döndürdüğünden farklıysa önce mock'u düzeltin.
Mock'unuzu gerçek bir broker'la karşılaştırın
Kodunuzu hızlıca test etmek için client'ı mock'layın, bileşenler etkileşime girdiğinde küçük bir fake ya da gömülü bir broker kullanın ve protokol garantilerini gerçek bir broker'a karşı çalışan entegrasyon testlerine bırakın. Bu ayrım, testlerin MQTT'nin gerçekte ne yaptığı konusunda yalan söylemesine izin vermeden test paketini hızlı tutar.
Sıkça sorulan sorular
Testlerde MQTT client'ını mı mock'lamalıyım, yoksa gerçek bir broker mı kullanmalıyım?
İkisini de, farklı katmanlarda. Kendi mantığınızı (hangi topic, hangi payload, bir mesaja nasıl tepki verdiğiniz) milisaniyeler içinde test etmek için client'ı mock'layın. QoS, retained mesajlar, ACL'ler ve yeniden bağlanma gibi protokol davranışlarını doğrulamak için entegrasyon testlerinde gerçek bir broker kullanın.
Aedes bir mock mu, yoksa gerçek bir MQTT broker mı?
Aedes, Node.js ile yazılmış gerçek bir MQTT broker'dır. Bellek içi kalıcılıkla test sürecine gömüldüğünde hızlı ve tek kullanımlık bir broker gibi davranır; bu yüzden onu kullanan testler saf birim testi değil, küçük entegrasyon testleridir.
Bir birim testinde gelen MQTT mesajını nasıl simüle ederim?
Mesaj handler'ını, kodunuzun okuduğu alanlara (topic, payload, qos, retain) sahip sahte bir mesaj nesnesiyle doğrudan çağırın. paho-mqtt'de bu on_message(client, userdata, msg) çağırmak; MQTT.js'te ise mock'lanmış EventEmitter client üzerinde bir 'message' olayı yaymak demektir.
