Tests MQTT con Testcontainers (Java, Python, Node)

Testcontainers arranca un broker real en Docker para tu ejecución de tests y lo desecha después. Cada ejecución obtiene un broker MQTT limpio en un puerto aleatorio: sin estado compartido, sin retained messages de sobra y sin versiones de broker del tipo "en mi máquina funciona". Necesitas tres cosas: la imagen eclipse-mosquitto, el puerto mapeado y una estrategia de espera que sepa cuándo está listo el broker.

Esta guía muestra la misma configuración en Java (JUnit 5), Python (pytest) y Node.js (Jest), y después los casos en los que un broker alojado es la mejor herramienta.

Cómo funciona

1. El framework de tests arranca eclipse-mosquitto:2 antes del primer test.
2. Docker mapea el puerto 1883 del contenedor a un puerto libre aleatorio del host.
3. La librería espera hasta que el broker escribe en el log mosquitto version … running.
4. Tus tests se conectan a host:mappedPort.
5. Al terminar, el contenedor se detiene y se elimina (el sidecar Ryuk de Testcontainers también limpia si el proceso se cae).

Un detalle de Mosquitto importa en todos los casos: sin un listener en su configuración, Mosquitto 2.x se ejecuta en "local only mode" y rechaza las conexiones que vienen de fuera de su contenedor. La imagen incluye /mosquitto-no-auth.conf (listener en 1883, acceso anónimo) justo para esto, así que todos los ejemplos siguientes arrancan el broker con mosquitto -c /mosquitto-no-auth.conf.

Java: JUnit 5 + Paho

Añade las dependencias (las versiones las gestiona el BOM de 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>

Un contenedor static se comparte entre todos los tests de la clase y se arranca una sola 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();
  }
}

Si prefieres HiveMQ como broker de pruebas, el módulo org.testcontainers:hivemq ofrece un HiveMQContainer con una estrategia de espera ya preparada y un accesor para el puerto MQTT. Tienes más patrones de JUnit —tests de QoS parametrizados, clientes asíncronos, Spring Boot— en tests MQTT con JUnit.

Configuración propia: autenticación y ACL

Para probar que tu aplicación gestiona correctamente Not authorized, dale al contenedor una configuración real con un archivo de contraseñas y ACL desde tus recursos de test:

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

La configuración está en la ruta estándar de la imagen, así que el comando por defecto la recoge. Genera el mosquitto.conf con el generador de configuración de Mosquitto y crea el archivo de contraseñas una vez con mosquitto_passwd -c passwd testuser.

Python: pytest + testcontainers

Instala pip install testcontainers paho-mqtt pytest. Un fixture con scope de sesión arranca un único broker para toda la sesión de tests:

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 el fixture solo devuelve (host, port), los mismos tests pueden ejecutarse contra cualquier broker: cambia el fixture por uno que devuelva ("broker.connectmq.com", 1883) y no cambia nada más. Fixtures de pytest para MQTT amplía esto con la selección del broker mediante variables de entorno.

Node.js: Jest + testcontainers

Instala npm i -D testcontainers mqtt jest. Arrancar el contenedor lleva unos segundos, así que aumenta el timeout del 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();
  }
});

En una suite más grande, arranca el contenedor una sola vez en el globalSetup de Jest y pasa la URL mediante una variable de entorno, en lugar de arrancar uno por archivo de test.

Testcontainers en CI

Los runners Ubuntu alojados por GitHub incluyen Docker, así que los tests con Testcontainers funcionan ahí sin configuración extra: basta con mvn verify, pytest o npm test. Conviene tener en cuenta algunos puntos:

• Los runners de macOS y Windows no ofrecen un motor Docker de Linux como los runners de Ubuntu, así que ejecuta los tests con contenedores en ubuntu-latest.
• Las descargas de imágenes cuestan tiempo en cada runner nuevo. Fija un tag concreto como eclipse-mosquitto:2.0.20 (la versión que hayas probado) para que los builds sean reproducibles, y descárgala una vez por job.
• Los runners autoalojados en Kubernetes a menudo no tienen socket de Docker. Testcontainers necesita entonces un host Docker remoto o Testcontainers Cloud, o los tests pasan a un broker alojado.

Probar otros brokers y MQTT 5

Mosquitto 2.x soporta MQTT 3.1.1 y MQTT 5, así que puedes probar funciones de MQTT 5 —user properties, message expiry, reason codes— contra el mismo contenedor cambiando el cliente a la versión 5 del protocolo. Para comprobar que tu código es portable entre brokers, ejecuta la misma clase de tests contra varias imágenes: emqx/emqx, hivemq/hivemq-ce y eclipse-mosquitto aceptan conexiones anónimas en sus configuraciones de prueba por defecto. Las estrategias de espera cambian: EMQX y HiveMQ tardan más en arrancar, así que la espera por defecto de Testcontainers sobre el puerto de escucha, con un timeout de arranque generoso, es más sencilla que buscar sus líneas de log.

¿Testcontainers o un broker de pruebas alojado?

Testcontainers es excelente para pruebas de integración aisladas y repetibles. No es la herramienta adecuada para todo:

Un reparto habitual es usar Testcontainers para la capa rápida de integración y un broker alojado para una suite más pequeña que cubra TLS, credenciales, ACL y tests a nivel de dispositivo: el enfoque descrito en pruebas de integración MQTT e integrado en CI en probar MQTT con GitHub Actions.

Verificar tu configuración

Cuando un test con contenedores se porta mal, revisa las piezas una por una:

1. docker ps mientras se ejecutan los tests: deberías ver el contenedor de Mosquitto y el contenedor testcontainers/ryuk.
2. Los logs del contenedor (Testcontainers los imprime si falla el arranque) no deberían contener Starting in local only mode.
3. Ejecuta un único test con un payload esperado incorrecto a propósito y confirma que falla dentro del timeout en lugar de quedarse colgado.
4. Apunta los mismos tests al broker alojado y observa testmqtt/tc/# en el cliente MQTT online para confirmar que los topics y los payloads son los que esperas.

Ver el tráfico de los tests en el navegador

Testcontainers da a cada ejecución de tests su propio broker real con unas pocas líneas de configuración en Java, Python o Node, siempre que arranques Mosquitto con un listener y esperes a su línea de log "running". Úsalo para pruebas de integración rápidas y aisladas, y reserva un broker alojado para las comprobaciones que dependen del TLS, las credenciales y las rutas de red de producción.

Preguntas frecuentes