// Integration tests for services/dectRelayHub.js. // // These spin up a real HTTP server on an ephemeral port, attach the // hub, and connect a real `ws` client that plays the role of the // dect-relay-agent. This gives us end-to-end coverage of the auth // path, the wire protocol, RPC correlation, timeouts, and clean // disconnect handling — none of which we can meaningfully test with // pure mocks. // // Every test creates its own hub + server so they can run in parallel // without port conflicts. All servers are torn down in the test's // finally block so a failing test can't leak file descriptors. import test from 'node:test'; import assert from 'node:assert/strict'; import http from 'node:http'; import WebSocket from 'ws'; import { DectRelayHub, RelayErrorCodes, DectRelayError } from '../services/dectRelayHub.js'; const TEST_TOKEN = 'super-secret-token-for-tests'; // ─── Test harness ─────────────────────────────────────────────────── /** * Spin up an HTTP server with the hub attached on an ephemeral port. * Returns { hub, port, closeAll }. Caller MUST call closeAll() (in a * try/finally) to release the port + socket handles. */ async function makeHub(token = TEST_TOKEN) { const server = http.createServer((req, res) => { res.writeHead(404); res.end(); }); const hub = new DectRelayHub({ token }); hub.attachTo(server); await new Promise((res) => server.listen(0, '127.0.0.1', res)); const { port } = server.address(); return { hub, port, async closeAll() { await hub.close(); await new Promise((res) => server.close(() => res())); }, }; } /** * Open a WebSocket client to the hub. Optional bearer overrides the * default token — useful for the "wrong token" test. */ function connectAgent(port, { bearer = TEST_TOKEN, useProtocol = false } = {}) { const url = `ws://127.0.0.1:${port}/dect-relay/ws`; const opts = useProtocol ? { headers: {}, protocol: `bearer.${bearer}` } : { headers: { Authorization: `Bearer ${bearer}` } }; return new WebSocket(url, opts.protocol ? opts.protocol : undefined, { headers: opts.headers, handshakeTimeout: 3000, }); } function waitOpen(ws) { return new Promise((resolve, reject) => { ws.once('open', resolve); ws.once('error', reject); }); } function waitClose(ws) { return new Promise((resolve) => ws.once('close', (code, reason) => resolve({ code, reason: reason?.toString() || '' }))); } // A tiny agent that immediately replies to every command with the // given handler. Handler receives the parsed inbound message and // returns either { ok:true, result:{...} } or throws. function attachAutoAgent(ws, handler) { ws.on('message', async (raw) => { const msg = JSON.parse(raw.toString('utf8')); if (msg.type === 'ping') { ws.send(JSON.stringify({ type: 'pong', at: Date.now() })); return; } if (!msg.id) return; try { const result = await handler(msg); ws.send(JSON.stringify({ id: msg.id, ok: true, result, elapsedMs: 1 })); } catch (err) { ws.send(JSON.stringify({ id: msg.id, ok: false, error: { code: err.code || 'AUTO_AGENT_ERR', message: err.message }, })); } }); } // ─── isConnected / status ─────────────────────────────────────────── test('hub: isConnected is false with no agent', async () => { const { hub, closeAll } = await makeHub(); try { assert.equal(hub.isConnected(), false); assert.equal(hub.status().connected, false); assert.equal(hub.status().agent, null); assert.equal(hub.status().inFlight, 0); } finally { await closeAll(); } }); test('hub: rpc without connection rejects with NOT_CONNECTED', async () => { const { hub, closeAll } = await makeHub(); try { await assert.rejects( hub.collect('10.0.0.100'), (err) => err instanceof DectRelayError && err.code === RelayErrorCodes.NOT_CONNECTED, ); } finally { await closeAll(); } }); // ─── Auth ────────────────────────────────────────────────────────── test('hub: rejects upgrade with no bearer', async () => { const { port, closeAll } = await makeHub(); try { const ws = new WebSocket(`ws://127.0.0.1:${port}/dect-relay/ws`, { handshakeTimeout: 3000, }); // Server writes a raw 401 before the WS handshake completes. // ws throws 'Unexpected server response: 401' as an error. await assert.rejects(waitOpen(ws), /401/); } finally { await closeAll(); } }); test('hub: rejects upgrade with wrong bearer', async () => { const { port, closeAll } = await makeHub(); try { const ws = connectAgent(port, { bearer: 'wrong-token' }); await assert.rejects(waitOpen(ws), /401/); } finally { await closeAll(); } }); test('hub: accepts upgrade with correct bearer via Authorization header', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); // Give the hub a tick to record the adoption. await new Promise((r) => setImmediate(r)); assert.equal(hub.isConnected(), true); ws.close(); await waitClose(ws); } finally { await closeAll(); } }); test('hub: accepts upgrade via Sec-WebSocket-Protocol fallback', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port, { useProtocol: true }); await waitOpen(ws); await new Promise((r) => setImmediate(r)); assert.equal(hub.isConnected(), true); ws.close(); await waitClose(ws); } finally { await closeAll(); } }); // ─── Hello frame ──────────────────────────────────────────────────── test('hub: records agent hello frame into status()', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); ws.send(JSON.stringify({ type: 'hello', agentVersion: '9.9.9', hostname: 'test-host', capabilities: ['collect', 'reboot'], })); // Wait until hub processes it (message events are queued). await new Promise((r) => setTimeout(r, 20)); const s = hub.status(); assert.equal(s.connected, true); assert.equal(s.agent.agentVersion, '9.9.9'); assert.equal(s.agent.hostname, 'test-host'); assert.deepEqual(s.agent.capabilities, ['collect', 'reboot']); ws.close(); await waitClose(ws); } finally { await closeAll(); } }); // ─── RPC correlation ──────────────────────────────────────────────── test('hub: RPC round-trip resolves with agent result', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); attachAutoAgent(ws, async (msg) => { assert.equal(msg.type, 'collect'); assert.equal(msg.baseIp, '10.4.11.87'); return { parsed: { device: { macAddress: 'aa:bb:cc:dd:ee:ff' } }, verdict: { healthy: true } }; }); const { result } = await hub.collect('10.4.11.87'); assert.equal(result.parsed.device.macAddress, 'aa:bb:cc:dd:ee:ff'); assert.equal(result.verdict.healthy, true); ws.close(); await waitClose(ws); } finally { await closeAll(); } }); test('hub: RPC error from agent surfaces as DectRelayError with code', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); attachAutoAgent(ws, async () => { const err = new Error('base rejected credentials'); err.code = 'DIGEST_401'; throw err; }); await assert.rejects( hub.collect('10.4.11.87'), (err) => err instanceof DectRelayError && err.code === 'DIGEST_401' && /base rejected credentials/i.test(err.message), ); } finally { await closeAll(); } }); test('hub: multiple concurrent RPCs correlate by id, not order', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); // Delay short IPs less than long IPs, deliberately reversing // response order relative to send order. ws.on('message', (raw) => { const msg = JSON.parse(raw.toString('utf8')); if (!msg.id) return; const delay = msg.baseIp === '10.0.0.1' ? 40 : 5; setTimeout(() => { ws.send(JSON.stringify({ id: msg.id, ok: true, result: { echo: msg.baseIp }, elapsedMs: delay, })); }, delay); }); const [a, b] = await Promise.all([ hub.collect('10.0.0.1'), // slower hub.collect('10.0.0.2'), // faster ]); assert.equal(a.result.echo, '10.0.0.1'); assert.equal(b.result.echo, '10.0.0.2'); } finally { await closeAll(); } }); // ─── Timeout ──────────────────────────────────────────────────────── test('hub: RPC that never gets a reply times out with TIMEOUT code', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); // Silent agent: acknowledge nothing. ws.on('message', () => { /* intentionally do nothing */ }); await assert.rejects( hub.collect('10.0.0.1', { timeoutMs: 50 }), (err) => err instanceof DectRelayError && err.code === RelayErrorCodes.TIMEOUT, ); } finally { await closeAll(); } }); // ─── Mid-flight disconnect ────────────────────────────────────────── test('hub: agent disconnect mid-RPC rejects the pending promise', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); // Close the socket the moment we receive a command. ws.on('message', () => ws.close(1000, 'test')); await assert.rejects( hub.collect('10.0.0.1', { timeoutMs: 2000 }), (err) => err instanceof DectRelayError && err.code === RelayErrorCodes.DISCONNECTED, ); } finally { await closeAll(); } }); // ─── Second agent replaces first ──────────────────────────────────── test('hub: second agent connection replaces the first (with clean close)', async () => { const { hub, port, closeAll } = await makeHub(); try { const wsA = connectAgent(port); await waitOpen(wsA); const closedA = waitClose(wsA); const wsB = connectAgent(port); await waitOpen(wsB); // wsA should have been closed by the hub with reason "replaced". const closeInfo = await closedA; assert.equal(closeInfo.code, 1000); assert.match(closeInfo.reason, /replaced/i); // The hub is still connected — to wsB now. assert.equal(hub.isConnected(), true); wsB.close(); await waitClose(wsB); } finally { await closeAll(); } }); // ─── execAction routing ──────────────────────────────────────────── test('hub: execAction routes action name into type field', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); let observed = null; attachAutoAgent(ws, async (msg) => { observed = msg; return { ok: 'done' }; }); await hub.execAction('10.0.0.1', 'reboot', { forced: false }); assert.equal(observed.type, 'reboot'); assert.equal(observed.baseIp, '10.0.0.1'); assert.equal(observed.forced, false); } finally { await closeAll(); } }); test('hub: phoneProbe RPC uses targetIp and returns probe summary', async () => { const { hub, port, closeAll } = await makeHub(); try { const ws = connectAgent(port); await waitOpen(ws); attachAutoAgent(ws, async (msg) => { assert.equal(msg.type, 'phone-probe'); assert.equal(msg.targetIp, '10.4.11.50'); return { probes: [{ path: '/admin/status.xml', status: 200, sizeBytes: 100 }], summary: { okCount: 1, total: 4, authOk: true, statusXmlFound: true }, statusXml: '', parsed: { fields: {} }, verdict: { healthy: true, warnings: [], info: [] }, }; }); const { result } = await hub.phoneProbe('10.4.11.50'); assert.equal(result.summary.statusXmlFound, true); assert.equal(result.probes[0].path, '/admin/status.xml'); } finally { await closeAll(); } });