Tests MQTT avec Testcontainers (Java, Python, Node)

Testcontainers démarre un vrai broker dans Docker pour votre exécution de tests et le supprime ensuite. Chaque exécution obtient un broker MQTT propre sur un port aléatoire : aucun état partagé, aucun retained message résiduel, aucune version de broker « qui marche sur ma machine ». Il vous faut trois choses : l'image eclipse-mosquitto, le port mappé et une stratégie d'attente qui sait quand le broker est prêt.

Ce guide montre la même configuration en Java (JUnit 5), Python (pytest) et Node.js (Jest), puis les cas où un broker hébergé est le meilleur outil.

Comment ça marche

1. Le framework de test démarre eclipse-mosquitto:2 avant le premier test.
2. Docker mappe le port 1883 du conteneur sur un port libre aléatoire de l'hôte.
3. La bibliothèque attend que le broker journalise mosquitto version … running.
4. Vos tests se connectent à host:mappedPort.
5. Après les tests, le conteneur est arrêté et supprimé (le sidecar Ryuk de Testcontainers fait aussi le ménage si le processus plante).

Un détail de Mosquitto compte partout : sans listener dans sa configuration, Mosquitto 2.x fonctionne en « local only mode » et refuse les connexions venant de l'extérieur de son conteneur. L'image contient /mosquitto-no-auth.conf (listener sur 1883, accès anonyme) précisément pour ce cas : tous les exemples ci-dessous démarrent donc le broker avec mosquitto -c /mosquitto-no-auth.conf.

Java : JUnit 5 + Paho

Ajoutez les dépendances (les versions sont gérées par le BOM 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 conteneur static est partagé par tous les tests de la classe et démarré une seule fois :

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 vous préférez HiveMQ comme broker de test, le module org.testcontainers:hivemq fournit un HiveMQContainer avec une stratégie d'attente prête à l'emploi et un accesseur pour le port MQTT. D'autres modèles JUnit — tests QoS paramétrés, clients asynchrones, Spring Boot — sont présentés dans tests MQTT avec JUnit.

Configuration personnalisée : authentification et ACL

Pour vérifier que votre application gère correctement Not authorized, donnez au conteneur une vraie configuration avec un fichier de mots de passe et des ACL issus de vos ressources 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 configuration se trouve au chemin standard de l'image, la commande par défaut la charge donc. Générez le mosquitto.conf avec le générateur de configuration Mosquitto et créez une fois le fichier de mots de passe avec mosquitto_passwd -c passwd testuser.

Python : pytest + testcontainers

Installez pip install testcontainers paho-mqtt pytest. Une fixture à portée de session démarre un seul broker pour toute la session 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"

Comme la fixture ne renvoie que (host, port), les mêmes tests peuvent tourner contre n'importe quel broker : remplacez la fixture par une autre qui renvoie ("broker.connectmq.com", 1883) et rien d'autre ne change. Fixtures pytest pour MQTT va plus loin avec une sélection du broker par variables d'environnement.

Node.js : Jest + testcontainers

Installez npm i -D testcontainers mqtt jest. Le démarrage du conteneur prend quelques secondes : augmentez donc le timeout du 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();
  }
});

Pour une suite plus importante, démarrez le conteneur une seule fois dans le globalSetup de Jest et transmettez l'URL via une variable d'environnement, au lieu d'en démarrer un par fichier de test.

Testcontainers en CI

Les runners Ubuntu hébergés par GitHub incluent Docker : les tests Testcontainers y tournent sans configuration supplémentaire, mvn verify, pytest ou npm test suffisent. Quelques points à connaître :

• Les runners macOS et Windows ne fournissent pas de moteur Docker Linux comme les runners Ubuntu : exécutez donc les tests avec conteneurs sur ubuntu-latest.
• Le téléchargement des images prend du temps sur chaque nouveau runner. Épinglez un tag précis comme eclipse-mosquitto:2.0.20 (la version que vous avez testée) pour des builds reproductibles, et téléchargez-la une fois par job.
• Les runners auto-hébergés dans Kubernetes n'ont souvent pas de socket Docker. Testcontainers a alors besoin d'un hôte Docker distant ou de Testcontainers Cloud — sinon les tests basculent sur un broker hébergé.

Tester d'autres brokers et MQTT 5

Mosquitto 2.x prend en charge MQTT 3.1.1 et MQTT 5 : vous pouvez donc tester les fonctionnalités de MQTT 5 — user properties, message expiry, reason codes — sur le même conteneur en passant le client en version 5 du protocole. Pour vérifier que votre code est portable d'un broker à l'autre, exécutez la même classe de tests contre plusieurs images : emqx/emqx, hivemq/hivemq-ce et eclipse-mosquitto acceptent les connexions anonymes dans leurs configurations de test par défaut. Les stratégies d'attente diffèrent : EMQX et HiveMQ démarrent plus lentement, l'attente par défaut de Testcontainers sur le port d'écoute, avec un timeout de démarrage généreux, est donc plus simple que de faire correspondre leurs lignes de log.

Testcontainers ou un broker de test hébergé ?

Testcontainers est excellent pour des tests d'intégration isolés et reproductibles. Ce n'est pas le bon outil pour tout :

Une répartition courante : Testcontainers pour la couche d'intégration rapide, et un broker hébergé pour une suite plus réduite couvrant le TLS, les identifiants, les ACL et les tests au niveau des appareils — l'approche décrite dans tests d'intégration MQTT et branchée sur la CI dans tester MQTT avec GitHub Actions.

Vérifier votre configuration

Quand un test avec conteneur se comporte mal, vérifiez les éléments un par un :

1. docker ps pendant l'exécution des tests — vous devez voir le conteneur Mosquitto et le conteneur testcontainers/ryuk.
2. Les logs du conteneur (Testcontainers les affiche en cas d'échec au démarrage) ne doivent pas contenir Starting in local only mode.
3. Lancez un seul test avec un payload attendu volontairement faux et vérifiez qu'il échoue dans le délai imparti au lieu de rester bloqué.
4. Pointez les mêmes tests vers le broker hébergé et observez testmqtt/tc/# dans le client MQTT en ligne pour confirmer que les topics et les payloads sont conformes à vos attentes.

Observer le trafic des tests dans le navigateur

Testcontainers offre à chaque exécution de tests son propre vrai broker en quelques lignes de configuration en Java, Python ou Node — à condition de démarrer Mosquitto avec un listener et d'attendre sa ligne de log « running ». Utilisez-le pour des tests d'intégration rapides et isolés, et gardez un broker hébergé pour les vérifications qui dépendent du TLS, des identifiants et des chemins réseau de production.

Questions fréquentes