Testes MQTT com Testcontainers (Java, Python, Node)

O Testcontainers sobe um broker real no Docker para a sua execução de testes e o descarta depois. Cada execução ganha um broker MQTT limpo em uma porta aleatória: sem estado compartilhado, sem retained messages sobrando, sem versão de broker do tipo "na minha máquina funciona". Você precisa de três coisas: a imagem eclipse-mosquitto, a porta mapeada e uma estratégia de espera que saiba quando o broker está pronto.

Este guia mostra a mesma configuração em Java (JUnit 5), Python (pytest) e Node.js (Jest), e depois os casos em que um broker hospedado é a ferramenta melhor.

Como funciona

1. O framework de testes inicia eclipse-mosquitto:2 antes do primeiro teste.
2. O Docker mapeia a porta 1883 do container para uma porta livre aleatória no host.
3. A biblioteca espera até o broker registrar mosquitto version … running no log.
4. Seus testes conectam em host:mappedPort.
5. Depois dos testes, o container é parado e removido (o sidecar Ryuk do Testcontainers também faz a limpeza se o processo cair).

Um detalhe do Mosquitto importa em todos os casos: sem um listener na configuração, o Mosquitto 2.x roda em "local only mode" e recusa conexões vindas de fora do container. A imagem traz /mosquitto-no-auth.conf (listener na 1883, acesso anônimo) exatamente para isso, então todos os exemplos abaixo iniciam o broker com mosquitto -c /mosquitto-no-auth.conf.

Java: JUnit 5 + Paho

Adicione as dependências (as versões são gerenciadas pelo BOM do Testcontainers):

<dependency>
  <groupId>org.testcontainers</groupId>
  <artifactId>junit-jupiter</artifactId>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.eclipse.paho</groupId>
  <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  <version>1.2.5</version>
  <scope>test</scope>
</dependency>

Um container static é compartilhado por todos os testes da classe e iniciado uma única vez:

package com.example;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.jupiter.api.Assertions.*;

import java.util.UUID;
import java.util.concurrent.*;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

@Testcontainers
class TelemetryIT {

  @Container
  static GenericContainer<?> mosquitto =
      new GenericContainer<>(DockerImageName.parse("eclipse-mosquitto:2"))
          .withExposedPorts(1883)
          .withCommand("mosquitto", "-c", "/mosquitto-no-auth.conf")
          .waitingFor(Wait.forLogMessage(".*mosquitto version .* running.*\\n", 1));

  @Test
  void publishedTelemetryReachesSubscriber() throws Exception {
    String uri = "tcp://" + mosquitto.getHost() + ":" + mosquitto.getMappedPort(1883);
    String topic = "testmqtt/tc/" + UUID.randomUUID() + "/telemetry";

    MqttClient client = new MqttClient(uri, MqttClient.generateClientId(), new MemoryPersistence());
    client.connect();

    BlockingQueue<String> received = new LinkedBlockingQueue<>();
    // synchronous subscribe: returns after the SUBACK
    client.subscribe(topic, 1, (t, msg) -> received.add(new String(msg.getPayload(), UTF_8)));
    client.publish(topic, "{\"temp\":21.5}".getBytes(UTF_8), 1, false);

    assertEquals("{\"temp\":21.5}", received.poll(5, TimeUnit.SECONDS));
    client.disconnect();
    client.close();
  }
}

Se você prefere o HiveMQ como broker de testes, o módulo org.testcontainers:hivemq oferece um HiveMQContainer com estratégia de espera pronta e um acessor para a porta MQTT. Mais padrões de JUnit — testes de QoS parametrizados, clientes assíncronos, Spring Boot — estão em testes MQTT com JUnit.

Configuração própria: autenticação e ACLs

Para testar se a sua aplicação lida corretamente com Not authorized, dê ao container uma configuração real com arquivo de senhas e ACLs a partir dos seus recursos de teste:

// import org.testcontainers.utility.MountableFile;
static GenericContainer<?> secured =
    new GenericContainer<>(DockerImageName.parse("eclipse-mosquitto:2"))
        .withExposedPorts(1883)
        .withCopyFileToContainer(MountableFile.forClasspathResource("mosquitto/mosquitto.conf"), "/mosquitto/config/mosquitto.conf")
        .withCopyFileToContainer(MountableFile.forClasspathResource("mosquitto/passwd"), "/mosquitto/config/passwd")
        .withCopyFileToContainer(MountableFile.forClasspathResource("mosquitto/acl"), "/mosquitto/config/acl")
        .waitingFor(Wait.forLogMessage(".*mosquitto version .* running.*\\n", 1));

A configuração fica no caminho padrão da imagem, então o comando padrão a carrega. Gere o mosquitto.conf com o gerador de configuração do Mosquitto e crie o arquivo de senhas uma vez com mosquitto_passwd -c passwd testuser.

Python: pytest + testcontainers

Instale com pip install testcontainers paho-mqtt pytest. Uma fixture com escopo de sessão sobe um único broker para toda a sessão de testes:

import uuid
import pytest
import paho.mqtt.client as mqtt
from testcontainers.core.container import DockerContainer
from testcontainers.core.waiting_utils import wait_for_logs


@pytest.fixture(scope="session")
def broker():
    container = (
        DockerContainer("eclipse-mosquitto:2")
        .with_exposed_ports(1883)
        .with_command("mosquitto -c /mosquitto-no-auth.conf")
    )
    with container:
        wait_for_logs(container, r"mosquitto version .* running", timeout=30)
        yield container.get_container_host_ip(), int(container.get_exposed_port(1883))


@pytest.fixture
def client(broker):
    host, port = broker
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"tc-{uuid.uuid4().hex[:8]}")
    c.connect(host, port)
    c.loop_start()
    yield c
    c.disconnect()
    c.loop_stop()

import queue, time, uuid


def test_round_trip(client):
    topic = f"testmqtt/tc/{uuid.uuid4().hex[:8]}/echo"
    inbox, subscribed = queue.Queue(), queue.Queue()
    client.on_message = lambda c, u, msg: inbox.put(msg.payload.decode())
    client.on_subscribe = lambda c, u, mid, codes, props: subscribed.put(codes[0])

    client.subscribe(topic, qos=1)
    assert not subscribed.get(timeout=5).is_failure   # wait for SUBACK

    client.publish(topic, "hello", qos=1)
    assert inbox.get(timeout=5) == "hello"

Como a fixture retorna apenas (host, port), os mesmos testes podem rodar contra qualquer broker: troque a fixture por uma que retorne ("broker.connectmq.com", 1883) e nada mais muda. Fixtures do pytest para MQTT expande isso com seleção de broker por variáveis de ambiente.

Node.js: Jest + testcontainers

Instale com npm i -D testcontainers mqtt jest. A inicialização do container leva alguns segundos, então aumente o timeout do hook:

const { GenericContainer, Wait } = require('testcontainers');
const mqtt = require('mqtt');
const { randomUUID } = require('node:crypto');

let container;
let url;

beforeAll(async () => {
  container = await new GenericContainer('eclipse-mosquitto:2')
    .withExposedPorts(1883)
    .withCommand(['mosquitto', '-c', '/mosquitto-no-auth.conf'])
    .withWaitStrategy(Wait.forLogMessage(/mosquitto version .* running/))
    .start();
  url = `mqtt://${container.getHost()}:${container.getMappedPort(1883)}`;
}, 60000);

afterAll(async () => {
  await container?.stop();
});

test('round trip through a disposable broker', async () => {
  const topic = `testmqtt/tc/${randomUUID().slice(0, 8)}/echo`;
  const client = await mqtt.connectAsync(url, { clientId: `tc-${randomUUID().slice(0, 8)}` });
  try {
    const received = new Promise((resolve) => client.once('message', (_t, p) => resolve(p.toString())));
    await client.subscribeAsync(topic, { qos: 1 });
    await client.publishAsync(topic, 'hello', { qos: 1 });
    await expect(received).resolves.toBe('hello');
  } finally {
    await client.endAsync();
  }
});

Em uma suíte maior, suba o container uma única vez no globalSetup do Jest e passe a URL por uma variável de ambiente, em vez de subir um por arquivo de teste.

Testcontainers no CI

Os runners Ubuntu hospedados pelo GitHub incluem Docker, então testes com Testcontainers rodam lá sem configuração extra: mvn verify, pytest ou npm test bastam. Alguns pontos merecem atenção:

• Runners macOS e Windows não oferecem um engine Docker Linux como os runners Ubuntu, então rode os testes com containers em ubuntu-latest.
• Downloads de imagens custam tempo em todo runner novo. Fixe uma tag específica como eclipse-mosquitto:2.0.20 (a versão que você testou) para builds reproduzíveis, e baixe-a uma vez por job.
• Runners auto-hospedados no Kubernetes muitas vezes não têm socket do Docker. O Testcontainers então precisa de um host Docker remoto ou do Testcontainers Cloud — ou os testes passam a usar um broker hospedado.

Testando outros brokers e MQTT 5

O Mosquitto 2.x suporta MQTT 3.1.1 e MQTT 5, então você pode testar recursos do MQTT 5 — user properties, message expiry, reason codes — no mesmo container, trocando o cliente para a versão 5 do protocolo. Para verificar se o seu código é portável entre brokers, rode a mesma classe de testes contra várias imagens: emqx/emqx, hivemq/hivemq-ce e eclipse-mosquitto aceitam conexões anônimas nas configurações de teste padrão. As estratégias de espera variam: EMQX e HiveMQ demoram mais para iniciar, então a espera padrão do Testcontainers pela porta de escuta, com um timeout de inicialização generoso, é mais simples do que casar com as linhas de log deles.

Testcontainers ou um broker de testes hospedado?

O Testcontainers é excelente para testes de integração isolados e repetíveis. Ele não é a ferramenta certa para tudo:

Uma divisão comum é Testcontainers para a camada rápida de integração e um broker hospedado para uma suíte menor que cobre TLS, credenciais, ACLs e testes no nível de dispositivo — a abordagem descrita em testes de integração MQTT e conectada ao CI em testando MQTT com GitHub Actions.

Verificando sua configuração

Quando um teste com container se comportar mal, verifique as peças uma a uma:

1. docker ps enquanto os testes rodam: você deve ver o container do Mosquitto e o container testcontainers/ryuk.
2. Os logs do container (o Testcontainers os imprime em caso de falha na inicialização) não devem conter Starting in local only mode.
3. Rode um único teste com um payload esperado propositalmente errado e confirme que ele falha dentro do timeout em vez de travar.
4. Aponte os mesmos testes para o broker hospedado e acompanhe testmqtt/tc/# no cliente MQTT online para confirmar que topics e payloads estão como você espera.

Veja o tráfego dos testes no navegador

O Testcontainers dá a cada execução de testes um broker real próprio com poucas linhas de configuração em Java, Python ou Node — desde que você inicie o Mosquitto com um listener e espere pela linha de log "running". Use-o para testes de integração rápidos e isolados, e mantenha um broker hospedado para as verificações que dependem do TLS, das credenciais e dos caminhos de rede de produção.

Perguntas frequentes