mesh-catalog/modules/nodered/test/mqtt.test.ts

// What holds nodered's MQTT step (mqtt/connection.ts): the broker nodes the mesh owns — the one it
// makes, or the ones settings name — are made to use the broker and login the mesh bound, only after
// the broker takes that login; every other field of a node is kept; nothing is deployed when nothing
// differs; a node that is not named is never touched; a loopback broker address is refused.
//
// Node-RED and the broker are fakes answering as the real ones do (admin API v2 of nodered/node-red
// 5.0.7, the build ace runs).

import { test } from "node:test";
import assert from "node:assert/strict";

import { MESH_BROKER_ID, namedBrokers, reconcileBrokers, type Binding, type FlowNode, type Marks, type NodeRed } from "../mqtt/connection.ts";
import type { Probe } from "../mqtt/probe.ts";

const MINTED = "mesh-minted-password";
const binding = (at = "ace.internal"): Binding => ({ provision: "mqtt-topic", from: "ace", at, as: "mesh_ace_nodered", serves: { scheme: "mqtt", port: 1883 } });

/** ace's flows, reduced: its one broker node (dead, zurag.be:1884) and a node that uses it. */
function aceFlows(): FlowNode[] {
  return [
    { id: "2b0aece9c5f3b307", type: "mqtt-broker", name: "MQTT Broker", broker: "zurag.be", port: "1884", clientid: "", usetls: false, protocolVersion: "4", keepalive: "60" },
    { id: "fe0cae96f1e3ae4d", type: "mqtt in", topic: "stat/sonoff_office_light_switch/RESULT", broker: "2b0aece9c5f3b307", z: "t" },
    { id: "other-broker", type: "mqtt-broker", name: "someone else's", broker: "test.mosquitto.org", port: "1883" },
  ];
}

function fakeNodeRed(flows: FlowNode[], creds: Record<string, { user?: string; password?: string }> = {}) {
  let rev = "r1";
  const deploys: FlowNode[][] = [];
  const nodered: NodeRed = {
    async flows() {
      return { rev, flows: structuredClone(flows) };
    },
    async credentials(_type, id) {
      const c = creds[id] ?? {};
      return { user: c.user, has_password: Boolean(c.password) };
    },
    async deploy(at, next) {
      if (at !== rev) throw new Error("Node-RED /flows: 409 version_mismatch");
      for (const n of next) {
        if (n.credentials) creds[n.id] = { ...(creds[n.id] ?? {}), ...(n.credentials as object) };
      }
      flows.splice(0, flows.length, ...next.map(({ credentials: _c, ...n }) => n as FlowNode));
      deploys.push(next);
      rev = `r${deploys.length + 1}`;
    },
  };
  return { nodered, deploys, flows, creds };
}

const marks = (): Marks & { store: Map<string, string> } => {
  const store = new Map<string, string>();
  return { store, get: async (k) => store.get(k), set: async (k, v) => void store.set(k, v) };
};
const takes: Probe = async (_h, _p, user, pass) => ({ connack: user === "mesh_ace_nodered" && pass === MINTED ? 0 : 5, suback: 0 });

test("ace: the named broker node is moved to the bound broker and login; the other broker is not touched", async () => {
  const f = fakeNodeRed(aceFlows(), { "2b0aece9c5f3b307": { user: "luffy", password: "old" }, "other-broker": { user: "x", password: "y" } });
  const m = marks();
  const out = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: m }, binding(), `${MINTED}\n`, ["2b0aece9c5f3b307"]);
  assert.deepEqual(out, [{ what: "broker 2b0aece9c5f3b307", result: "written", fields: ["broker", "port", "user", "password"] }]);
  const node = f.flows.find((n) => n.id === "2b0aece9c5f3b307");
  assert.deepEqual(node, { ...aceFlows()[0], broker: "ace.internal", port: "1883", usetls: false });
  assert.deepEqual(f.creds["2b0aece9c5f3b307"], { user: "mesh_ace_nodered", password: MINTED });
  assert.deepEqual(f.flows.find((n) => n.id === "other-broker"), aceFlows()[2]);
  assert.deepEqual(f.creds["other-broker"], { user: "x", password: "y" });
  // Only the changed node carried credentials in the deploy.
  assert.deepEqual(f.deploys[0].filter((n) => n.credentials).map((n) => n.id), ["2b0aece9c5f3b307"]);

  // Again: nothing differs, nothing is deployed.
  const again = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: m }, binding(), MINTED, ["2b0aece9c5f3b307"]);
  assert.deepEqual(again, [{ what: "broker 2b0aece9c5f3b307", result: "unchanged" }]);
  assert.equal(f.deploys.length, 1);
});

test("fresh: with nothing named, the step makes its own broker node", async () => {
  const f = fakeNodeRed([]);
  const out = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: marks() }, binding(), MINTED, []);
  assert.deepEqual(out, [{ what: `broker ${MESH_BROKER_ID}`, result: "written", fields: ["node"] }]);
  assert.equal(f.flows[0].type, "mqtt-broker");
  assert.equal(f.flows[0].broker, "ace.internal");
  assert.deepEqual(f.creds[MESH_BROKER_ID], { user: "mesh_ace_nodered", password: MINTED });
});

test("a login the broker does not take is never written", async () => {
  const f = fakeNodeRed(aceFlows());
  const out = await reconcileBrokers({ nodered: f.nodered, probe: async () => ({ connack: 5 }), marks: marks() }, binding(), MINTED, ["2b0aece9c5f3b307"]);
  assert.equal(out[0].result, "refused");
  assert.equal(f.deploys.length, 0);
});

test("a named node that is not there, or a loopback broker, is refused", async () => {
  const f = fakeNodeRed(aceFlows());
  const out = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: marks() }, binding(), MINTED, ["gone"]);
  assert.match((out[0] as { problem: string }).problem, /no such node/);
  const lo = await reconcileBrokers({ nodered: f.nodered, probe: takes, marks: marks() }, binding("127.0.0.1"), MINTED, []);
  assert.match((lo[0] as { problem: string }).problem, /Node-RED itself/);
  assert.equal(f.deploys.length, 0);
});

test("a deploy that raced another is read again once", async () => {
  const f = fakeNodeRed(aceFlows());
  let first = true;
  const racing: NodeRed = {
    ...f.nodered,
    async flows() {
      const got = await f.nodered.flows();
      if (first) {
        first = false;
        return { ...got, rev: "stale" };
      }
      return got;
    },
  };
  const out = await reconcileBrokers({ nodered: racing, probe: takes, marks: marks() }, binding(), MINTED, ["2b0aece9c5f3b307"]);
  assert.equal(out[0].result, "written");
  assert.equal(f.deploys.length, 1);
});

test("settings name broker nodes under mqtt.brokers", () => {
  assert.deepEqual(namedBrokers({ mqtt: { brokers: ["a", "", 3, "b"] } }), ["a", "b"]);
  assert.deepEqual(namedBrokers({ endpoints: {} }), []);
  assert.deepEqual(namedBrokers(undefined), []);
});