Tests d'intégration MQTT en Java avec JUnit 5

Un test d'intégration MQTT en Java connecte un vrai client à un vrai broker, s'abonne, publie et vérifie ce qui arrive. Avec JUnit 5, la difficulté n'est pas l'API mais la livraison asynchrone : les messages arrivent sur un autre thread, un peu plus tard, ou jamais. Ce guide présente une configuration de tests qui gère cela correctement avec Eclipse Paho, le HiveMQ MQTT Client, Awaitility et Testcontainers.

Dépendances

Les exemples utilisent Maven ; les mêmes coordonnées fonctionnent avec Gradle. Les versions étaient à jour au moment de la rédaction — des versions correctives plus récentes conviennent.

pom.xml (test dependencies)

<dependencies>
  <dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
  </dependency>
  <dependency>
    <groupId>com.hivemq</groupId>
    <artifactId>hivemq-mqtt-client</artifactId>
    <version>1.3.3</version>
  </dependency>
  <dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <version>4.2.2</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>1.20.3</version>
    <scope>test</scope>
  </dependency>
</dependencies>

Vous n'avez besoin que du client MQTT réellement utilisé par votre application. Si vous découvrez ces bibliothèques, commencez par MQTT en Java avec Paho ou le guide du HiveMQ MQTT Client.

Règles pour des tests MQTT stables

Règle	Pourquoi
Client ID unique par test	Deux clients avec le même ID se déconnectent mutuellement ("session taken over")
Préfixe de topic unique par test	Ni les tests parallèles ni d'anciens messages retenus ne polluent vos assertions
S'abonner avant de publier et attendre le SUBACK	Un message publié avant que l'abonnement soit actif n'est tout simplement pas livré
Attendre avec un timeout, jamais avec `Thread.sleep`	Rapide quand ça marche, un échec clair quand ça ne marche pas
Se déconnecter dans `@AfterEach`	Les connexions oubliées épuisent les limites du broker et maintiennent des sessions
Adresse du broker issue de la configuration	Les mêmes tests tournent contre Testcontainers en local et un broker hébergé en CI

Le générateur de client IDs montre des formats sûrs pour tous les brokers ; en test, un préfixe suivi d'un suffixe aléatoire suffit.

Un test Paho avec Awaitility

Le MqttClient synchrone de Paho bloque connect() et subscribe() jusqu'à ce que le broker les ait acquittés, ce qui simplifie la préparation du test. La livraison a toujours lieu sur le thread de callbacks de Paho : collectez donc les messages dans une file thread-safe et laissez Awaitility l'interroger.

PahoRoundTripTest.java

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;

import java.time.Duration;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.*;

class PahoRoundTripTest {

  static final String BROKER =
      System.getenv().getOrDefault("MQTT_URL", "tcp://broker.connectmq.com:1883");

  MqttClient client;
  String topic;

  @BeforeEach
  void connect() throws MqttException {
    String id = UUID.randomUUID().toString().substring(0, 8);
    topic = "testmqtt/junit/" + id + "/temperature";
    client = new MqttClient(BROKER, "junit-" + id, new MemoryPersistence());

    MqttConnectOptions options = new MqttConnectOptions();
    options.setCleanSession(true);
    options.setConnectionTimeout(10);
    String user = System.getenv("MQTT_USER");
    if (user != null) {
      options.setUserName(user);
      options.setPassword(System.getenv("MQTT_PASS").toCharArray());
    }
    client.connect(options);
  }

  @AfterEach
  void disconnect() throws MqttException {
    if (client.isConnected()) client.disconnect();
    client.close();
  }

  @Test
  void publishedMessageIsReceived() throws MqttException {
    var received = new ConcurrentLinkedQueue<String>();
    client.subscribe(topic, 1, (t, msg) -> received.add(new String(msg.getPayload(), UTF_8)));

    client.publish(topic, "21.5".getBytes(UTF_8), 1, false);

    await().atMost(Duration.ofSeconds(5))
        .untilAsserted(() -> assertEquals("21.5", received.peek()));
  }
}

MemoryPersistence compte : la persistance fichier par défaut crée un dossier par client ID dans le répertoire de travail, ce qui encombre votre répertoire de build et peut provoquer des collisions entre exécutions parallèles.

Le même test avec le HiveMQ MQTT Client

Le client HiveMQ prend en charge MQTT 5. Son API bloquante propose une file publishes() idéale pour les tests : enregistrez-la avant de vous abonner, puis appelez receive() avec un timeout.

HiveMqRoundTripTest.java

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

import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.*;

class HiveMqRoundTripTest {

  Mqtt5BlockingClient client;
  String topic;

  @BeforeEach
  void connect() {
    String id = UUID.randomUUID().toString().substring(0, 8);
    topic = "testmqtt/junit/" + id + "/status";
    client = Mqtt5Client.builder()
        .identifier("junit-" + id)
        .serverHost("broker.connectmq.com")
        .serverPort(1883)
        .buildBlocking();
    client.connect();
  }

  @AfterEach
  void disconnect() {
    client.disconnect();
  }

  @Test
  void statusMessageIsDelivered() throws InterruptedException {
    try (var publishes = client.publishes(MqttGlobalPublishFilter.ALL)) {
      client.subscribeWith().topicFilter(topic).qos(MqttQos.AT_LEAST_ONCE).send();
      client.publishWith().topic(topic).qos(MqttQos.AT_LEAST_ONCE)
          .payload("online".getBytes(UTF_8)).send();

      Optional<Mqtt5Publish> msg = publishes.receive(5, TimeUnit.SECONDS);
      assertTrue(msg.isPresent(), "no message within 5 s on " + topic);
      assertEquals("online", new String(msg.get().getPayloadAsBytes(), UTF_8));
    }
  }
}

Pour TLS, ajoutez .sslWithDefaultConfig() et le port 8883 ; pour les identifiants, ajoutez .simpleAuth().username(...).password(...).applySimpleAuth() à l'appel de connexion.

Un broker jetable avec Testcontainers

Pour les tests qui doivent tourner hors ligne ou nécessitent une configuration de broker précise, démarrez un conteneur de broker par classe de test. Sans configuration, Mosquitto 2.x n'écoute que sur localhost : utilisez donc la configuration sans authentification fournie dans l'image.

LocalBrokerTest.java

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

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

  static String brokerUrl() {
    return "tcp://" + mosquitto.getHost() + ":" + mosquitto.getMappedPort(1883);
  }

  @Test
  void brokerAcceptsConnections() throws Exception {
    var client = new org.eclipse.paho.client.mqttv3.MqttClient(
        brokerUrl(), "junit-local",
        new org.eclipse.paho.client.mqttv3.persist.MemoryPersistence());
    client.connect();
    client.disconnect();
    client.close();
  }
}

Un conteneur static démarre une fois par classe ; un champ d'instance en démarrerait un par méthode de test, ce qui est lent. Testcontainers fournit aussi un module HiveMQ (org.testcontainers:hivemq) si vous voulez tester contre HiveMQ CE. Les compromis sont détaillés dans tests MQTT avec Testcontainers.

Conteneur local ou broker hébergé ?

Un conteneur prouve que votre code fonctionne avec un broker. Il ne prouve pas qu'il fonctionne avec votre broker — celui qui a TLS, l'authentification, les ACL et le chemin réseau de vos appareils. Un bon découpage :

Tests unitaires et de composants : Testcontainers, aucune dépendance réseau, retour rapide.
Tests d'intégration en CI : un broker de test hébergé avec la même configuration TLS et d'identifiants que la production, sélectionné via MQTT_URL, MQTT_USER et MQTT_PASS.
Smoke tests après déploiement : le broker de production, un topic de test dédié, des assertions en lecture seule.

Tester les cas négatifs

Les tests d'intégration doivent aussi prouver ce qui ne doit pas arriver. Deux patterns :

Échec de connexion attendu. Avec Paho, assertThrows(MqttSecurityException.class, () -> client.connect(badOptions)) vérifie que de mauvais identifiants sont refusés ; getReasonCode() renvoie 4 ou 5 pour "bad username or password" et "not authorized" en MQTT 3.1.1. Consultez tout autre code avec l'outil des reason codes.
Silence attendu. Pour vérifier qu'une ACL bloque un topic, abonnez-vous, attendez un temps borné puis vérifiez que la file reste vide : await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(3)).until(received::isEmpty). during fait vérifier à Awaitility que la condition reste vraie.

Exécuter la suite

Marquez les tests dépendant du réseau avec @Tag("integration") et exécutez-les avec Maven Failsafe (*IT.java) ou une tâche Gradle séparée, pour que mvn test reste rapide et hors ligne. En CI, exportez les variables du broker et lancez mvn verify ; les tests MQTT dans GitHub Actions présentent un workflow complet.

Pendant qu'un test tourne, ouvrez le client MQTT en ligne et abonnez-vous à testmqtt/junit/# pour voir le trafic généré par votre suite — le moyen le plus rapide de savoir si un test en échec n'a jamais publié ou a publié sur le mauvais topic.

Suivre en direct le trafic de vos tests JUnit

De bons tests MQTT avec JUnit 5 reposent sur l'isolation et la patience : un client ID et un topic uniques par test, un abonnement acquitté avant de publier et des attentes bornées plutôt que des sleeps. Gardez l'adresse du broker configurable et la même suite couvrira un conteneur sur votre portable, un broker de CI et un smoke test après déploiement. Pour une stratégie plus large, consultez les tests d'intégration MQTT.

Questions fréquentes

Faut-il utiliser Thread.sleep dans les tests MQTT ?

Non. Un sleep fixe est trop court sur un runner CI lent ou fait perdre du temps sur un runner rapide. Utilisez Awaitility ou un CompletableFuture avec timeout : le test continue dès que le message arrive et échoue avec un message clair s'il n'arrive pas.

Paho ou client HiveMQ pour les tests JUnit ?

Testez avec le client qu'utilise votre application. Eclipse Paho (mqttv3) est le plus répandu pour MQTT 3.1.1 ; le HiveMQ MQTT Client prend en charge MQTT 5 et propose des API bloquante, asynchrone et réactive. Les patterns de ce guide fonctionnent avec les deux.

Comment éviter que des tests MQTT parallèles interfèrent ?

Donnez à chaque test un client ID unique et un préfixe de topic unique, par exemple testmqtt/junit/<id-aleatoire>/. Les tests peuvent alors tourner en parallèle sur le même broker sans recevoir les messages des autres ni se déconnecter mutuellement à cause de client IDs dupliqués.