Ein MQTT-Smoke-Test-Skript in einer Minute (Bash, Python, Node)

export MQTT_HOST=broker.connectmq.com   # your broker
export MQTT_PORT=8883                   # 1883 plain, 8883 TLS
export MQTT_USER=username               # optional
export MQTT_PASS=password               # optional
export MQTT_TIMEOUT=10                  # seconds per step

#!/usr/bin/env bash
# MQTT smoke test: connect, subscribe, publish, receive. Exit 0 = pass.
set -u
HOST=${MQTT_HOST:-broker.connectmq.com}
PORT=${MQTT_PORT:-1883}
TIMEOUT=${MQTT_TIMEOUT:-10}
TOPIC="testmqtt/smoke/$(hostname -s)-$$-$RANDOM"
TOKEN="smoke-$(date +%s)-$RANDOM"
OUT=$(mktemp); ERR=$(mktemp)
trap 'rm -f "$OUT" "$ERR"' EXIT

ARGS=(-h "$HOST" -p "$PORT" -q 1)
[ -n "${MQTT_USER:-}" ] && ARGS+=(-u "$MQTT_USER" -P "${MQTT_PASS:-}")
[ "$PORT" = 8883 ] && ARGS+=(--capath /etc/ssl/certs)

fail() { echo "FAIL  $1"; exit 1; }
command -v mosquitto_sub >/dev/null || fail "mosquitto clients are not installed"

mosquitto_sub "${ARGS[@]}" -t "$TOPIC" -C 1 -W "$TIMEOUT" >"$OUT" 2>"$ERR" &
SUB=$!
sleep 1
kill -0 "$SUB" 2>/dev/null || fail "connect/subscribe: $(cat "$ERR")"
echo "PASS  connect + subscribe  $HOST:$PORT"

mosquitto_pub "${ARGS[@]}" -t "$TOPIC" -m "$TOKEN" 2>"$ERR" || fail "publish: $(cat "$ERR")"
echo "PASS  publish  $TOPIC"

wait "$SUB" || fail "receive: nothing within ${TIMEOUT}s $(cat "$ERR")"
grep -qx "$TOKEN" "$OUT" || fail "receive: unexpected payload '$(cat "$OUT")'"
echo "PASS  round trip  ${SECONDS}s total"

PASS  connect + subscribe  broker.connectmq.com:8883
PASS  publish  testmqtt/smoke/build-01-4127-18832
PASS  round trip  1s total

#!/usr/bin/env python3
"""MQTT smoke test: exit 0 if connect, subscribe, publish and receive all pass."""
import os, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "1883"))
TIMEOUT = float(os.getenv("MQTT_TIMEOUT", "10"))
TOPIC = f"testmqtt/smoke/{uuid.uuid4().hex[:12]}"
TOKEN = uuid.uuid4().hex
connected, subscribed, received = threading.Event(), threading.Event(), threading.Event()
state = {}

def step(ok, name, detail=""):
    print(f"{'PASS' if ok else 'FAIL'}  {name}  {detail}".rstrip())
    if not ok:
        sys.exit(1)

def on_connect(client, userdata, flags, reason_code, properties):
    state["connack"] = reason_code
    connected.set()

def on_subscribe(client, userdata, mid, reason_code_list, properties):
    state["suback"] = reason_code_list[0]
    subscribed.set()

def on_message(client, userdata, msg):
    if msg.payload.decode(errors="replace") == TOKEN:
        received.set()

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"smoke-{TOKEN[:8]}")
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
if os.getenv("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS", ""))
if PORT == 8883:
    client.tls_set()

start = time.perf_counter()
try:
    client.connect(HOST, PORT, keepalive=30)
except Exception as exc:
    step(False, "connect", f"{HOST}:{PORT} {exc}")
client.loop_start()

ok = connected.wait(TIMEOUT) and not state["connack"].is_failure
step(ok, "connect", f"{HOST}:{PORT} {state.get('connack', 'timeout')}")

client.subscribe(TOPIC, qos=1)
ok = subscribed.wait(TIMEOUT) and not state["suback"].is_failure
step(ok, "subscribe", str(state.get("suback", "timeout")))

info = client.publish(TOPIC, TOKEN, qos=1)
info.wait_for_publish(TIMEOUT)
step(info.is_published(), "publish", TOPIC)

step(received.wait(TIMEOUT), "round trip", f"{(time.perf_counter() - start) * 1000:.0f} ms")
client.disconnect()
client.loop_stop()

// MQTT smoke test: node smoke.mjs — exit code 0 = pass
import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

const host = process.env.MQTT_HOST ?? 'broker.connectmq.com';
const port = process.env.MQTT_PORT ?? '1883';
const url = process.env.MQTT_URL ?? `${port === '8883' ? 'mqtts' : 'mqtt'}://${host}:${port}`;
const timeout = Number(process.env.MQTT_TIMEOUT ?? 10) * 1000;
const topic = `testmqtt/smoke/${randomUUID().slice(0, 12)}`;
const token = randomUUID();

const step = (ok, name, detail = '') => {
  console.log(`${ok ? 'PASS' : 'FAIL'}  ${name}  ${detail}`.trimEnd());
  if (!ok) process.exit(1);
};
const within = (promise, what) =>
  Promise.race([promise, new Promise((_, reject) =>
    setTimeout(() => reject(new Error(`${what} timed out after ${timeout} ms`)), timeout))]);

const start = performance.now();
let client;
try {
  client = await within(mqtt.connectAsync(url, {
    clientId: `smoke-${token.slice(0, 8)}`,
    username: process.env.MQTT_USER,
    password: process.env.MQTT_PASS,
    reconnectPeriod: 0,
    connectTimeout: timeout,
  }, false), 'connect');
  step(true, 'connect', url);
} catch (err) {
  step(false, 'connect', `${url} ${err.message}`);
}

const received = new Promise((resolve) => {
  client.on('message', (t, payload) => {
    if (t === topic && payload.toString() === token) resolve();
  });
});

try {
  const [grant] = await within(client.subscribeAsync(topic, { qos: 1 }), 'subscribe');
  step(grant.qos < 128, 'subscribe', `granted QoS ${grant.qos}`);
  await within(client.publishAsync(topic, token, { qos: 1 }), 'publish');
  step(true, 'publish', topic);
  await within(received, 'receive');
  step(true, 'round trip', `${Math.round(performance.now() - start)} ms`);
} catch (err) {
  step(false, 'step', err.message);
}
await client.endAsync();

- name: MQTT smoke test
  env:
    MQTT_HOST: broker.connectmq.com
    MQTT_PORT: "8883"
    MQTT_USER: ${{ secrets.MQTT_USER }}
    MQTT_PASS: ${{ secrets.MQTT_PASS }}
  run: |
    sudo apt-get update && sudo apt-get install -y mosquitto-clients
    bash smoke.sh