Testcontainers ile MQTT Testleri (Java, Python, Node)

Testcontainers, test çalıştırmanız için Docker'da gerçek bir broker başlatır ve ardından onu atar. Her çalıştırma rastgele bir portta temiz bir MQTT broker'ı alır: paylaşılan durum yok, artakalan retained mesaj yok, "benim makinemde çalışıyor" türü broker sürümü yok. Üç şeye ihtiyacınız var: eclipse-mosquitto image'ı, eşlenmiş port ve broker'ın ne zaman hazır olduğunu bilen bir bekleme stratejisi.

Bu rehber aynı kurulumu Java (JUnit 5), Python (pytest) ve Node.js (Jest) ile gösteriyor; ardından barındırılan bir broker'ın daha iyi araç olduğu durumları ele alıyor.

Nasıl çalışır

1. Test framework'ü ilk testten önce eclipse-mosquitto:2'yi başlatır.
2. Docker, container'ın 1883 portunu host üzerindeki rastgele boş bir porta eşler.
3. Kütüphane, broker mosquitto version … running loglayana kadar bekler.
4. Testleriniz host:mappedPort adresine bağlanır.
5. Testlerden sonra container durdurulur ve kaldırılır (süreç çökerse Testcontainers'ın Ryuk sidecar'ı da temizlik yapar).

Bir Mosquitto ayrıntısı her yerde önemlidir: yapılandırmasında listener olmadan Mosquitto 2.x "local only mode"da çalışır ve container'ı dışından gelen bağlantıları reddeder. Image tam da bu amaçla /mosquitto-no-auth.conf dosyasını (1883'te listener, anonim erişim) içerir; bu yüzden aşağıdaki tüm örnekler broker'ı mosquitto -c /mosquitto-no-auth.conf ile başlatır.

Java: JUnit 5 + Paho

Bağımlılıkları ekleyin (sürümler Testcontainers BOM tarafından yönetilir):

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

static bir container, sınıftaki tüm testler tarafından paylaşılır ve yalnızca bir kez başlatılır:

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

Test broker'ı olarak HiveMQ'yu tercih ediyorsanız org.testcontainers:hivemq modülü, hazır bir bekleme stratejisine ve MQTT portu için bir erişim metoduna sahip bir HiveMQContainer sunar. Parametreli QoS testleri, asenkron istemciler ve Spring Boot gibi daha fazla JUnit kalıbı JUnit ile MQTT testi yazısında.

Özel yapılandırma: kimlik doğrulama ve ACL'ler

Uygulamanızın Not authorized durumunu doğru ele aldığını test etmek için container'a, test kaynaklarınızdan bir parola dosyası ve ACL'ler içeren gerçek bir yapılandırma verin:

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

Yapılandırma image'ın standart yolunda olduğu için varsayılan komut onu yükler. mosquitto.conf dosyasını Mosquitto config oluşturucu ile üretin ve parola dosyasını bir kez mosquitto_passwd -c passwd testuser ile oluşturun.

Python: pytest + testcontainers

pip install testcontainers paho-mqtt pytest ile kurun. Oturum kapsamlı bir fixture, tüm test oturumu için tek bir broker başlatır:

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"

Fixture yalnızca (host, port) döndürdüğü için aynı testler herhangi bir broker'a karşı çalışabilir: fixture'ı ("broker.connectmq.com", 1883) döndüren bir fixture ile değiştirin, başka hiçbir şey değişmez. MQTT için pytest fixture'ları bunu ortam değişkenleriyle broker seçimiyle genişletiyor.

Node.js: Jest + testcontainers

npm i -D testcontainers mqtt jest ile kurun. Container'ın başlaması birkaç saniye sürer; bu yüzden hook timeout'unu artırın:

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

Daha büyük bir test paketinde, her test dosyası için ayrı bir container başlatmak yerine container'ı Jest'in globalSetup aşamasında bir kez başlatın ve URL'yi bir ortam değişkeniyle aktarın.

CI'da Testcontainers

GitHub'ın barındırdığı Ubuntu runner'larında Docker bulunur; bu yüzden Testcontainers testleri orada ek kurulum gerektirmeden çalışır: mvn verify, pytest veya npm test yeterlidir. Bilinmesi gereken birkaç nokta:

• macOS ve Windows runner'ları, Ubuntu runner'larının sağladığı şekilde bir Linux Docker motoru sunmaz; container testlerini ubuntu-latest üzerinde çalıştırın.
• Image indirmeleri her yeni runner'da zaman alır. Build'lerin tekrarlanabilir olması için eclipse-mosquitto:2.0.20 gibi belirli bir tag'i (test ettiğiniz sürümü) sabitleyin ve job başına bir kez indirin.
• Kubernetes'te kendi barındırdığınız runner'larda çoğu zaman Docker soketi yoktur. Bu durumda Testcontainers uzak bir Docker host'una veya Testcontainers Cloud'a ihtiyaç duyar ya da testler barındırılan bir broker'a geçer.

Diğer broker'ları ve MQTT 5'i test etmek

Mosquitto 2.x, MQTT 3.1.1 ve MQTT 5'i destekler; bu yüzden istemciyi protokol sürümü 5'e geçirerek MQTT 5 özelliklerini — user properties, message expiry, reason code'lar — aynı container'a karşı test edebilirsiniz. Kodunuzun broker'lar arasında taşınabilir olduğunu kontrol etmek için aynı test sınıfını birkaç image'a karşı çalıştırın: emqx/emqx, hivemq/hivemq-ce ve eclipse-mosquitto varsayılan test kurulumlarında anonim bağlantıları kabul eder. Bekleme stratejileri farklıdır: EMQX ve HiveMQ daha geç başlar; bu yüzden Testcontainers'ın dinleme portuna yönelik varsayılan beklemesini cömert bir başlangıç timeout'uyla kullanmak, log satırlarını eşleştirmekten daha basittir.

Testcontainers mı, barındırılan bir test broker'ı mı?

Testcontainers, izole ve tekrarlanabilir entegrasyon testleri için mükemmeldir. Ancak her iş için doğru araç değildir:

Yaygın bir ayrım, hızlı entegrasyon katmanı için Testcontainers ve TLS, kimlik bilgileri, ACL'ler ve cihaz düzeyindeki testleri kapsayan daha küçük bir test paketi için barındırılan bir broker kullanmaktır. Bu yaklaşım MQTT entegrasyon testi yazısında anlatılıyor ve GitHub Actions ile MQTT testi yazısında CI'a bağlanıyor.

Kurulumunuzu doğrulamak

Bir container testi tuhaf davrandığında parçaları tek tek kontrol edin:

1. Testler çalışırken docker ps: Mosquitto container'ını ve testcontainers/ryuk container'ını görmelisiniz.
2. Container logları (Testcontainers başlatma hatasında bunları yazdırır) Starting in local only mode içermemelidir.
3. Kasıtlı olarak yanlış beklenen payload ile tek bir test çalıştırın ve asılı kalmak yerine timeout içinde başarısız olduğunu doğrulayın.
4. Aynı testleri barındırılan broker'a yönlendirin ve topic'lerin ve payload'ların beklediğiniz gibi göründüğünü doğrulamak için online MQTT istemcisinde testmqtt/tc/# topic'ini izleyin.

Test trafiğini tarayıcıda izleyin

Testcontainers; Java, Python veya Node'da birkaç satırlık kurulumla her test çalıştırmasına kendi gerçek broker'ını verir; yeter ki Mosquitto'yu bir listener ile başlatın ve "running" log satırını bekleyin. Hızlı, izole entegrasyon testleri için onu kullanın; production TLS'ine, kimlik bilgilerine ve ağ yollarına bağlı kontroller için ise barındırılan bir broker tutun.

Sıkça sorulan sorular