Como nomear topics de teste para que os testes nunca colidam

Testes MQTT que passam sozinhos e falham juntos quase sempre compartilham topics. Dois jobs de CI publicam em test/temp, uma mensagem retida de ontem responde à asserção de hoje ou um subscriber em # engole o tráfego de outra pessoa. A solução é uma convenção de nomes: cada execução recebe um prefixo único como testmqtt/<suite>/<run-id>/, os testes assinam apenas abaixo dele e o teardown apaga tudo o que ficou retido.

Este guia mostra um esquema de nomes que escala, como gerar IDs de execução, helpers para Python e JavaScript e como limpar os dados de teste retidos, incluindo os restos de execuções que travaram.

Por que os testes MQTT colidem

Um broker é estado compartilhado. Ao contrário de um mock em memória, ele lembra e distribui, então os testes vazam uns nos outros de algumas formas previsíveis:

• Execuções em paralelo: dois pipelines, dois desenvolvedores ou dois workers do pytest publicam no mesmo topic e recebem as mensagens um do outro.
• Restos retidos: uma mensagem retida sobrevive ao teste que a criou e é entregue na hora ao próximo subscriber, fazendo uma funcionalidade quebrada parecer funcionar.
• Wildcards amplos demais: um teste que assina sensors/# recebe todas as mensagens de sensores de todos os outros testes no broker.
• Client IDs reaproveitados: duas execuções com o mesmo client ID derrubam uma à outra do broker, e sessões persistentes entregam mensagens enfileiradas de uma execução anterior.
• Brokers públicos compartilhados: em um sandbox público, desconhecidos podem usar os mesmos nomes de topic "óbvios" que você.

Um esquema de nomes que escala

Monte os topics de teste a partir de níveis fixos, do mais geral para o mais específico:

O resultado é testmqtt/orders/20261017T101500-gh812-a3f9/test-retained-status/device/42/status. Parece longo, mas a parte lógica no final não muda: o código sob teste só recebe um prefixo, o mesmo mecanismo usado para separar ambientes (veja configurar desenvolvimento, staging e produção).

Algumas regras mantêm esses nomes seguros:

• Apenas ASCII minúsculo, dígitos, - e _; sem espaços, sem / no início e nunca + ou # dentro de um nome.
• Não comece com $: topics iniciados por $ são reservados ao broker ($SYS) e ficam de fora das assinaturas em #.
• Mantenha cada nível curto. Os topics viajam em todo pacote PUBLISH; veja boas práticas de design de topics para as regras gerais.

Escolhendo um ID de execução

A melhor opção combina tudo: primeiro um timestamp UTC (para limpar restos por idade), depois o número e a tentativa da execução do CI quando existirem, e alguns caracteres hexadecimais aleatórios para separar execuções que começam no mesmo segundo.

Python: um fixture de prefixo 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)                     # as mensagens retidas chegam logo após o 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()

Assim, um teste monta todos os topics a partir de prefix e nunca fixa um no código. Como o nome do teste faz parte do prefixo, testes parametrizados e workers em paralelo (pytest-xdist) ficam separados automaticamente. Para mais padrões de fixtures, veja fixtures do pytest para MQTT.

JavaScript: a mesma ideia para Jest ou 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, "-")}`;

Com o MQTT.js 5, um teste que precisa publicar dados retidos também pode definir uma expiração, para que o broker os descarte mesmo que o teardown nunca rode:

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

A expiração de mensagens exige uma conexão MQTT 5 (protocolVersion: 5) e um broker que a suporte; os detalhes estão em expiração de mensagens MQTT.

Assine de forma restrita

Um teste deve assinar <seu prefixo>/# ou topics exatos abaixo dele, nada mais amplo. Antes de confiar em um filtro, confira-o com os topics esperados no topic matcher; um + fora do lugar é o jeito clássico de um teste começar a receber as mensagens de outro.

Em um broker que você controla, reforce isso com ACLs: dê ao usuário do CI acesso de leitura e escrita apenas a testmqtt/#. Um teste que, por engano, mire um topic de produção falha com "not authorized" em vez de publicar.

Limpe os restos de execuções interrompidas

O teardown não roda quando um job de CI é cancelado ou um notebook entra em suspensão. Como todo ID de execução começa com um timestamp, uma rotina de limpeza agendada pode apagar as mensagens retidas de execuções com mais de um dia:

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

Rode-a toda noite, por exemplo como job agendado ao lado do seu pipeline de testes no GitHub Actions. Se uma mensagem retida se recusa a sumir, mensagem retida que não é apagada lista os motivos de sempre.

Client IDs e sessões

Isolar topics não adianta nada se duas execuções compartilham o client ID. Inclua o ID da execução em todo client ID (como faz o fixture), conecte com sessão limpa (clean_start=True no MQTT 5, clean_session=True no 3.1.1), a menos que o teste seja justamente sobre persistência, e mantenha os IDs em até 23 caracteres se precisar suportar brokers MQTT 3.1.1 rígidos. O gerador de client ID cria IDs em um formato consistente.

Checklist

1. Todo topic de um teste é montado a partir de um prefixo; nenhum topic fixo no código.
2. O prefixo contém um ID de execução que começa com um timestamp UTC.
3. As assinaturas nunca ficam acima do prefixo do próprio teste.
4. O teardown limpa as mensagens retidas sob o prefixo; uma rotina noturna cuida do resto.
5. Mensagens de teste retidas têm expiração quando o MQTT 5 está disponível.
6. Os client IDs incluem o ID da execução; as sessões são limpas por padrão.

Inspecione os seus topics de teste no navegador

Nomear topics de teste é uma decisão de cinco minutos que elimina toda uma categoria de testes instáveis. Coloque um prefixo único com timestamp na frente de tudo, mantenha as assinaturas dentro dele e limpe o que deixar para trás: assim você pode rodar em paralelo, contra um único broker, quantas suítes quiser.

Perguntas frequentes