refactor: run hwlab gateway through bun ts

2026-05-29 15:01:56 +08:00
parent a248e91b43
commit d7ff08f3f2
8 changed files with 232 additions and 8 deletions
@@ -0,0 +1,223 @@
+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.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.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(",");
+}
@@ -1,4 +1,4 @@
-#!/usr/bin/env node
+#!/usr/bin/env bun
 import os from "node:os";
 import { spawn } from "node:child_process";