Mock de MQTT em testes unitários: fakes e brokers em memória

Para testar com testes unitários um código que fala MQTT, não faça mock do broker — faça mock da fronteira que o seu código controla. Coloque o cliente atrás de uma interface pequena, substitua-o por um mock ou um fake em memória nos testes unitários e deixe um broker real para uma camada mais fina de testes de integração. Assim seus testes unitários rodam em milissegundos, nunca falham por causa da rede e ainda provam que o topic e o payload certos são enviados e que as mensagens recebidas são tratadas corretamente.

Este guia mostra os três test doubles que funcionam na prática — um cliente mockado, um fake escrito à mão e um broker embutido em memória — com exemplos em Python (paho-mqtt 2.x) e JavaScript (MQTT.js 5), e explica onde cada um deixa de ser útil.

Escolha o test double certo

Double	O que é	Bom para	Pontos cegos
Cliente mock / stub	`unittest.mock`, `jest.fn()` no lugar do cliente da biblioteca	Verificar chamadas: topic, payload, QoS, flag retain	Wildcards, QoS, mensagens retidas, reconexões — nada do protocolo é real
Fake escrito à mão	Uma pequena classe pub/sub em memória com a mesma interface	Testar fluxos entre vários componentes, roteamento por topic	Só é tão correto quanto o seu fake; sem rede, sem estado de sessão
Broker embutido	aedes (Node.js) iniciado dentro do processo de testes	Pacotes MQTT reais sem Docker, muito rápido	Não é o seu broker de produção; ACLs, limites e clustering são diferentes
Broker real	Mosquitto/EMQX em um container ou uma instância de teste hospedada	Testes de integração e end-to-end	Mais lento, exige preparação e limpeza

Os dois primeiros são testes unitários. Os dois últimos são testes de integração, mesmo quando parecem rápidos. Manter essa distinção clara evita escrever "testes unitários" que dependem secretamente de um broker rodando — e evita mockar justamente o comportamento que você queria verificar.

Desenhe a costura primeiro

Fazer mock é simples quando o seu código de negócio nunca chama paho.mqtt.client ou mqtt.connect diretamente. Injete o cliente (ou um wrapper fino):

telemetry.py

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)

Nada aqui abre um socket, então nada aqui precisa de um broker para ser testado.

Python: mock do paho-mqtt com unittest.mock

Use create_autospec para que o mock tenha a assinatura real de Client — um erro de digitação como client.publsh(...) faz o teste falhar em vez de passar silenciosamente.

test_telemetry.py

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

Duas técnicas fazem quase todo o trabalho: verificar as chamadas a publish para o tráfego de saída e chamar o callback você mesmo com uma mensagem falsa para o tráfego de entrada. Um SimpleNamespace com topic, payload, qos e retain basta, porque o seu handler só lê esses atributos. Se o seu código chama client.connect() no momento do import, use unittest.mock.patch("paho.mqtt.client.Client") — mas encare isso como um sinal para mover a conexão para uma factory injetável.

JavaScript: mock do MQTT.js com Jest

Um cliente MQTT.js é um EventEmitter, então o mock mais fiel é um EventEmitter real com métodos jest.fn(). O seu código registra handlers on('message') normalmente e o teste emite os eventos.

alarm.test.js

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)
  );
});

O evento message entrega (topic, payload, packet) com payload como Buffer — mantenha esse formato no mock, ou o teste vai passar com strings enquanto a produção recebe Buffers. Para mais padrões (fake timers para lógica de reconexão, helpers assíncronos) veja testes de MQTT com Jest.

Um fake em memória escrito à mão

Quando vários componentes conversam por topics, verificar chamadas individuais a publish fica frágil. Um fake pequeno que realmente roteia mensagens costuma ser mais claro. O paho-mqtt traz topic_matches_sub, então o matching de wildcards sai correto de graça:

fake_mqtt.py

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)   # payload retido vazio apaga
        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))

Mantenha o fake minúsculo e teste-o uma vez contra um broker real (o mesmo cenário pelos dois caminhos); caso contrário, você acaba testando o fake. Confira o que você espera dos wildcards com o topic matcher.

Node.js: aedes como broker em memória

O aedes é um broker MQTT real que você pode iniciar dentro de um processo de Jest ou Mocha. Ele fala o protocolo de verdade em uma porta TCP local, então o seu código MQTT.js roda sem alterações — sem mock nenhum:

broker.test.js

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()]);
});

Escutar na porta 0 dá a cada arquivo de teste sua própria porta livre, então workers paralelos do Jest não colidem. O Python não tem um broker embutido tão comum; a maioria das equipes usa um container Mosquitto — veja Testcontainers para MQTT e fixtures do pytest.

Onde os testes unitários terminam

Mocks são ótimos para verificar as suas decisões. Eles não conseguem verificar as do broker. Leve um teste para a camada de integração quando ele depender de:

QoS e sessões — reentrega após uma desconexão, sessões persistentes, tratamento de duplicatas.
Mensagens retidas e Last Will — são recursos do broker; um fake só reproduz as suas suposições.
Autenticação e ACLs — um mock nunca devolve Not authorized; um broker real devolve, às vezes em silêncio.
Limites de payload, topic aliases, propriedades do MQTT 5 — os limites vêm da configuração do broker.
TLS, WebSockets, tempo de reconexão — comportamento de rede precisa de uma rede.

Uma pirâmide saudável tem muitos testes rápidos com mock, algumas dezenas de testes de integração contra um broker real (veja testes de integração MQTT) e um punhado de verificações end-to-end no ambiente em que você faz o deploy.

Conferindo o mock com a realidade

Antes de confiar em um teste com mock, confirme uma vez que as suposições dele batem com um broker real. Publique a mesma mensagem pelo cliente MQTT online em broker.connectmq.com e compare: o tipo do payload, a flag retain para assinantes tardios e as strings exatas dos topics. Se forem diferentes do que o seu mock retorna, corrija o mock primeiro.

Compare seu mock com um broker real

Faça mock do cliente para testar seu código rapidamente, use um fake pequeno ou um broker embutido quando componentes interagem e deixe as garantias do protocolo para testes de integração contra um broker real. Essa divisão mantém a suíte rápida sem deixar os testes mentirem sobre o que o MQTT realmente faz.

Perguntas frequentes

Devo fazer mock do cliente MQTT ou usar um broker real nos testes?

Os dois, em camadas diferentes. Faça mock do cliente para testar a sua própria lógica (qual topic, qual payload, como você reage a uma mensagem) em milissegundos. Use um broker real nos testes de integração para verificar o comportamento do protocolo: QoS, mensagens retidas, ACLs e reconexões.

O aedes é um mock ou um broker MQTT real?

O aedes é um broker MQTT real escrito em Node.js. Embutido no processo de testes com persistência em memória, ele se comporta como um broker rápido e descartável, então os testes que o usam são pequenos testes de integração, e não testes unitários puros.

Como simular uma mensagem MQTT recebida em um teste unitário?

Chame o handler de mensagens diretamente com um objeto de mensagem falso que tenha os campos que o seu código lê (topic, payload, qos, retain). No paho-mqtt isso significa chamar on_message(client, userdata, msg); no MQTT.js, emitir um evento 'message' em um cliente EventEmitter simulado.

Mock de MQTT em testes unitários: fakes, stubs e brokers em memória