de
Managed Broker Client öffnen
Broker-Tests

MQTT.js-Code mit Jest und Vitest testen

Unit- und Integrationstests für MQTT.js-5-Apps mit Jest oder Vitest: Fake-Clients, ein waitForMessage-Helper mit Timeout, echter Broker und sauberer Teardown.

Aktualisiert · 9 Min. Lesezeit

MQTT.js-Code mit Jest und Vitest testen

Beim Testen von MQTT.js-Code geht es um zwei Ebenen. Unit-Tests prüfen deine Nachrichtenlogik mit einem Fake-Client, in Millisekunden und ohne Netzwerk. Integrationstests verbinden sich mit einem echten Broker und beweisen, dass Topics, QoS und Payloads Ende-zu-Ende funktionieren. Beides läuft problemlos in Jest oder Vitest, solange du die drei klassischen Fallen meidest: publishen, bevor die Subscription aktiv ist, mit festen Sleeps warten und Clients nach dem Lauf offen lassen.

Diese Anleitung nutzt MQTT.js 5 und seine Promise-API (connectAsync, subscribeAsync, publishAsync, endAsync). Die Beispiele sind für Vitest geschrieben; Unterschiede zu Jest werden dort genannt, wo sie wichtig sind.

Zuerst den Code testbar machen

Die nützlichste Entscheidung ist, den Client hineinzureichen, statt ihn im Modul zu erzeugen. Hier ein kleiner Command-Handler: Er hört auf <prefix>/cmd, führt den Befehl aus und publisht eine Bestätigung auf <prefix>/ack.

src/commands.ts
import type { MqttClient } from 'mqtt';

export interface Command {
  id: string;
  action: string;
}

export async function handleCommands(
  client: MqttClient,
  prefix: string,
  run: (cmd: Command) => Promise<void> | void,
): Promise<() => void> {
  const cmdTopic = `${prefix}/cmd`;
  const ackTopic = `${prefix}/ack`;

  const onMessage = async (topic: string, payload: Buffer) => {
    if (topic !== cmdTopic) return;
    let cmd: Command;
    try {
      cmd = JSON.parse(payload.toString());
    } catch {
      client.publish(ackTopic, JSON.stringify({ ok: false, error: 'invalid json' }), { qos: 1 });
      return;
    }
    try {
      await run(cmd);
      client.publish(ackTopic, JSON.stringify({ id: cmd.id, ok: true }), { qos: 1 });
    } catch (err) {
      client.publish(ackTopic, JSON.stringify({ id: cmd.id, ok: false, error: String(err) }), { qos: 1 });
    }
  };

  client.on('message', onMessage);
  await client.subscribeAsync(cmdTopic, { qos: 1 });

  return () => {
    client.off('message', onMessage);
    client.unsubscribe(cmdTopic);
  };
}

Zwei Details machen ihn testfreundlich. Die Funktion wartet auf das SUBACK, bevor sie zurückkehrt, der Aufrufer weiß also, dass die Subscription aktiv ist. Und sie gibt eine Stop-Funktion zurück, damit Tests den Handler sauber abhängen können. Wenn dir die Client-API selbst neu ist, starte mit MQTT in JavaScript oder MQTT mit TypeScript; der Code-Generator erzeugt ein funktionierendes Verbindungs-Snippet für deinen Broker.

Unit-Tests mit einem Fake-Client

MqttClient ist ein EventEmitter. Ein Fake braucht nur die Methoden, die dein Code aufruft, plus die Möglichkeit, message auszulösen:

test/unit/commands.test.ts
import { EventEmitter } from 'node:events';
import type { MqttClient } from 'mqtt';
import { describe, expect, it, vi } from 'vitest';
import { handleCommands } from '../../src/commands';

function createFakeClient() {
  const fake = Object.assign(new EventEmitter(), {
    subscribeAsync: vi.fn().mockResolvedValue([]),
    unsubscribe: vi.fn(),
    publish: vi.fn(),
  });
  return { fake, client: fake as unknown as MqttClient };
}

describe('handleCommands', () => {
  it('runs a valid command and acknowledges it', async () => {
    const { fake, client } = createFakeClient();
    const run = vi.fn();
    await handleCommands(client, 'dev/1', run);

    expect(fake.subscribeAsync).toHaveBeenCalledWith('dev/1/cmd', { qos: 1 });
    fake.emit('message', 'dev/1/cmd', Buffer.from('{"id":"c1","action":"reboot"}'));

    await vi.waitFor(() => expect(fake.publish).toHaveBeenCalledTimes(1));
    const [topic, body] = fake.publish.mock.calls[0];
    expect(topic).toBe('dev/1/ack');
    expect(JSON.parse(body)).toEqual({ id: 'c1', ok: true });
    expect(run).toHaveBeenCalledWith({ id: 'c1', action: 'reboot' });
  });

  it('rejects malformed JSON without calling run', async () => {
    const { fake, client } = createFakeClient();
    const run = vi.fn();
    await handleCommands(client, 'dev/1', run);

    fake.emit('message', 'dev/1/cmd', Buffer.from('{not json'));

    await vi.waitFor(() => expect(fake.publish).toHaveBeenCalled());
    expect(JSON.parse(fake.publish.mock.calls[0][1])).toMatchObject({ ok: false });
    expect(run).not.toHaveBeenCalled();
  });

  it('ignores other topics', async () => {
    const { fake, client } = createFakeClient();
    await handleCommands(client, 'dev/1', vi.fn());
    fake.emit('message', 'dev/2/cmd', Buffer.from('{"id":"x","action":"y"}'));
    expect(fake.publish).not.toHaveBeenCalled();
  });
});

Der Message-Handler ist async, die Bestätigung wird also einige Microtasks nach emit() gepublisht. vi.waitFor() wiederholt die Assertion, bis sie besteht oder die Zeit abläuft. In Jest ersetzt du vi.fn() durch jest.fn() und leerst ausstehende Arbeit mit einem kleinen Helper statt vi.waitFor:

test/helpers/flush.ts (Jest)
export const flush = () => new Promise((resolve) => setImmediate(resolve));
// usage: fake.emit(...); await flush(); expect(fake.publish).toHaveBeenCalled();

Du kannst auch das ganze Modul mit vi.mock('mqtt') oder jest.mock('mqtt') mocken, aber ein injizierter Fake ist einfacher, übersteht Refactorings und hängt nicht von Importpfaden ab. Einen Vergleich von Fakes, In-Memory-Brokern und Containern findest du unter Mock-Broker für Unit-Tests.

Ein waitForMessage-Helper

Integrationstests müssen auf eine bestimmte Nachricht warten, ohne setTimeout(…, 2000). Dieser Helper löst bei der ersten passenden Nachricht auf und lehnt nach einer Deadline mit einer lesbaren Fehlermeldung ab:

test/helpers/waitForMessage.ts
import type { MqttClient } from 'mqtt';

export function waitForMessage(
  client: MqttClient,
  topic: string,
  { timeout = 5000, match = (_: Buffer) => true } = {},
): Promise<{ topic: string; payload: Buffer }> {
  return new Promise((resolve, reject) => {
    const onMessage = (t: string, payload: Buffer) => {
      if (t !== topic || !match(payload)) return;
      clearTimeout(timer);
      client.off('message', onMessage);
      resolve({ topic: t, payload });
    };
    const timer = setTimeout(() => {
      client.off('message', onMessage);
      reject(new Error(`no message on "${topic}" within ${timeout} ms`));
    }, timeout);
    client.on('message', onMessage);
  });
}

Er entfernt seinen Listener auf beiden Wegen, sodass eine lange Suite keine Handler ansammelt (und keine MaxListenersExceededWarning von Node). Die wichtige Nutzungsregel: Erzeuge das Promise, bevor du publishst, und warte dann darauf. Publishst du zuerst, kann ein schneller Broker die Antwort liefern, bevor der Listener existiert.

Integrationstests gegen einen echten Broker

test/integration/commands.int.test.ts
import { randomUUID } from 'node:crypto';
import { connectAsync, type MqttClient } from 'mqtt';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { handleCommands } from '../../src/commands';
import { waitForMessage } from '../helpers/waitForMessage';

const url = process.env.MQTT_URL ?? 'mqtt://broker.connectmq.com:1883';
const options = {
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  connectTimeout: 5000,
  reconnectPeriod: 0, // fail fast instead of retrying forever
};

describe('command handler on a real broker', () => {
  const prefix = `testmqtt/vitest/${randomUUID().slice(0, 8)}`;
  let device: MqttClient;
  let operator: MqttClient;
  let stop: (() => void) | undefined;

  beforeAll(async () => {
    // third argument false: reject on the first failed attempt
    [device, operator] = await Promise.all([
      connectAsync(url, { ...options, clientId: `device-${randomUUID()}` }, false),
      connectAsync(url, { ...options, clientId: `operator-${randomUUID()}` }, false),
    ]);
    stop = await handleCommands(device, prefix, async () => {});
    await operator.subscribeAsync(`${prefix}/ack`, { qos: 1 });
  });

  afterAll(async () => {
    stop?.();
    await Promise.all([device?.endAsync(), operator?.endAsync()]);
  });

  it('acknowledges a command end to end', async () => {
    const ack = waitForMessage(operator, `${prefix}/ack`);
    await operator.publishAsync(`${prefix}/cmd`, JSON.stringify({ id: 'c1', action: 'ping' }), { qos: 1 });

    const { payload } = await ack;
    expect(JSON.parse(payload.toString())).toEqual({ id: 'c1', ok: true });
  });

  it('reports malformed commands', async () => {
    const ack = waitForMessage(operator, `${prefix}/ack`, {
      match: (p) => JSON.parse(p.toString()).ok === false,
    });
    await operator.publishAsync(`${prefix}/cmd`, 'not json', { qos: 1 });
    await expect(ack).resolves.toBeDefined();
  });
});

Das zufällige prefix isoliert jeden Lauf, sodass parallele CI-Jobs und Teamkollegen nie die Nachrichten der anderen sehen. Während ein Test läuft, abonniere testmqtt/vitest/# im Online-MQTT-Client und sieh die Commands und Acks vorbeifliegen – der schnellste Weg herauszufinden, ob ein Fehler in deinem Code oder auf der Leitung liegt.

Runner-Konfiguration

vitest.config.ts
import { defineConfig } from 'vitest/config';

export default defineConfig({
  test: {
    environment: 'node',
    testTimeout: 15_000,
    hookTimeout: 15_000,
  },
});
jest.config.js
/** @type {import('jest').Config} */
module.exports = {
  preset: 'ts-jest',
  testEnvironment: 'node',
  testTimeout: 15000,
};
package.json (scripts)
{
  "scripts": {
    "test:unit": "vitest run test/unit",
    "test:int": "vitest run test/integration"
  }
}

Nutze die node-Umgebung für alles, was eine TCP-Verbindung öffnet: Unter jsdom kann die Bundler-artige Auflösung den Browser-Build von MQTT.js wählen, der nur WebSockets spricht. Halte Unit- und Integrationstests in getrennten Ordnern, damit die schnelle Suite bei jedem Speichern läuft und die Broker-Suite in CI – die Pipeline-Seite beschreibt MQTT-Tests in GitHub Actions.

Fehlersuche bei Testläufen

  • Jest did not exit one second after the test run has completed. Ein Client ist noch offen. Beende jeden Client in afterAll mit await client.endAsync() und starte jest --detectOpenHandles, um Nachzügler zu finden. Vitest zeigt dasselbe Problem als hängenden Lauf oder mit der Meldung close timed out after 10000ms.
  • Exceeded timeout of 5000 ms for a hook (Jest) oder ein Hook-Timeout in Vitest. Der Broker war nicht erreichbar oder hat die Zugangsdaten abgelehnt. Mit reconnectPeriod: 0 und connectAsync(url, opts, false) kommt der echte Fehler (ECONNREFUSED, Not authorized) zum Vorschein statt endloser Wiederholungen; schlag ihn in der Reason-Code-Referenz nach.
  • Zufällige Fehler, Nachricht kommt nie an. Du hast vor dem Subscribe gepublisht oder vor dem SUBACK des anderen Clients. Warte auf subscribeAsync (oder eine Funktion wie handleCommands, die das tut) und erzeuge waitForMessage vor dem Publish.
  • Tests bestehen einzeln, scheitern gemeinsam. Geteilte Topics oder eine feste clientId. Zwei Verbindungen mit derselben Client-ID werfen sich gegenseitig vom Broker. Nutze zufällige IDs und ein Präfix pro Lauf.
  • Eingefrorene Tests nach vi.useFakeTimers(). Fake Timers stoppen das Keep-Alive von MQTT.js und den Timeout deines waitForMessage. Beschränke sie auf Unit-Tests mit Fake-Client und rufe vi.useRealTimers() in afterEach auf.

Dieselben Ideen gelten in Python – die Anleitung zu pytest-Fixtures für MQTT baut die entsprechenden Collector mit Threads statt Promises. Client injizieren, für Logik faken, für die Verdrahtung einen echten Broker nutzen, auf jede Bestätigung warten und jede Verbindung beenden: Diese wenigen Gewohnheiten machen eine MQTT.js-Testsuite schnell, deterministisch und im besten Sinne langweilig.

Häufig gestellte Fragen

Sollte ich das mqtt-Modul mocken oder einen Client injizieren?

Injiziere den Client. Wenn deine Funktionen einen MqttClient übergeben bekommen, können Unit-Tests einen kleinen EventEmitter-basierten Fake ohne Modul-Mocking reichen, und dieselben Funktionen laufen in Integrationstests unverändert gegen einen echten Broker. Modul-Mocks mit vi.mock oder jest.mock funktionieren, koppeln Tests aber an Importpfade.

Warum meldet Jest, dass es nach meinen MQTT-Tests nicht beendet wurde?

"Jest did not exit one second after the test run has completed" bedeutet, dass noch ein Socket oder Timer offen ist – fast immer ein MQTT-Client, der nie beendet wurde. Rufe für jeden verbundenen Client await client.endAsync() in afterAll auf und starte mit --detectOpenHandles, um den vergessenen zu finden.

Kann ich Fake Timers mit MQTT.js verwenden?

Nur mit einem Fake-Client. Ein echter MqttClient braucht Timer für Keep-Alive, Connect-Timeouts und Reconnects; Fake Timers frieren sie ein, sodass Integrationstests hängen oder seltsam in Timeouts laufen. Nutze Fake Timers nur für Unit-Tests deiner eigenen Retry- oder Debounce-Logik.

Weiterlesen

Broker-Tests

MQTT-Integrationstests: Deine App gegen einen echten Broker testen

Zuverlässige MQTT-Integrationstests mit pytest und Jest gegen einen echten Broker: eindeutige Topic-Präfixe, auf SUBACK warten, Aufräumen, keine Flaky-Sleeps.

10 Min. LesezeitBroker-Tests

MQTT in Unit-Tests mocken: Fakes, Stubs und In-Memory-Broker

So mockst du MQTT in Unit-Tests: unittest.mock für paho-mqtt, Jest-Mocks für MQTT.js, ein In-Memory-Fake und aedes als eingebetteter Broker, mit klaren Grenzen.

9 Min. LesezeitTutorials

MQTT in JavaScript: MQTT.js-Tutorial für Node.js und den Browser

Lerne MQTT in JavaScript mit MQTT.js v5: verbinden, subscriben und publishen in Node.js und im Browser über WebSockets, mit async/await, React und Fehlerbehandlung.

8 Min. Lesezeit