MQTT için pytest Fixture'ları: İzole Broker Testleri

Çoğu MQTT test paketi aynı şekilde başlar: bağlanan, iki saniye uyuyan, publish eden, tekrar uyuyan ve global bir listeyi kontrol eden bir test. İlk gün çalışır, onuncu gün flaky olur. pytest fixture'ları bağlantı kurulumunu, topic adlandırmayı, beklemeyi ve temizliği birkaç yeniden kullanılabilir yapı taşına taşıyarak bunu çözer; böylece her testin gövdesi üç satırlık bir niyete iner.

Bu rehber paho-mqtt 2.x için bir conftest.py kurar: ortam değişkenleriyle seçilen bir broker, otomatik teardown'lu bağlı client fabrikası, test başına benzersiz bir topic, timeout'lu bir mesaj collector'ı ve retained mesaj temizleyicisi.

İyi bir MQTT fixture seti ne sağlar

Fixture	Scope	Sorumluluk
`broker`	session	Ortam değişkenlerinden host, port, kimlik bilgileri ve TLS; erişilemiyorsa skip
`run_id`	session	Bu çalıştırmayı aynı broker'daki diğerlerinden ayıran rastgele ID
`topic`	function	Test adından türetilen, test başına benzersiz bir topic öneki
`mqtt_client`	function	Bağlı client'lar döndüren ve sonra bağlantılarını kesen fabrika
`subscriber`	function	Fabrika: herhangi bir filtreye subscribe olur, bir collector döndürür
`collector`	function	Testin kendi topic'inde, `wait_for()` ve timeout içeren bir subscriber
`retained`	function	Retained topic'leri kaydeder ve teardown'da temizler

Üç özellik önemlidir: izolasyon (hiçbir test başka bir testin mesajlarını görmez), sabit sleep yok (olayları bir son tarihle beklemek) ve garantili temizlik (test başarısız olsa bile teardown çalışır).

conftest.py

Bağımlılıkları kurun: pip install pytest paho-mqtt. Ardından şunu tests/conftest.py dosyasına koyun:

tests/conftest.py

import os, re, socket, threading, uuid
from dataclasses import dataclass

import paho.mqtt.client as mqtt
import pytest


@dataclass(frozen=True)
class BrokerConfig:
    host: str
    port: int
    username: str | None
    password: str | None
    tls: bool


@pytest.fixture(scope="session")
def broker():
    cfg = BrokerConfig(
        host=os.getenv("MQTT_HOST", "broker.connectmq.com"),
        port=int(os.getenv("MQTT_PORT", "1883")),
        username=os.getenv("MQTT_USER"),
        password=os.getenv("MQTT_PASS"),
        tls=os.getenv("MQTT_TLS") == "1",
    )
    try:
        socket.create_connection((cfg.host, cfg.port), timeout=3).close()
    except OSError as exc:
        pytest.skip(f"MQTT broker {cfg.host}:{cfg.port} unreachable: {exc}")
    return cfg


@pytest.fixture(scope="session")
def run_id():
    return uuid.uuid4().hex[:8]


@pytest.fixture
def topic(request, run_id):
    name = re.sub(r"[^A-Za-z0-9_-]", "_", request.node.name)
    return f"testmqtt/pytest/{run_id}/{name}"


@pytest.fixture
def mqtt_client(broker):
    clients = []

    def make(client_id=None, protocol=mqtt.MQTTv311, **connect_kwargs):
        client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=client_id or f"pytest-{uuid.uuid4().hex[:10]}",
            protocol=protocol,
        )
        if broker.username:
            client.username_pw_set(broker.username, broker.password)
        if broker.tls:
            client.tls_set()
        clients.append(client)

        connected, result = threading.Event(), {}

        def on_connect(c, userdata, flags, reason_code, properties):
            result["rc"] = reason_code
            connected.set()

        client.on_connect = on_connect
        client.connect(broker.host, broker.port, keepalive=30, **connect_kwargs)
        client.loop_start()
        assert connected.wait(10), f"no CONNACK from {broker.host}:{broker.port}"
        assert not result["rc"].is_failure, f"connection refused: {result['rc']}"
        return client

    yield make

    for client in clients:
        client.disconnect()
        client.loop_stop()

Bilinçli birkaç tercih:

Broker ortamdan gelir. Yerelde MQTT_HOST=localhost bir Mosquitto container'ını gösterir; CI'da aynı testler bir Testcontainers broker'ına ya da bir staging instance'ına karşı çalışır. Varsayılan değer, hızlı çalıştırmalar için yeterli olan paylaşımlı broker.connectmq.com endpoint'idir.
Erişilemeyen broker → başarısızlık değil skip. Docker'ı çalışmayan bir geliştirici otuz bağlantı hatası yerine net bir skip nedeni görür. CI'ın kesin olarak başarısız olmasını istiyorsanız skip'i pytest.fail'e çeviren bir bayrak ekleyin.
Fabrika CONNACK'i bekler. connect() yalnızca CONNECT paketini gönderir; Event broker'ın yanıtını bekler ve broker reddederse assertion reason code değerini gösterir.
Teardown, loop'u durdurmadan önce bağlantıyı keser; böylece DISCONNECT paketi gerçekten gönderilir ve broker bir test client'ı için Last Will yayınlamaz.
Rastgele client ID'ler. Aynı ID'yi kullanan iki test birbirini broker'dan düşürür; bkz. client ID en iyi uygulamaları ya da okunabilir ID'leri client ID üreticisi ile oluşturun.

Sleep olmadan mesaj beklemek

Collector, testlerinizden time.sleep() çağrılarını kaldıran parçadır. Gelen mesajları bir condition variable altında listeye ekler ve testin belirli sayıda mesaj gelene kadar beklemesini sağlar:

tests/conftest.py (continued)

class MessageCollector:
    def __init__(self):
        self.messages = []
        self._cond = threading.Condition()

    def __call__(self, client, userdata, msg):  # used as on_message
        with self._cond:
            self.messages.append(msg)
            self._cond.notify_all()

    def wait_for(self, count=1, timeout=5.0):
        with self._cond:
            if not self._cond.wait_for(lambda: len(self.messages) >= count, timeout):
                got = [m.topic for m in self.messages]
                raise AssertionError(f"expected {count} message(s) in {timeout}s, got {len(got)}: {got}")
            return list(self.messages[:count])

    def assert_silent(self, duration=1.0):
        with self._cond:
            self._cond.wait_for(lambda: len(self.messages) > 0, duration)
            assert not self.messages, f"unexpected messages: {[m.topic for m in self.messages]}"


def subscribe(client, topic_filter, qos=1, timeout=5.0):
    done, result = threading.Event(), {}

    def on_subscribe(c, userdata, mid, reason_code_list, properties):
        result["codes"] = reason_code_list
        done.set()

    client.on_subscribe = on_subscribe
    client.subscribe(topic_filter, qos=qos)
    assert done.wait(timeout), f"no SUBACK for {topic_filter}"
    assert not result["codes"][0].is_failure, f"subscription refused: {result['codes'][0]}"


@pytest.fixture
def subscriber(mqtt_client):
    def make(topic_filter, qos=1):
        messages = MessageCollector()
        sub = mqtt_client()
        sub.on_message = messages
        subscribe(sub, topic_filter, qos)
        return messages
    return make


@pytest.fixture
def collector(subscriber, topic):
    return subscriber(f"{topic}/#")


@pytest.fixture
def retained(mqtt_client):
    topics = []
    yield topics.append
    if topics:
        cleaner = mqtt_client()
        for t in topics:
            cleaner.publish(t, b"", qos=1, retain=True).wait_for_publish(5)

SUBACK'i beklemek, mesajları beklemek kadar önemlidir: bir test subscription aktif olmadan publish ederse broker mesajı düşürür ve test rastgele başarısız olur. subscriber ve collector fixture'ları ancak broker subscription'ı onayladıktan sonra döner.

assert_silent() negatif durumu kapsar; örneğin bir ACL'in bir topic'i reddettiğini veya retain=False olan bir mesajın tekrar gönderilmediğini. Sürenin tamamını bekler ve bir şey gelir gelmez başarısız olur.

Spesifikasyon gibi okunan testler

Fixture'lar hazır olduğunda test gövdeleri niyete indirgenir:

tests/test_telemetry.py

import json
import pytest


@pytest.mark.parametrize("qos", [0, 1, 2])
def test_roundtrip(mqtt_client, collector, topic, qos):
    pub = mqtt_client()
    pub.publish(f"{topic}/temp", "21.5", qos=qos).wait_for_publish(5)

    [msg] = collector.wait_for(1)
    assert msg.topic == f"{topic}/temp"
    assert msg.payload == b"21.5"


def test_ordering_within_one_topic(mqtt_client, collector, topic):
    pub = mqtt_client()
    for i in range(20):
        pub.publish(f"{topic}/seq", json.dumps({"i": i}), qos=1)

    received = [json.loads(m.payload)["i"] for m in collector.wait_for(20, timeout=10)]
    assert received == list(range(20))


def test_retained_status_reaches_late_subscriber(mqtt_client, subscriber, topic, retained):
    status = f"{topic}/status"
    retained(status)
    mqtt_client().publish(status, "online", qos=1, retain=True).wait_for_publish(5)

    late = subscriber(status)          # subscribes after the publish
    [msg] = late.wait_for(1)
    assert msg.payload == b"online" and msg.retain

Her test testmqtt/pytest/3f9a1c2e/test_roundtrip_1_ gibi kendi topic önekini alır (test_roundtrip[1] adından); bu sayede parametrize edilmiş varyantlar birbirini asla görmez. Hata ayıklarken test paketinin trafiğini canlı izlemek için online MQTT client üzerinden testmqtt/pytest/# topic'ine subscribe olun.

Retained testi topic'ini publish etmeden önce kaydeder. Bir assertion başarısız olsa bile retained teardown'u boş bir retained payload yayınlar ve bir sonraki çalıştırma temiz başlar. Eski retained durumun neden sık görülen bir flaky sonuç kaynağı olduğunu retained mesajları test etmek yazısında bulabilirsiniz.

Fabrika üzerinden MQTT 5 seçenekleri

Ek keyword argümanları doğrudan connect()'e gittiği için MQTT 5 özellikleri yeni bir fixture gerektirmez:

tests/test_sessions.py

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties


def test_mqtt5_connect_with_session_expiry(mqtt_client):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 300
    client = mqtt_client(protocol=mqtt.MQTTv5, clean_start=False, properties=props)
    assert client.is_connected()

Aynı kalıbı user properties, request-response veya message expiry testleri için de kullanın.

Test paketini çalıştırmak

pytest.ini

[pytest]
testpaths = tests
markers =
    mqtt: tests that need a live MQTT broker

Run locally and in CI

pytest -q                                            # default broker
MQTT_HOST=localhost pytest -q                        # local Mosquitto
MQTT_PORT=8883 MQTT_TLS=1 MQTT_USER=$MQTT_USER MQTT_PASS=$MQTT_PASS pytest -q
pip install pytest-xdist && pytest -n 4              # parallel workers

Son savunma hattı olarak pytest.ini içinde global timeout = 30 ile pytest-timeout ekleyin: hiç gelmeyen bir mesajı bekleyen test, CI job'ını kilitlemek yerine başarısız olur. Eksiksiz bir pipeline için GitHub Actions'ta MQTT testleri yazısına bakın.

Tuzaklar

Callback içinde assertion. on_message içindeki bir assert, paho'nun ağ thread'inde çalışır; hata yutulur ve test geçer. Callback'te toplayın, testte doğrulayın.
Session scope'lu client'lar. Daha hızlıdır ama bir testin yaptığı subscription bir sonraki teste mesaj taşımaya devam eder. Bir client'ı paylaşıyorsanız function scope'lu bir sarmalayıcıda unsubscribe olun ve callback'leri sıfırlayın.
Wildcard içeren topic adları. Parametrize ID'leri + veya # gibi karakterler içerebilir; topic fixture'ındaki re.sub() bunları temizler.
Broker gerektirmeyen unit testler. Yalnızca payload biçimlendiren veya topic seçen mantık, sahte bir client ile hızlı unit testlere aittir; bkz. unit testler için mock broker'lar.

Bir düzine satırlık fixture kısa sürede kendini amorti eder: testler uyumayı, birbirine karışmayı bırakır ve arkalarını temizler. conftest.py dosyasını deponuzda tutun, broker'ı ortam değişkenleriyle seçin; aynı test paketi dizüstü bilgisayarda, CI'da ve staging'e karşı değişiklik yapmadan çalışır.

Sıkça sorulan sorular

MQTT fixture'ları session scope mu olmalı, function scope mu?

Broker yapılandırmasını session scope'ta, client'ları function scope'ta tutun. Her test için yeni bir client bir CONNECT gidiş-dönüşüne mal olur ama hiçbir subscription'ın, callback'in veya kuyruktaki mesajın bir testten diğerine sızmamasını garanti eder. Client'ı testler arasında yalnızca bağlantı süresi ölçülmüş bir sorunsa paylaşın.

pytest'te sleep kullanmadan bir MQTT mesajını nasıl beklerim?

Mesajları bir threading.Condition altında listeye ekleyen ve wait_for(count, timeout) sunan bir collector kullanın. Test yalnızca mesaj gelene kadar bekler; timeout olursa sabit bir süre uyuyup umut etmek yerine anlaşılır bir mesajla başarısız olur.

MQTT testleri pytest-xdist ile paralel çalışabilir mi?

Evet, her test kendi topic'ini ve kendi client ID'sini kullanıyorsa. Topic'i çalıştırma başına bir ID ile test düğümünün adından türetmek ve rastgele client ID'ler üretmek, paralel worker'ların birbirinin mesajlarını almasını veya birbirini broker'dan düşürmesini önler.

MQTT için pytest Fixture'ları: Temiz, Hızlı ve İzole Testler