MQTT Tests with Testcontainers (Java, Python, Node)

Testcontainers starts a real broker in Docker for your test run and throws it away afterwards. Every run gets a clean MQTT broker on a random port: no shared state, no leftover retained messages, no "works on my machine" broker version. You need three things — the eclipse-mosquitto image, the mapped port, and a wait strategy that knows when the broker is ready.

This guide shows the same setup in Java (JUnit 5), Python (pytest) and Node.js (Jest), and then the cases where a hosted broker is the better tool.

How it works

1. The test framework starts eclipse-mosquitto:2 before the first test.
2. Docker maps container port 1883 to a random free port on the host.
3. The library waits until the broker logs mosquitto version … running.
4. Your tests connect to host:mappedPort.
5. After the tests, the container is stopped and removed (Testcontainers' Ryuk sidecar also cleans up if the process crashes).

One Mosquitto detail matters everywhere: without a listener in its configuration, Mosquitto 2.x runs in "local only mode" and refuses connections from outside its container. The image contains /mosquitto-no-auth.conf (listener on 1883, anonymous access) for exactly this purpose, so every example below starts the broker with mosquitto -c /mosquitto-no-auth.conf.

Java: JUnit 5 + Paho

Add the dependencies (versions are managed by the 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>

A static container is shared by all tests in the class and started once:

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

If you prefer HiveMQ as the test broker, the org.testcontainers:hivemq module provides a HiveMQContainer with a ready-made wait strategy and an MQTT port accessor. More JUnit patterns — parameterized QoS tests, async clients, Spring Boot — are in MQTT testing with JUnit.

Custom configuration: authentication and ACLs

To test that your application handles Not authorized correctly, give the container a real configuration with a password file and ACLs from your test resources:

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

The configuration is the stock image path, so the default command picks it up. Generate the mosquitto.conf with the Mosquitto config generator and create the password file once with mosquitto_passwd -c passwd testuser.

Python: pytest + testcontainers

Install pip install testcontainers paho-mqtt pytest. A session-scoped fixture starts one broker for the whole test session:

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"

Because the fixture yields only (host, port), the same tests can run against any broker: swap the fixture for one that returns ("broker.connectmq.com", 1883) and nothing else changes. pytest fixtures for MQTT builds on this with environment-driven broker selection.

Node.js: Jest + testcontainers

Install npm i -D testcontainers mqtt jest. Container start-up takes a few seconds, so raise the 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();
  }
});

For a larger suite, start the container once in Jest's globalSetup and pass the URL through an environment variable, instead of starting one per test file.

Testcontainers in CI

GitHub-hosted Ubuntu runners include Docker, so Testcontainers tests run there without extra setup: mvn verify, pytest or npm test is enough. A few points are worth knowing:

• macOS and Windows runners do not provide a Linux Docker engine the way Ubuntu runners do, so run container tests on ubuntu-latest.
• Image pulls cost time on every fresh runner. Pin a specific tag such as eclipse-mosquitto:2.0.20 (whichever release you have tested) so builds are reproducible, and pull it once per job.
• Self-hosted runners in Kubernetes often have no Docker socket. Testcontainers then needs a remote Docker host or Testcontainers Cloud — or the tests fall back to a hosted broker.

Testing other brokers and MQTT 5

Mosquitto 2.x supports MQTT 3.1.1 and MQTT 5, so you can test MQTT 5 features — user properties, message expiry, reason codes — against the same container by switching the client to protocol version 5. To check that your code is portable across brokers, run the same test class against several images: emqx/emqx, hivemq/hivemq-ce and eclipse-mosquitto all accept anonymous connections in their default test setups. Wait strategies differ: EMQX and HiveMQ take longer to start, so Testcontainers' default listening-port wait with a generous start-up timeout is simpler than matching their log lines.

Testcontainers or a hosted test broker?

Testcontainers is excellent for isolated, repeatable integration tests. It is not the right tool for everything:

A common split is Testcontainers for the fast integration layer and a hosted broker for a smaller suite that covers TLS, credentials, ACLs and device-level tests — the approach described in MQTT integration testing and wired into CI in testing MQTT with GitHub Actions.

Verifying your setup

Check the pieces one by one when a container test misbehaves:

1. docker ps while tests run — you should see the Mosquitto container and the testcontainers/ryuk container.
2. Container logs (Testcontainers prints them on start-up failure) should not contain Starting in local only mode.
3. Run a single test with a deliberately wrong expected payload and confirm it fails within the timeout instead of hanging.
4. Point the same tests at the hosted broker and watch testmqtt/tc/# in the online MQTT client to confirm the topics and payloads look as you expect.

Watch test traffic in the browser

Testcontainers gives every test run its own real broker with a few lines of setup in Java, Python or Node — as long as you start Mosquitto with a listener and wait for its "running" log line. Use it for fast, isolated integration tests, and keep a hosted broker for the checks that depend on production TLS, credentials and network paths.

Frequently asked questions