pikasTech-HWLAB/cmd/hwlab-gateway/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";
import { setTimeout as delay } from "node:timers/promises";

const bunCommand = process.env.HWLAB_TEST_BUN_COMMAND || process.env.HWLAB_BUN_COMMAND || (process.versions.bun ? process.execPath : "bun");

test("gateway outbound transport registers session and keeps shell execution disabled by default", async () => {
  const cloud = await startFakeCloud();
  const gateway = await startGateway(cloud.port, {
    HWLAB_GATEWAY_SESSION_ID: "gws_gateway_disabled_test",
    HWLAB_GATEWAY_CMD_EXEC_ENABLED: ""
  });
  try {
    const registration = await cloud.waitForRegistration((item) => item.gatewaySessionId === "gws_gateway_disabled_test");
    assert.equal(registration.serviceId, "hwlab-gateway");
    assert.equal(registration.outbound.commandExecutionEnabled, false);

    const health = await fetchJson(`http://127.0.0.1:${gateway.port}/health/live`);
    assert.equal(health.serviceId, "hwlab-gateway");
    assert.equal(health.gatewaySessionId, "gws_gateway_disabled_test");

    const status = await fetchJson(`http://127.0.0.1:${gateway.port}/status`);
    assert.equal(status.outbound.commandExecutionEnabled, false);

    const capabilities = await fetchJson(`http://127.0.0.1:${gateway.port}/capabilities`);
    assert.equal(capabilities.capabilities[0].constraints.demoOpen, false);

    cloud.enqueue(shellRequest({ id: "req_disabled_shell", traceId: "trc_disabled_shell", command: "echo should-not-run" }));
    const result = await cloud.waitForResult("req_disabled_shell");
    assert.equal(result.response.error.data.reason, "cmd_exec_disabled");
    assert.equal(result.response.meta.environment, "v02");
    assert.equal(result.gatewaySessionId, "gws_gateway_disabled_test");
  } finally {
    await gateway.stop();
    await cloud.stop();
  }
});

test("gateway outbound transport executes cloud requests without blocking poll loop", async () => {
  const cloud = await startFakeCloud();
  const gateway = await startGateway(cloud.port, {
    HWLAB_GATEWAY_SESSION_ID: "gws_gateway_nonblocking_test",
    HWLAB_GATEWAY_CMD_EXEC_ENABLED: "1",
    HWLAB_GATEWAY_MAX_INFLIGHT: "2"
  });
  try {
    await cloud.waitForRegistration((item) => item.gatewaySessionId === "gws_gateway_nonblocking_test");
    cloud.enqueue(shellRequest({
      id: "req_slow_shell",
      traceId: "trc_slow_shell",
      command: "node -e \"setTimeout(()=>console.log('slow-done'),800)\""
    }));
    await cloud.waitForRegistration((item) => item.outbound.inflightCount >= 1);

    const quickStartedAt = Date.now();
    cloud.enqueue(shellRequest({ id: "req_quick_shell", traceId: "trc_quick_shell", command: "echo quick-done" }));
    const quick = await cloud.waitForResult("req_quick_shell");
    const quickDurationMs = Date.now() - quickStartedAt;
    const slow = await cloud.waitForResult("req_slow_shell");

    assert.equal(quick.response.result.shellExecuted, true, JSON.stringify(quick.response, null, 2));
    assert.equal(quick.response.result.exitCode, 0, JSON.stringify(quick.response, null, 2));
    assert.equal(quick.response.meta.environment, "v02");
    assert.equal(quick.response.result.environment, "v02");
    assert.equal(quick.response.result.audit.environment, "v02");
    assert.equal(quick.response.result.evidence.environment, "v02");
    assert.match(quick.response.result.stdout, /quick-done/u);
    assert.equal(slow.response.result.shellExecuted, true, JSON.stringify(slow.response, null, 2));
    assert.equal(slow.response.result.exitCode, 0, JSON.stringify(slow.response, null, 2));
    assert.match(slow.response.result.stdout, /slow-done/u);
    assert.ok(quick.receivedAt < slow.receivedAt, "quick request should complete before the earlier slow request");
    assert.ok(quickDurationMs < 700, `quick request was blocked for ${quickDurationMs}ms`);
    assert.ok(cloud.registrations.some((item) => item.outbound.inflightCount >= 1));
  } finally {
    await gateway.stop();
    await cloud.stop();
  }
});

async function startGateway(cloudPort: number, env: Record<string, string>) {
  const port = await freePort();
  const child = spawn(bunCommand, ["run", "cmd/hwlab-gateway/main.ts"], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      ...env,
      PORT: String(port),
      HWLAB_GATEWAY_ID: env.HWLAB_GATEWAY_ID || "gtw_gateway_test",
      HWLAB_GATEWAY_CLOUD_URL: `http://127.0.0.1:${cloudPort}`,
      HWLAB_GATEWAY_POLL_INTERVAL_MS: "20",
      HWLAB_GATEWAY_CMD_TIMEOUT_MS: "3000",
      HWLAB_GATEWAY_CMD_OUTPUT_LIMIT_BYTES: "4096",
      NO_PROXY: mergeNoProxy(process.env.NO_PROXY, "127.0.0.1,localhost,::1"),
      no_proxy: mergeNoProxy(process.env.no_proxy, "127.0.0.1,localhost,::1")
    },
    stdio: ["ignore", "pipe", "pipe"]
  });
  await waitForListening(child, "hwlab-gateway");
  return {
    port,
    stop: () => stopChild(child)
  };
}

async function startFakeCloud() {
  const requests: any[] = [];
  const registrations: any[] = [];
  const results: any[] = [];
  const server = createServer(async (request, response) => {
    try {
      const body = await readJson(request);
      if (request.method === "POST" && request.url === "/v1/gateway/poll") {
        registrations.push({ ...body, receivedAt: Date.now() });
        sendJson(response, 200, { ok: true, request: requests.shift() ?? null });
        return;
      }
      if (request.method === "POST" && request.url === "/v1/gateway/result") {
        results.push({ ...body, receivedAt: Date.now() });
        sendJson(response, 200, { ok: true });
        return;
      }
      sendJson(response, 404, { error: "not_found" });
    } catch (error) {
      sendJson(response, 500, { error: error instanceof Error ? error.message : String(error) });
    }
  });
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
  return {
    port: (server.address() as any).port as number,
    registrations,
    enqueue: (rpcRequest: any) => requests.push(rpcRequest),
    waitForRegistration: (predicate: (item: any) => boolean) => waitFor(() => registrations.find(predicate), "gateway registration"),
    waitForResult: (id: string) => waitFor(() => results.find((item) => item.response?.id === id), `gateway result ${id}`),
    stop: () => new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()))
  };
}

function shellRequest({ id, traceId, command }: { id: string; traceId: string; command: string }) {
  return {
    jsonrpc: "2.0",
    id,
    method: "hardware.invoke.shell",
    params: {
      projectId: "prj_v02_device_pod",
      gatewaySessionId: "gws_gateway_test",
      resourceId: "res_d601_windows_host",
      capabilityId: "cap_d601_windows_cmd_exec",
      operationId: `op_${id}`,
      input: { command, cwd: process.cwd(), timeoutMs: 3000 }
    },
    meta: { traceId, actorId: "svc_hwlab-device-pod", serviceId: "hwlab-cloud-api", environment: "v02" }
  };
}

async function waitFor<T>(read: () => T | undefined, label: string, timeoutMs = 5000): Promise<T> {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const value = read();
    if (value) return value;
    await delay(20);
  }
  throw new Error(`timed out waiting for ${label}`);
}

async function fetchJson(url: string) {
  const response = await fetch(url, { headers: { accept: "application/json" } });
  assert.equal(response.ok, true, `${url} returned ${response.status}`);
  return response.json();
}

function readJson(request: any) {
  return new Promise<any>((resolve, reject) => {
    const chunks: Buffer[] = [];
    request.on("data", (chunk: Buffer) => chunks.push(Buffer.from(chunk)));
    request.on("end", () => {
      const text = Buffer.concat(chunks).toString("utf8").trim();
      resolve(text ? JSON.parse(text) : {});
    });
    request.on("error", reject);
  });
}

function sendJson(response: any, statusCode: number, body: any) {
  const payload = JSON.stringify(body);
  response.writeHead(statusCode, { "content-type": "application/json", "content-length": Buffer.byteLength(payload) });
  response.end(payload);
}

async function freePort() {
  const server = createServer();
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
  const port = (server.address() as any).port as number;
  await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  return port;
}

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

async function stopChild(child: any) {
  if (child.exitCode !== null || child.signalCode !== null) return;
  child.kill("SIGTERM");
  await Promise.race([
    new Promise<void>((resolve) => child.on("exit", () => resolve())),
    delay(1000).then(() => child.kill("SIGKILL"))
  ]);
}

function mergeNoProxy(...values: Array<string | undefined>) {
  return [...new Set(values.flatMap((value) => String(value ?? "").split(",").map((item) => item.trim()).filter(Boolean)))].join(",");
}