Nombrar los topics de prueba para que las pruebas nunca colisionen

Las pruebas MQTT que pasan por separado y fallan juntas casi siempre comparten topics. Dos jobs de CI publican en test/temp, un mensaje retenido de ayer responde a la aserción de hoy o un suscriptor en # se traga el tráfico de otro. La solución es una convención de nombres: cada ejecución recibe un prefijo único como testmqtt/<suite>/<run-id>/, las pruebas se suscriben solo por debajo de él y el teardown borra todo lo que quedó retenido.

Esta guía muestra un esquema de nombres que escala, cómo construir IDs de ejecución, helpers para Python y JavaScript, y cómo limpiar los datos de prueba retenidos, incluidos los restos de ejecuciones que se colgaron.

Por qué colisionan las pruebas MQTT

Un broker es estado compartido. A diferencia de un mock en memoria, recuerda y distribuye, así que las pruebas se contaminan entre sí de unas pocas formas previsibles:

• Ejecuciones en paralelo: dos pipelines, dos desarrolladores o dos workers de pytest publican en el mismo topic y reciben los mensajes del otro.
• Restos retenidos: un mensaje retenido sobrevive a la prueba que lo creó y se entrega al instante al siguiente suscriptor, haciendo que una funcionalidad rota parezca funcionar.
• Wildcards demasiado amplios: una prueba suscrita a sensors/# recibe todos los mensajes de sensores de todas las demás pruebas del broker.
• Client IDs reutilizados: dos ejecuciones con el mismo client ID se expulsan mutuamente del broker, y las sesiones persistentes entregan mensajes en cola de una ejecución anterior.
• Brokers públicos compartidos: en un sandbox público, desconocidos pueden usar los mismos nombres de topic «obvios» que tú.

Un esquema de nombres que escala

Construye los topics de prueba con niveles fijos, del más general al más concreto:

El resultado es testmqtt/orders/20261017T101500-gh812-a3f9/test-retained-status/device/42/status. Parece largo, pero la parte lógica del final no cambia: el código bajo prueba solo recibe un prefijo, el mismo mecanismo que se usa para separar entornos (consulta configurar desarrollo, staging y producción).

Unas pocas reglas mantienen estos nombres seguros:

• Solo ASCII en minúsculas, dígitos, - y _; sin espacios, sin / inicial y nunca + ni # dentro de un nombre.
• No empieces por $: los topics que empiezan por $ están reservados para el broker ($SYS) y quedan excluidos de las suscripciones a #.
• Mantén cada nivel corto. Los topics viajan en cada paquete PUBLISH; consulta buenas prácticas de diseño de topics para las reglas generales.

Elegir un ID de ejecución

La mejor opción los combina: primero una marca de tiempo UTC (para poder limpiar restos por antigüedad), luego el número y el intento de la ejecución del CI si están disponibles, y unos caracteres hexadecimales aleatorios para separar ejecuciones que empiezan en el mismo segundo.

Python: un fixture de prefijo para pytest

import os, re, secrets, time
import pytest
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
SUITE = "orders"

def make_run_id() -> str:
    rid = time.strftime("%Y%m%dT%H%M%S", time.gmtime())
    if os.getenv("GITHUB_RUN_ID"):
        rid += f"-gh{os.environ['GITHUB_RUN_ID']}-{os.getenv('GITHUB_RUN_ATTEMPT', '1')}"
    return f"{rid}-{secrets.token_hex(2)}"

RUN_ID = make_run_id()

def clear_retained(client, prefix, settle=1.5):
    found = set()
    def collect(c, userdata, msg):
        if msg.retain and msg.payload:
            found.add(msg.topic)
    client.message_callback_add(f"{prefix}/#", collect)
    client.subscribe(f"{prefix}/#", qos=1)
    time.sleep(settle)                     # los mensajes retenidos llegan justo después del SUBACK
    client.unsubscribe(f"{prefix}/#")
    client.message_callback_remove(f"{prefix}/#")
    for topic in found:
        client.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
    return found

@pytest.fixture
def prefix(request):
    case = re.sub(r"[^a-z0-9_-]+", "-", request.node.name.lower()).strip("-")
    return f"testmqtt/{SUITE}/{RUN_ID}/{case}"

@pytest.fixture
def mqttc(prefix):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                    client_id=f"t-{RUN_ID}-{secrets.token_hex(3)}")
    if os.getenv("MQTT_USER"):
        c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.connect(HOST, 1883)
    c.loop_start()
    yield c
    clear_retained(c, prefix)
    c.loop_stop()
    c.disconnect()

Así, una prueba construye cada topic a partir de prefix y nunca escribe uno a mano. Como el nombre de la prueba forma parte del prefijo, las pruebas parametrizadas y los workers en paralelo (pytest-xdist) quedan separados automáticamente. Para más patrones de fixtures, consulta fixtures de pytest para MQTT.

JavaScript: la misma idea para Jest o Vitest

import { randomBytes } from "node:crypto";

const ts = new Date().toISOString().replace(/[-:]/g, "").slice(0, 15); // 20261017T101500
const ci = process.env.GITHUB_RUN_ID ? `-gh${process.env.GITHUB_RUN_ID}` : "";
export const RUN_ID = `${ts}${ci}-${randomBytes(2).toString("hex")}`;

export const testPrefix = (suite, name) =>
  `testmqtt/${suite}/${RUN_ID}/${name.toLowerCase().replace(/[^a-z0-9_-]+/g, "-")}`;

Con MQTT.js 5, una prueba que deba publicar datos retenidos también puede añadir una caducidad, para que el broker los descarte aunque el teardown nunca se ejecute:

client.publish(`${prefix}/device/42/status`, "online", {
  qos: 1,
  retain: true,
  properties: { messageExpiryInterval: 3600 }, // solo MQTT 5
});

La caducidad de mensajes requiere una conexión MQTT 5 (protocolVersion: 5) y un broker que la soporte; los detalles están en caducidad de mensajes MQTT.

Suscríbete de forma acotada

Una prueba debe suscribirse a <su prefijo>/# o a topics exactos por debajo de él, nada más amplio. Antes de fiarte de un filtro, compruébalo contra los topics que esperas con el topic matcher; un + mal colocado es la forma clásica en que una prueba empieza a recibir los mensajes de otra.

En un broker que controles, refuérzalo con ACL: da al usuario de CI acceso de lectura y escritura solo a testmqtt/#. Una prueba que apunte por error a un topic de producción fallará entonces con «not authorized» en lugar de publicar.

Limpia los restos de ejecuciones interrumpidas

El teardown no se ejecuta cuando se cancela un job de CI o un portátil entra en reposo. Como cada ID de ejecución empieza con una marca de tiempo, un proceso de limpieza programado puede borrar los mensajes retenidos de ejecuciones con más de un día:

import os, time
import paho.mqtt.client as mqtt

cutoff = time.strftime("%Y%m%dT%H%M%S", time.gmtime(time.time() - 86400))
stale = set()

def on_message(client, userdata, msg):
    parts = msg.topic.split("/")          # testmqtt/<suite>/<run-id>/...
    if msg.retain and msg.payload and len(parts) > 3 and parts[2][:15] < cutoff:
        stale.add(msg.topic)

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-janitor")
if os.getenv("MQTT_USER"):
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
c.on_message = on_message
c.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883)
c.loop_start()
c.subscribe("testmqtt/#", qos=1)
time.sleep(5)
for topic in stale:
    c.publish(topic, b"", qos=1, retain=True).wait_for_publish(5)
print(f"cleared {len(stale)} stale retained messages")
c.loop_stop(); c.disconnect()

Ejecútalo cada noche, por ejemplo como job programado junto a tu pipeline de pruebas en GitHub Actions. Si un mensaje retenido se niega a desaparecer, el mensaje retenido no se borra recoge los motivos habituales.

Client IDs y sesiones

El aislamiento de topics no sirve de nada si dos ejecuciones comparten client ID. Incluye el ID de ejecución en cada client ID (como hace el fixture), conéctate con sesión limpia (clean_start=True en MQTT 5, clean_session=True en 3.1.1) salvo que la prueba trate precisamente de persistencia, y mantén los IDs dentro de 23 caracteres si tienes que soportar brokers MQTT 3.1.1 estrictos. El generador de client ID produce IDs con un formato coherente.

Checklist

1. Cada topic de una prueba se construye a partir de un prefijo; no hay topics escritos a mano.
2. El prefijo contiene un ID de ejecución que empieza con una marca de tiempo UTC.
3. Las suscripciones nunca van por encima del prefijo de la propia prueba.
4. El teardown borra los mensajes retenidos bajo el prefijo; un proceso nocturno se encarga del resto.
5. Los mensajes de prueba retenidos llevan caducidad cuando hay MQTT 5 disponible.
6. Los client IDs incluyen el ID de ejecución; las sesiones son limpias por defecto.

Inspecciona tus topics de prueba en el navegador

Nombrar los topics de prueba es una decisión de cinco minutos que elimina toda una categoría de pruebas inestables. Pon un prefijo único con marca de tiempo delante de todo, mantén las suscripciones dentro de él y limpia lo que dejes: así podrás ejecutar en paralelo contra un mismo broker tantas suites como quieras.

Preguntas frecuentes