Fixtures de pytest para MQTT: tests aislados con broker

La mayoría de las suites de test MQTT empiezan igual: un test que se conecta, duerme dos segundos, publica, vuelve a dormir y revisa una lista global. Funciona el primer día y se vuelve flaky hacia el décimo. Las fixtures de pytest lo arreglan moviendo la conexión, el nombrado de topics, las esperas y la limpieza a unas pocas piezas reutilizables, de modo que el cuerpo de cada test se queda en tres líneas de intención.

Esta guía construye un conftest.py para paho-mqtt 2.x con un broker elegido por variables de entorno, una factory de clientes conectados con teardown automático, un topic único por test, un colector de mensajes con timeouts y un limpiador de mensajes retenidos.

Qué debe aportar un buen conjunto de fixtures MQTT

Fixture	Scope	Responsabilidad
`broker`	session	Host, puerto, credenciales y TLS desde variables de entorno; skip si no es alcanzable
`run_id`	session	ID aleatorio que separa esta ejecución de cualquier otra en el mismo broker
`topic`	function	Un prefijo de topic único por test, derivado del nombre del test
`mqtt_client`	function	Factory que devuelve clientes conectados y los desconecta después
`subscriber`	function	Factory: se suscribe a cualquier filtro y devuelve un colector
`collector`	function	Un subscriber sobre el topic propio del test, con `wait_for()` y timeout
`retained`	function	Registra topics retenidos y los limpia en el teardown

Importan tres propiedades: aislamiento (ningún test ve los mensajes de otro), nada de sleeps fijos (se esperan eventos, con un plazo máximo) y limpieza garantizada (el teardown se ejecuta aunque el test falle).

El conftest.py

Instala las dependencias: pip install pytest paho-mqtt. Después coloca esto en tests/conftest.py:

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()

Algunas decisiones deliberadas:

El broker viene del entorno. En local, MQTT_HOST=localhost apunta a un contenedor Mosquitto; en CI los mismos tests corren contra un broker de Testcontainers o una instancia de staging. El valor por defecto es el endpoint compartido broker.connectmq.com, suficiente para ejecuciones rápidas.
Broker inalcanzable → skip, no fallo. Un desarrollador sin Docker arrancado recibe un motivo de skip claro en lugar de treinta errores de conexión. Si quieres que CI falle sin contemplaciones, añade un flag que convierta el skip en pytest.fail.
La factory espera el CONNACK. connect() solo envía el paquete CONNECT; el Event espera la respuesta del broker y la aserción muestra el reason code si el broker la rechaza.
El teardown desconecta antes de parar el loop, así el paquete DISCONNECT se envía de verdad y el broker no publica un Last Will por un cliente de test.
Client IDs aleatorios. Dos tests con el mismo ID se expulsarían mutuamente del broker: consulta buenas prácticas de client ID o genera IDs legibles con el generador de client ID.

Esperar mensajes sin sleep

El colector es la pieza que elimina time.sleep() de tus tests. Añade los mensajes entrantes bajo una variable de condición y deja que el test espere hasta que haya llegado un número dado:

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)

Esperar el SUBACK importa tanto como esperar los mensajes: si un test publica antes de que la suscripción esté activa, el broker descarta el mensaje y el test falla al azar. Las fixtures subscriber y collector solo devuelven el control cuando el broker ha confirmado la suscripción.

assert_silent() cubre el caso negativo: por ejemplo, que una ACL deniega un topic o que un mensaje con retain=False no se reproduce. Espera la duración completa y falla en cuanto llega algo.

Tests que se leen como especificaciones

Con las fixtures listas, el cuerpo de los tests se reduce a la intención:

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

Cada test recibe su propio prefijo de topic, como testmqtt/pytest/3f9a1c2e/test_roundtrip_1_ (a partir de test_roundtrip[1]), así que las variantes parametrizadas nunca se ven entre sí. Mientras depuras, suscríbete a testmqtt/pytest/# en el cliente MQTT online para ver el tráfico de la suite en directo.

El test de retenidos registra su topic antes de publicar. Aunque falle una aserción, el teardown de retained publica un payload retenido vacío y la siguiente ejecución empieza limpia; consulta probar mensajes retenidos para ver por qué el estado retenido obsoleto es una fuente habitual de resultados flaky.

Opciones de MQTT 5 a través de la factory

Como los argumentos con nombre adicionales van directos a connect(), las funciones de MQTT 5 no necesitan una fixture nueva:

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()

Usa el mismo patrón para tests de user properties, request-response o message expiry.

Ejecutar la suite

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

Añade pytest-timeout con un timeout = 30 global en pytest.ini como última línea de defensa: un test atascado esperando un mensaje que no llega falla en lugar de colgar el job de CI. Para un pipeline completo, consulta tests MQTT en GitHub Actions.

Trampas habituales

Aserciones dentro de callbacks. Un assert en on_message se ejecuta en el hilo de red de paho; el fallo se pierde y el test pasa. Recoge en el callback, comprueba en el test.
Clientes con scope de sesión. Son más rápidos, pero una suscripción hecha por un test sigue entregando mensajes en el siguiente. Si compartes un cliente, cancela suscripciones y restablece los callbacks en un wrapper con scope de función.
Nombres de topic con comodines. Los IDs de parametrize pueden contener caracteres como + o #; el re.sub() de la fixture topic los elimina.
Tests unitarios que no necesitan broker. La lógica que solo formatea payloads o elige topics pertenece a tests unitarios rápidos con un cliente falso; consulta brokers simulados para tests unitarios.

Una docena de líneas de fixtures se amortiza enseguida: los tests dejan de dormir, dejan de interferir entre sí y limpian lo que ensucian. Guarda el conftest.py en tu repositorio, elige el broker con variables de entorno, y la misma suite correrá en un portátil, en CI y contra staging sin cambios.

Preguntas frecuentes

¿Las fixtures MQTT deben tener scope de sesión o de función?

Deja la configuración del broker con scope de sesión y los clientes con scope de función. Un cliente nuevo por test cuesta un round trip de CONNECT, pero garantiza que ninguna suscripción, callback o mensaje en cola se filtre de un test al siguiente. Comparte un cliente entre tests solo si el tiempo de conexión es un problema medido.

¿Cómo espero un mensaje MQTT en pytest sin sleep?

Usa un colector que añada los mensajes bajo un threading.Condition y exponga wait_for(count, timeout). El test se bloquea solo hasta que llega el mensaje y falla con un mensaje claro si vence el timeout, en lugar de dormir un tiempo fijo y cruzar los dedos.

¿Pueden los tests MQTT ejecutarse en paralelo con pytest-xdist?

Sí, si cada test usa su propio topic y su propio client ID. Derivar el topic de un ID por ejecución más el nombre del nodo de test, y generar client IDs aleatorios, evita que los workers paralelos reciban los mensajes de otros o se expulsen mutuamente del broker.

Fixtures de pytest para MQTT: tests limpios, rápidos y aislados