pikasTech-HWLAB/cmd/hwlab-edge-proxy/main.test.ts

import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { createServer } from "node:http";
import test from "node:test";

const nodeCommand = process.env.HWLAB_TEST_NODE_COMMAND || process.env.HWLAB_NODE_COMMAND || "node";

test("edge proxy reports local health and proxies live health to upstream", async () => {
  const upstream = await startUpstream();
  const proxy = await startEdgeProxy(upstream.port);
  try {
    const health = await fetchJson(`http://127.0.0.1:${proxy.port}/health`);
    assert.equal(health.serviceId, "hwlab-edge-proxy");
    assert.equal(health.status, "ok");
    assert.equal(health.service.role, "public-dev-ingress");
    assert.equal(health.details.upstream, `http://127.0.0.1:${upstream.port}`);
    assert.equal(health.details.idleTimeoutSeconds, 120);
    assert.equal(health.details.mode, "dev-edge-proxy");

    const routes = await fetchJson(`http://127.0.0.1:${proxy.port}/routes`);
    assert.equal(routes.serviceId, "hwlab-edge-proxy");
    assert.deepEqual(routes.routes, [{
      pathPrefix: "/",
      upstream: `http://127.0.0.1:${upstream.port}`,
      mode: "http-proxy"
    }]);

    const live = await fetchJson(`http://127.0.0.1:${proxy.port}/health/live`);
    assert.equal(live.serviceId, "hwlab-cloud-api");
    assert.equal(live.environment, "v02");
    assert.equal(upstream.captured.method, "GET");
    assert.equal(upstream.captured.url, "/health/live");

  } finally {
    await proxy.stop();
    await upstream.stop();
  }
});

test("edge proxy proxies upstream port 6667 without using fetch bad-port handling", async () => {
  const upstream = await startUpstream(6667);
  const proxy = await startEdgeProxy(upstream.port);
  try {
    const live = await fetchJson(`http://127.0.0.1:${proxy.port}/health/live`);
    assert.equal(live.serviceId, "hwlab-cloud-api");
    assert.equal(upstream.captured.url, "/health/live");
  } finally {
    await proxy.stop();
    await upstream.stop();
  }
});

test("edge proxy tunnels websocket upgrade to upstream", async () => {
  const upstream = await startWebSocketUpstream();
  const proxy = await startEdgeProxy(upstream.port);
  try {
    const observed = await openWebSocket(`ws://127.0.0.1:${proxy.port}/v1/hwpod-node/ws`, () => ({ proxyStderr: proxy.stderr, upstreamEvents: upstream.events }));
    assert.equal(observed.type, "welcome");
    assert.equal(observed.path, "/v1/hwpod-node/ws");
  } finally {
    await proxy.stop();
    await upstream.stop();
  }
});

async function startUpstream(port = 0) {
  let captured = null;
  const server = createServer(async (request, response) => {
    const chunks = [];
    for await (const chunk of request) chunks.push(chunk);
    captured = {
      method: request.method,
      url: request.url,
      body: Buffer.concat(chunks).toString("utf8")
    };
    response.writeHead(200, { "content-type": "application/json" });
    response.end(JSON.stringify({
      serviceId: "hwlab-cloud-api",
      environment: "v02",
      status: "live"
    }));
  });
  await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve));
  return {
    port: server.address().port,
    get captured() {
      return captured;
    },
    stop: () => new Promise((resolve) => server.close(resolve))
  };
}

async function startWebSocketUpstream() {
  const events = [];
  const server = Bun.serve({
    hostname: "127.0.0.1",
    port: 0,
    fetch(request, bunServer) {
      const url = new URL(request.url);
      events.push({ event: "upgrade", url: url.pathname });
      const upgraded = bunServer.upgrade(request, { data: { path: url.pathname } });
      return upgraded ? undefined : new Response("upgrade required", { status: 426 });
    },
    websocket: {
      open(socket) {
        setTimeout(() => {
          socket.send(JSON.stringify({ type: "welcome", path: socket.data.path }));
        }, 50);
      }
    }
  });
  return {
    port: server.port,
    events,
    stop: () => server.stop(true)
  };
}

async function openWebSocket(url, debug = () => ({})) {
  return await new Promise((resolve, reject) => {
    const socket = new WebSocket(url);
    let finished = false;
    const done = (error, value) => {
      if (finished) return;
      finished = true;
      clearTimeout(timer);
      try { socket.close(); } catch {}
      if (error) reject(error);
      else resolve(value);
    };
    const timer = setTimeout(() => done(new Error(`websocket did not receive message: ${url}; debug=${JSON.stringify(debug())}`)), 2500);
    socket.addEventListener("message", (event) => {
      dataToText(event.data)
        .then((text) => done(null, JSON.parse(text)))
        .catch((error) => done(error));
    });
    socket.addEventListener("error", () => done(new Error(`websocket error: ${url}`)));
    socket.addEventListener("close", (event) => done(new Error(`websocket closed before message: code=${event.code} reason=${event.reason || ""}`)));
  });
}

async function dataToText(data) {
  if (typeof data === "string") return data;
  if (data instanceof ArrayBuffer) return Buffer.from(data).toString("utf8");
  if (ArrayBuffer.isView(data)) return Buffer.from(data.buffer, data.byteOffset, data.byteLength).toString("utf8");
  if (data && typeof data.text === "function") return await data.text();
  return String(data);
}

async function startEdgeProxy(upstreamPort) {
  const port = await freePort();
  let stderr = "";
  const child = spawn(nodeCommand, ["cmd/hwlab-edge-proxy/main.mjs"], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      HWLAB_EDGE_HOST: "127.0.0.1",
      HWLAB_EDGE_PORT: String(port),
      HWLAB_EDGE_UPSTREAM: `http://127.0.0.1:${upstreamPort}`
    },
    stdio: ["ignore", "pipe", "pipe"]
  });
  child.stderr.on("data", (chunk) => {
    stderr += chunk;
  });
  await waitForListening(child, "edge proxy");
  return {
    port,
    get stderr() {
      return stderr;
    },
    stop: () => stopChild(child)
  };
}

async function fetchJson(url, options = {}) {
  const response = await fetch(url, options);
  assert.equal(response.ok, true, `${url} returned ${response.status}`);
  return response.json();
}

async function freePort() {
  const server = createServer();
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
  const port = server.address().port;
  await new Promise((resolve) => server.close(resolve));
  return port;
}

async function waitForListening(child, label) {
  let stderr = "";
  child.stderr.on("data", (chunk) => {
    stderr += chunk;
  });
  await new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error(`${label} did not start: ${stderr}`)), 5000);
    child.stdout.on("data", (chunk) => {
      if (chunk.toString().includes("listening")) {
        clearTimeout(timer);
        resolve();
      }
    });
    child.on("exit", (code) => {
      clearTimeout(timer);
      reject(new Error(`${label} exited early code=${code}: ${stderr}`));
    });
  });
}

async function stopChild(child) {
  if (child.exitCode !== null) return;
  child.kill("SIGTERM");
  await new Promise((resolve) => child.once("exit", resolve));
}