kuber/tests/server/exec-service.test.ts

import { describe, expect, test } from "bun:test";
import {
  createExecService,
  EXEC_MANAGED_BY_LABEL,
  EXEC_MANAGED_BY_VALUE,
  EXEC_WORKSPACE_UID_LABEL,
  ExecOutputLimitError,
  type ExecChunk,
  type ExecDeployment,
  type ExecPod,
  type KubernetesExecBackend,
  type KubernetesExecExit,
  type KubernetesExecProcess,
  type KubernetesExecRequest,
} from "../../server/exec-service";

const workspace = { project: "shop", uid: "workspace-1" };

async function* chunks(...values: ExecChunk[]): AsyncGenerator<ExecChunk> {
  for (const value of values) yield value;
}

function deferred<T>() {
  let resolve!: (value: T) => void;
  let reject!: (reason?: unknown) => void;
  const promise = new Promise<T>((resolvePromise, rejectPromise) => {
    resolve = resolvePromise;
    reject = rejectPromise;
  });
  return { promise, resolve, reject };
}

class FakeProcess implements KubernetesExecProcess {
  stdout: AsyncIterable<ExecChunk> = chunks();
  stderr: AsyncIterable<ExecChunk> = chunks();
  status: Promise<KubernetesExecExit> = Promise.resolve({ exitCode: 0 });
  stdin: Uint8Array[] = [];
  resizes: Array<[number, number]> = [];
  stdinClosed = false;
  closeCalls = 0;

  writeStdin(data: Uint8Array) {
    this.stdin.push(data);
  }

  closeStdin() {
    this.stdinClosed = true;
  }

  resize(columns: number, rows: number) {
    this.resizes.push([columns, rows]);
  }

  wait() {
    return this.status;
  }

  close() {
    this.closeCalls += 1;
  }
}

class FakeBackend implements KubernetesExecBackend {
  deployment: ExecDeployment | undefined = {
    name: "api",
    uid: "deployment-1",
    labels: {
      [EXEC_MANAGED_BY_LABEL]: EXEC_MANAGED_BY_VALUE,
      [EXEC_WORKSPACE_UID_LABEL]: workspace.uid,
    },
    selector: { app: "api" },
    containers: ["api", "sidecar"],
  };
  pods: ExecPod[] = [
    {
      name: "api-old",
      uid: "pod-old",
      deploymentUid: "deployment-1",
      phase: "Running",
      containers: [{ name: "api", running: true, ready: false }],
    },
    {
      name: "api-new",
      uid: "pod-new",
      deploymentUid: "deployment-1",
      phase: "Running",
      containers: [
        { name: "api", running: true, ready: true },
        { name: "sidecar", running: true },
      ],
    },
  ];
  process = new FakeProcess();
  requests: KubernetesExecRequest[] = [];
  signals: AbortSignal[] = [];
  selectors: Array<Readonly<Record<string, string>>> = [];

  async getDeployment() {
    return this.deployment;
  }

  async listPods(
    _namespace: string,
    selector: Readonly<Record<string, string>>,
  ) {
    this.selectors.push(selector);
    return this.pods;
  }

  async exec(request: KubernetesExecRequest, signal: AbortSignal) {
    this.requests.push(request);
    this.signals.push(signal);
    return this.process;
  }
}

function input(overrides: Record<string, unknown> = {}) {
  return {
    workspace,
    deployment: "api",
    command: ["sh", "-c", "printf ok"],
    ...overrides,
  };
}

async function allFrames<T>(iterable: AsyncIterable<T>): Promise<T[]> {
  const result: T[] = [];
  for await (const value of iterable) result.push(value);
  return result;
}

describe("ExecService target resolution", () => {
  test("selects a ready running pod owned by the named managed deployment", async () => {
    const backend = new FakeBackend();
    const target = await createExecService(backend).resolveTarget(input());
    expect(target).toEqual({
      namespace: "shop",
      deployment: "api",
      deploymentUid: "deployment-1",
      pod: "api-new",
      podUid: "pod-new",
      container: "api",
    });
    expect(backend.selectors).toEqual([{ app: "api" }]);
  });

  test("rejects unmanaged and differently owned deployments", async () => {
    const backend = new FakeBackend();
    backend.deployment = {
      ...backend.deployment!,
      labels: { [EXEC_MANAGED_BY_LABEL]: EXEC_MANAGED_BY_VALUE },
    };
    await expect(
      createExecService(backend).resolveTarget(input()),
    ).rejects.toMatchObject({
      code: "EXEC_TARGET_FORBIDDEN",
    });
    backend.deployment.labels = {
      [EXEC_MANAGED_BY_LABEL]: "someone-else",
      [EXEC_WORKSPACE_UID_LABEL]: workspace.uid,
    };
    await expect(
      createExecService(backend).resolveTarget(input()),
    ).rejects.toMatchObject({
      code: "EXEC_TARGET_FORBIDDEN",
    });
  });

  test("does not trust selector collisions or terminating pods", async () => {
    const backend = new FakeBackend();
    backend.pods = [
      {
        name: "foreign",
        uid: "foreign-pod",
        deploymentUid: "different-deployment",
        phase: "Running",
        containers: [{ name: "api", running: true, ready: true }],
      },
      {
        name: "terminating",
        uid: "terminating-pod",
        deploymentUid: "deployment-1",
        phase: "Running",
        deletionTimestamp: "2026-09-02T00:00:00Z",
        containers: [{ name: "api", running: true, ready: true }],
      },
    ];
    await expect(
      createExecService(backend).resolveTarget(input()),
    ).rejects.toMatchObject({
      code: "EXEC_TARGET_NOT_READY",
    });
  });

  test("supports an explicit container and rejects one absent from the pod", async () => {
    const backend = new FakeBackend();
    expect(
      await createExecService(backend).resolveTarget(
        input({ container: "sidecar" }),
      ),
    ).toMatchObject({ container: "sidecar" });
    await expect(
      createExecService(backend).resolveTarget(input({ container: "missing" })),
    ).rejects.toMatchObject({ code: "EXEC_TARGET_NOT_FOUND" });
  });
});

describe("ExecService non-TTY execution", () => {
  test("captures stdout, stderr, and the remote exit status", async () => {
    const backend = new FakeBackend();
    backend.process.stdout = chunks("hel", new TextEncoder().encode("lo"));
    backend.process.stderr = chunks("warning\n");
    backend.process.status = Promise.resolve({
      exitCode: 7,
      reason: "NonZeroExitCode",
    });
    const result = await createExecService(backend).execute(input());
    expect(result).toMatchObject({
      pod: "api-new",
      container: "api",
      stdout: "hello",
      stderr: "warning\n",
      exitCode: 7,
      reason: "NonZeroExitCode",
    });
    expect(backend.requests[0]).toMatchObject({ tty: false });
  });

  test("enforces independent output caps and closes the process", async () => {
    const backend = new FakeBackend();
    backend.process.stdout = chunks("123", "456");
    await expect(
      createExecService(backend, {
        maxOutputCapBytes: 10,
        defaultOutputCapBytes: 10,
      }).execute(input({ stdoutCapBytes: 5 })),
    ).rejects.toBeInstanceOf(ExecOutputLimitError);
    expect(backend.process.closeCalls).toBeGreaterThan(0);
    expect(backend.signals[0]?.aborted).toBe(true);
  });

  test("propagates cancellation and cleans up the process", async () => {
    const backend = new FakeBackend();
    const status = deferred<KubernetesExecExit>();
    backend.process.status = status.promise;
    const controller = new AbortController();
    const execution = createExecService(backend).execute(
      input({ signal: controller.signal }),
    );
    await Bun.sleep(1);
    controller.abort();
    status.reject(new Error("cancelled"));
    await expect(execution).rejects.toMatchObject({ code: "EXEC_ABORTED" });
    expect(backend.process.closeCalls).toBeGreaterThan(0);
  });
});

describe("ExecService validation", () => {
  test("validates names, command count, command bytes, and requested caps before exec", async () => {
    const backend = new FakeBackend();
    const service = createExecService(backend, {
      maxCommandArguments: 2,
      maxCommandBytes: 6,
      maxArgumentBytes: 4,
      maxOutputCapBytes: 10,
      defaultOutputCapBytes: 10,
    });
    await expect(
      service.execute(input({ deployment: "Bad_Name" })),
    ).rejects.toMatchObject({
      code: "EXEC_INVALID",
    });
    await expect(
      service.execute(input({ command: ["a", "b", "c"] })),
    ).rejects.toThrow("more than 2");
    await expect(
      service.execute(input({ command: ["12345"] })),
    ).rejects.toThrow("argument exceeds");
    await expect(
      service.execute(input({ command: ["1234", "1234"] })),
    ).rejects.toThrow("Command exceeds");
    await expect(
      service.execute(input({ command: ["ok"], stdoutCapBytes: 11 })),
    ).rejects.toThrow("stdoutCapBytes");
    expect(backend.requests).toHaveLength(0);
  });
});

describe("ExecService interactive sessions", () => {
  test("maps process streams, stdin, resize, EOF, and exit into duplex frames", async () => {
    const backend = new FakeBackend();
    backend.process.stdout = chunks("out");
    backend.process.stderr = chunks(new TextEncoder().encode("err"));
    backend.process.status = Promise.resolve({
      exitCode: 3,
      message: "finished",
    });
    const controller = new AbortController();
    const session = await createExecService(backend).openInteractive({
      ...input(),
      signal: controller.signal,
    });
    await session.send({ type: "stdin", data: "hello", eof: true });
    await session.send({ type: "resize", columns: 120, rows: 40 });
    const frames = await allFrames(session);
    expect(frames.map((frame) => frame.type).sort()).toEqual([
      "exit",
      "stderr",
      "stdout",
    ]);
    expect(new TextDecoder().decode(backend.process.stdin[0])).toBe("hello");
    expect(backend.process.stdinClosed).toBe(true);
    expect(backend.process.resizes).toEqual([[120, 40]]);
    expect(frames.find((frame) => frame.type === "exit")).toMatchObject({
      exitCode: 3,
      message: "finished",
    });
    expect(backend.requests[0]).toMatchObject({ tty: true });
  });

  test("supports streaming sessions without allocating a TTY", async () => {
    const backend = new FakeBackend();
    const session = await createExecService(backend).openInteractive({
      ...input(),
      tty: false,
      signal: new AbortController().signal,
    });
    await allFrames(session);
    expect(backend.requests[0]).toMatchObject({ tty: false });
  });

  test("aborts and closes without emitting an error frame when the API signal ends", async () => {
    const backend = new FakeBackend();
    const status = deferred<KubernetesExecExit>();
    backend.process.status = status.promise;
    backend.process.stdout = (async function* () {
      yield "before-abort";
      await status.promise;
    })();
    const controller = new AbortController();
    const session = await createExecService(backend).openInteractive({
      ...input(),
      signal: controller.signal,
    });
    const iterator = session[Symbol.asyncIterator]();
    expect((await iterator.next()).value?.type).toBe("stdout");
    controller.abort();
    status.reject(new Error("transport closed"));
    expect((await iterator.next()).done).toBe(true);
    await Bun.sleep(1);
    expect(backend.process.closeCalls).toBeGreaterThan(0);
  });

  test("validates interactive input frames", async () => {
    const backend = new FakeBackend();
    const status = deferred<KubernetesExecExit>();
    backend.process.status = status.promise;
    const session = await createExecService(backend, {
      maxStdinFrameBytes: 3,
    }).openInteractive({
      ...input(),
      signal: new AbortController().signal,
    });
    await expect(session.send({ type: "stdin", data: "four" })).rejects.toThrow(
      "stdin frame",
    );
    await expect(
      session.send({ type: "resize", columns: 0, rows: 24 }),
    ).rejects.toThrow("Terminal dimensions");
    status.resolve({ exitCode: 0 });
    await allFrames(session);
  });
});