Testing MQTT.js code comes down to two layers. Unit tests check your message handling logic with a fake client, in milliseconds and without a network. Integration tests connect to a real broker and prove that topics, QoS and payloads actually work end to end. Both run fine in Jest or Vitest, as long as you avoid the three classic traps: publishing before the subscription is active, waiting with fixed sleeps, and leaving clients open after the run.
This guide uses MQTT.js 5 and its promise API (connectAsync, subscribeAsync, publishAsync, endAsync). Examples are written for Vitest; the Jest differences are noted where they matter.
Make the code testable first
The single most useful decision is to pass the client in instead of creating it inside your module. Here is a small command handler: it listens on <prefix>/cmd, runs the command and publishes an acknowledgement on <prefix>/ack.
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);
};
}Two details make it test-friendly. The function awaits the SUBACK before returning, so callers know the subscription is live. And it returns a stop function, so tests can detach the handler cleanly. If you are new to the client API itself, start with MQTT in JavaScript or MQTT with TypeScript; the code generator produces a working connection snippet for your broker.
Unit tests with a fake client
MqttClient is an EventEmitter. A fake only needs the methods your code calls plus the ability to emit message:
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();
});
});The message handler is async, so the acknowledgement is published a few microtasks after emit(). vi.waitFor() retries the assertion until it passes or times out. In Jest, replace vi.fn() with jest.fn() and flush pending work with a tiny helper instead of vi.waitFor:
export const flush = () => new Promise((resolve) => setImmediate(resolve));
// usage: fake.emit(...); await flush(); expect(fake.publish).toHaveBeenCalled();You can also mock the whole module with vi.mock('mqtt') or jest.mock('mqtt'), but an injected fake is simpler, survives refactors and does not depend on import paths. For a comparison of fakes, in-memory brokers and containers, see mock brokers for unit tests.
A waitForMessage helper
Integration tests need to wait for a specific message without setTimeout(…, 2000). This helper resolves on the first matching message and rejects with a readable error after a deadline:
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);
});
}It removes its listener on both paths, so a long suite does not accumulate handlers (and Node's MaxListenersExceededWarning). The important usage rule: create the promise before you publish, then await it. If you publish first, a fast broker can deliver the reply before the listener exists.
Integration tests against a real broker
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();
});
});The random prefix isolates each run, so parallel CI jobs and teammates never see each other's messages. While a test is running, subscribe to testmqtt/vitest/# in the online MQTT client to watch the commands and acks fly by — the quickest way to tell whether a failure is in your code or on the wire.
Runner configuration
import { defineConfig } from 'vitest/config';
export default defineConfig({
test: {
environment: 'node',
testTimeout: 15_000,
hookTimeout: 15_000,
},
});/** @type {import('jest').Config} */
module.exports = {
preset: 'ts-jest',
testEnvironment: 'node',
testTimeout: 15000,
};{
"scripts": {
"test:unit": "vitest run test/unit",
"test:int": "vitest run test/integration"
}
}Use the node environment for anything that opens a TCP connection: under jsdom, bundler-style resolution can pick MQTT.js's browser build, which only speaks WebSockets. Keep unit and integration tests in separate folders so the fast suite runs on every save and the broker suite runs in CI — see MQTT testing in GitHub Actions for the pipeline side.
Troubleshooting test runs
Jest did not exit one second after the test run has completed.A client is still open. End every client inafterAllwithawait client.endAsync()and runjest --detectOpenHandlesto locate stragglers. Vitest shows the same problem as a hanging run or aclose timed out after 10000msmessage.Exceeded timeout of 5000 ms for a hook(Jest) or a hook timeout in Vitest. The broker was unreachable or rejected the credentials. WithreconnectPeriod: 0andconnectAsync(url, opts, false), the real error (ECONNREFUSED,Not authorized) surfaces instead of an endless retry; look it up in the reason code reference.- Random failures, message never arrives. You published before subscribing, or before the other client's SUBACK. Await
subscribeAsync(or a function likehandleCommandsthat does), and createwaitForMessagebefore publishing. - Tests pass alone, fail together. Shared topics or a fixed
clientId. Two connections with the same client ID kick each other off the broker. Use random IDs and a per-run prefix. - Frozen tests after
vi.useFakeTimers(). Fake timers stop MQTT.js keep-alive and yourwaitForMessagetimeout. Restrict fake timers to unit tests with a fake client, and callvi.useRealTimers()inafterEach.
The same ideas apply in Python — the pytest fixtures for MQTT guide builds the equivalent collectors with threads instead of promises. Inject the client, fake it for logic, use a real broker for wiring, await every acknowledgement and end every connection: those few habits are what make an MQTT.js test suite fast, deterministic and boring in the best sense.
Frequently asked questions
Should I mock the mqtt module or inject a client?
Inject the client. Passing an MqttClient into your functions lets unit tests hand in a small EventEmitter-based fake with no module mocking, and the same functions run unchanged against a real broker in integration tests. Module mocks with vi.mock or jest.mock work, but they couple tests to import paths.
Why does Jest say it did not exit after my MQTT tests?
"Jest did not exit one second after the test run has completed" means a socket or timer is still open — almost always an MQTT client that was never ended. Call await client.endAsync() in afterAll for every client you connect, and run with --detectOpenHandles to find the one you missed.
Can I use fake timers with MQTT.js?
Only with a fake client. A real MqttClient relies on timers for keep-alive, connect timeouts and reconnects, and faked timers freeze them, so integration tests hang or time out strangely. Keep fake timers for unit tests of your own retry or debounce logic.
