Alex2Node/test/helpers/harness.js

"use strict";
// Shared by every test file: an in-process MQTT broker (aedes), a client that plays the Alexa side of the Alex2MQTT
// topic contract, and until() for polling. Everything listens on 127.0.0.1, on a port the OS picks. node --test runs
// each file in its own process, so the cleanups registered here close that file's brokers and clients when it ends.
const { after } = require("node:test");
const net = require("node:net");
const aedes = require("aedes");
const mqtt = require("mqtt");

const cleanups = [];
// Newest first: a client closes before the broker it is connected to.
after(async () => {
  for (const close of cleanups.reverse()) {
    try { await close(); } catch { /* already closed */ }
  }
});

/** Run fn when the test file ends. */
function cleanup(fn) {
  cleanups.push(fn);
}

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// What crypto.randomUUID() returns. Alexa wants a new messageId in every response and event and recommends a
// version 4 UUID (Message guide, "Header object").
const UUID_V4 = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;

/** Poll fn every 15 ms until it returns something truthy. Throws "timeout waiting for <what>" after ms. */
async function until(fn, ms = 3000, what = "condition") {
  const start = Date.now();
  while (Date.now() - start < ms) {
    if (await fn()) return true;
    await sleep(15);
  }
  throw new Error(`timeout waiting for ${what}`);
}

/** An MQTT broker on 127.0.0.1. */
async function broker() {
  const instance = aedes();
  const server = net.createServer(instance.handle);
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
  cleanup(() => new Promise((resolve) => instance.close(() => server.close(() => resolve()))));
  const { port } = server.address();
  return { aedes: instance, port, url: `mqtt://127.0.0.1:${port}` };
}

/**
 * The Alexa side (what Alex2MQTT does on the broker): a client subscribed to filter that keeps every message it
 * receives, in order, as { topic, payload } - the payload parsed when it is JSON, the text otherwise.
 */
async function watcher(url, filter) {
  const client = mqtt.connect(url);
  cleanup(() => new Promise((resolve) => client.end(true, {}, resolve)));
  await new Promise((resolve, reject) => { client.once("connect", resolve); client.once("error", reject); });
  const got = [];
  client.on("message", (topic, message) => {
    let payload = message.toString();
    try { payload = JSON.parse(payload); } catch { /* not JSON: kept as text */ }
    got.push({ topic, payload });
  });
  await new Promise((resolve, reject) => client.subscribe(filter, (err) => (err ? reject(err) : resolve())));
  const publish = (topic, payload) => client.publish(topic, typeof payload === "string" ? payload : JSON.stringify(payload));
  return {
    client,
    got,
    publish,
    /** The payloads received on one topic, oldest first. */
    on: (topic) => got.filter((m) => m.topic === topic).map((m) => m.payload),
    /** Publish a directive where Alex2MQTT does: <root>/<endpointId>/alexaDirective. */
    send: (root, message) => publish(`${root}/${message.endpoint.endpointId}/alexaDirective`, message),
  };
}

/** A directive as Alex2MQTT publishes it: the Alexa directive with endpoint.scope (the user's token) removed. */
function directive(namespace, name, endpointId, correlationToken, payload = {}) {
  return {
    header: { namespace, name, payloadVersion: "3", messageId: `message-${correlationToken}`, correlationToken },
    endpoint: { endpointId },
    payload,
  };
}

/**
 * connect() the bridge and wait for its "connect" event, which fires once it is subscribed - a directive published
 * earlier would be lost. The bridge is disconnected when the test file ends.
 */
async function connected(bridge, ms = 3000) {
  let subscribed = false;
  bridge.once("connect", () => { subscribed = true; });
  bridge.connect();
  cleanup(() => bridge.disconnect());
  await until(() => subscribed, ms, "the bridge to connect and subscribe");
  return bridge;
}

/**
 * A broker, the Alexa side subscribed to <root>/#, and a connected bridge. The Alex2MQTT class is passed in: the test
 * file decides which build of the library it loads, not the harness.
 */
async function setup(Alex2MQTT, root, options = {}) {
  const { url } = await broker();
  const alexa = await watcher(url, `${root}/#`);
  const bridge = await connected(new Alex2MQTT("u", "p", root, false, { host: url, ...options }));
  return { url, alexa, bridge };
}

module.exports = { broker, watcher, directive, connected, setup, until, sleep, cleanup, UUID_V4 };