MQTT-Tests mit Testcontainers (Java, Python, Node)

Testcontainers startet für deinen Testlauf einen echten Broker in Docker und wirft ihn danach weg. Jeder Lauf bekommt einen sauberen MQTT-Broker auf einem zufälligen Port: kein geteilter Zustand, keine übrig gebliebenen Retained Messages, keine „läuft auf meinem Rechner"-Broker-Version. Du brauchst drei Dinge – das eclipse-mosquitto-Image, den gemappten Port und eine Wait-Strategie, die weiß, wann der Broker bereit ist.

Diese Anleitung zeigt dasselbe Setup in Java (JUnit 5), Python (pytest) und Node.js (Jest) und danach die Fälle, in denen ein gehosteter Broker das bessere Werkzeug ist.

So funktioniert es

1. Das Test-Framework startet eclipse-mosquitto:2 vor dem ersten Test.
2. Docker mappt den Container-Port 1883 auf einen zufälligen freien Port auf dem Host.
3. Die Bibliothek wartet, bis der Broker mosquitto version … running loggt.
4. Deine Tests verbinden sich mit host:mappedPort.
5. Nach den Tests wird der Container gestoppt und entfernt (der Ryuk-Sidecar von Testcontainers räumt auch auf, wenn der Prozess abstürzt).

Ein Mosquitto-Detail ist überall wichtig: Ohne Listener in der Konfiguration läuft Mosquitto 2.x im „Local Only Mode" und lehnt Verbindungen von außerhalb seines Containers ab. Das Image enthält genau dafür /mosquitto-no-auth.conf (Listener auf 1883, anonymer Zugriff), deshalb starten alle folgenden Beispiele den Broker mit mosquitto -c /mosquitto-no-auth.conf.

Java: JUnit 5 + Paho

Füge die Abhängigkeiten hinzu (die Versionen verwaltet die Testcontainers-BOM):

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

Ein static-Container wird von allen Tests der Klasse geteilt und nur einmal gestartet:

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

Wenn du HiveMQ als Test-Broker bevorzugst, bietet das Modul org.testcontainers:hivemq einen HiveMQContainer mit fertiger Wait-Strategie und einem Accessor für den MQTT-Port. Weitere JUnit-Muster – parametrisierte QoS-Tests, asynchrone Clients, Spring Boot – findest du in MQTT-Tests mit JUnit.

Eigene Konfiguration: Authentifizierung und ACLs

Um zu testen, ob deine Anwendung Not authorized korrekt behandelt, gib dem Container eine echte Konfiguration mit Passwortdatei und ACLs aus deinen Test-Ressourcen:

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

Die Konfiguration liegt am Standardpfad des Images, sodass der Standardbefehl sie lädt. Erzeuge die mosquitto.conf mit dem Mosquitto-Config-Generator und lege die Passwortdatei einmalig mit mosquitto_passwd -c passwd testuser an.

Python: pytest + testcontainers

Installiere pip install testcontainers paho-mqtt pytest. Eine Fixture mit Session-Scope startet einen einzigen Broker für die gesamte Testsitzung:

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"

Da die Fixture nur (host, port) liefert, können dieselben Tests gegen jeden Broker laufen: Tausche die Fixture gegen eine, die ("broker.connectmq.com", 1883) zurückgibt, und sonst ändert sich nichts. pytest-Fixtures für MQTT baut darauf mit Broker-Auswahl über Umgebungsvariablen auf.

Node.js: Jest + testcontainers

Installiere npm i -D testcontainers mqtt jest. Der Container-Start dauert ein paar Sekunden, also erhöhe den Hook-Timeout:

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

Bei einer größeren Suite startest du den Container einmal in Jests globalSetup und reichst die URL über eine Umgebungsvariable weiter, statt pro Testdatei einen zu starten.

Testcontainers in CI

Die von GitHub gehosteten Ubuntu-Runner enthalten Docker, sodass Testcontainers-Tests dort ohne zusätzliches Setup laufen: mvn verify, pytest oder npm test reichen. Ein paar Punkte solltest du kennen:

• macOS- und Windows-Runner stellen keine Linux-Docker-Engine bereit wie die Ubuntu-Runner, also führe Container-Tests auf ubuntu-latest aus.
• Image-Pulls kosten auf jedem frischen Runner Zeit. Pinne einen konkreten Tag wie eclipse-mosquitto:2.0.20 (das Release, das du getestet hast), damit Builds reproduzierbar sind, und ziehe es einmal pro Job.
• Selbst gehostete Runner in Kubernetes haben oft keinen Docker-Socket. Testcontainers braucht dann einen entfernten Docker-Host oder Testcontainers Cloud – oder die Tests weichen auf einen gehosteten Broker aus.

Andere Broker und MQTT 5 testen

Mosquitto 2.x unterstützt MQTT 3.1.1 und MQTT 5, du kannst also MQTT-5-Features – User Properties, Message Expiry, Reason Codes – gegen denselben Container testen, indem du den Client auf Protokollversion 5 umstellst. Um zu prüfen, ob dein Code zwischen Brokern portabel ist, lass dieselbe Testklasse gegen mehrere Images laufen: emqx/emqx, hivemq/hivemq-ce und eclipse-mosquitto akzeptieren in ihren Standard-Test-Setups anonyme Verbindungen. Die Wait-Strategien unterscheiden sich: EMQX und HiveMQ brauchen länger zum Starten, daher ist das Standard-Warten von Testcontainers auf den Listening-Port mit großzügigem Start-Timeout einfacher, als ihre Log-Zeilen abzugleichen.

Testcontainers oder ein gehosteter Test-Broker?

Testcontainers ist hervorragend für isolierte, wiederholbare Integrationstests. Es ist aber nicht für alles das richtige Werkzeug:

Eine verbreitete Aufteilung ist Testcontainers für die schnelle Integrationsebene und ein gehosteter Broker für eine kleinere Suite, die TLS, Zugangsdaten, ACLs und Tests auf Geräteebene abdeckt – der Ansatz aus MQTT-Integrationstests, in CI verdrahtet in MQTT mit GitHub Actions testen.

Dein Setup überprüfen

Wenn sich ein Container-Test seltsam verhält, prüfe die Bausteine einzeln:

1. docker ps, während die Tests laufen – du solltest den Mosquitto-Container und den testcontainers/ryuk-Container sehen.
2. Die Container-Logs (Testcontainers gibt sie bei Startfehlern aus) sollten kein Starting in local only mode enthalten.
3. Führe einen einzelnen Test mit absichtlich falscher erwarteter Payload aus und bestätige, dass er innerhalb des Timeouts fehlschlägt, statt zu hängen.
4. Richte dieselben Tests auf den gehosteten Broker und beobachte testmqtt/tc/# im Online-MQTT-Client, um zu bestätigen, dass Topics und Payloads so aussehen wie erwartet.

Test-Traffic im Browser ansehen

Testcontainers gibt jedem Testlauf mit wenigen Zeilen Setup in Java, Python oder Node seinen eigenen echten Broker – solange du Mosquitto mit einem Listener startest und auf seine „running"-Logzeile wartest. Nutze es für schnelle, isolierte Integrationstests und behalte einen gehosteten Broker für die Checks, die von Produktions-TLS, Zugangsdaten und Netzwerkpfaden abhängen.

Häufig gestellte Fragen