From 303e764603125bda463ae9f2271e7d94fc9bd569 Mon Sep 17 00:00:00 2001 From: Ming Wen Date: Mon, 20 Apr 2026 08:20:13 +0800 Subject: [PATCH] test(e2e): scaffold tests/e2e harness + first smoke case MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per plan §12. The harness writes a fresh YAML config per test, picks two free ports, spawns the aisix binary, and waits up to 10s for both `/health` and `/admin/v1/health` to respond. Each test gets a unique etcd prefix (`/aisix-e2e-`) and best-effort cleanup on exit. Modules: - harness/app.ts — spawn + wait + cleanup - harness/etcd.ts — JSON gateway client (ping, prefix delete) - harness/admin.ts — typed Admin API client - harness/proxy.ts — typed Proxy client - harness/upstream-openai.ts — node http mock with stream support - harness/http.ts — undici wrapper that bypasses HTTP_PROXY env First case: admin POST Model + ApiKey → wait 500ms → GET /v1/models sees the model; chat completion forwards to the mock upstream's /v1/chat/completions. CI workflow already wires this in (build-bin → e2e job with etcd service). The job stays `continue-on-error: true` until the harness proves stable across a few CI runs; we'll tighten the gate then. Notes for future work: - Anthropic mock upstream still TODO - SSE assertion helpers (waitForChunks, expectDoneEvent) still TODO - Coverage instrumentation (RUSTFLAGS=instrument-coverage path) handled by the existing build-bin job; the harness only needs to invoke the binary. --- .github/workflows/ci.yml | 5 +- tests/e2e/.gitignore | 5 + tests/e2e/package.json | 23 +++ tests/e2e/src/cases/smoke.test.ts | 97 ++++++++++++ tests/e2e/src/harness/admin.ts | 59 ++++++++ tests/e2e/src/harness/app.ts | 180 +++++++++++++++++++++++ tests/e2e/src/harness/etcd.ts | 78 ++++++++++ tests/e2e/src/harness/http.ts | 56 +++++++ tests/e2e/src/harness/index.ts | 6 + tests/e2e/src/harness/ports.ts | 30 ++++ tests/e2e/src/harness/proxy.ts | 51 +++++++ tests/e2e/src/harness/upstream-openai.ts | 130 ++++++++++++++++ tests/e2e/tsconfig.json | 14 ++ tests/e2e/vitest.config.ts | 21 +++ 14 files changed, 753 insertions(+), 2 deletions(-) create mode 100644 tests/e2e/.gitignore create mode 100644 tests/e2e/package.json create mode 100644 tests/e2e/src/cases/smoke.test.ts create mode 100644 tests/e2e/src/harness/admin.ts create mode 100644 tests/e2e/src/harness/app.ts create mode 100644 tests/e2e/src/harness/etcd.ts create mode 100644 tests/e2e/src/harness/http.ts create mode 100644 tests/e2e/src/harness/index.ts create mode 100644 tests/e2e/src/harness/ports.ts create mode 100644 tests/e2e/src/harness/proxy.ts create mode 100644 tests/e2e/src/harness/upstream-openai.ts create mode 100644 tests/e2e/tsconfig.json create mode 100644 tests/e2e/vitest.config.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f837e69d..22e02f8b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -118,8 +118,9 @@ jobs: name: e2e (vitest) + coverage needs: [build-bin, build-ui] runs-on: ubuntu-latest - # Scaffold-phase soft gate: the harness lands with PR #2+. - # Flip `continue-on-error` to false at PR #5 per plan §5.2 ratchet. + # Soft gate while the harness stabilises across CI runners (etcd + # service startup, port detection, undici quirks). Flip to `false` + # once we have ~5 consecutive green runs on main. continue-on-error: true services: etcd: diff --git a/tests/e2e/.gitignore b/tests/e2e/.gitignore new file mode 100644 index 00000000..6ed3f905 --- /dev/null +++ b/tests/e2e/.gitignore @@ -0,0 +1,5 @@ +node_modules/ +coverage/ +*.tsbuildinfo +package-lock.json +pnpm-lock.yaml diff --git a/tests/e2e/package.json b/tests/e2e/package.json new file mode 100644 index 00000000..099cc4a6 --- /dev/null +++ b/tests/e2e/package.json @@ -0,0 +1,23 @@ +{ + "name": "aisix-e2e", + "version": "0.1.0", + "private": true, + "type": "module", + "description": "End-to-end test harness for the aisix AI Gateway (spec §12).", + "scripts": { + "test": "vitest run", + "test:watch": "vitest", + "coverage": "vitest run --coverage" + }, + "devDependencies": { + "@types/node": "20.11.30", + "@vitest/coverage-v8": "2.1.9", + "typescript": "5.5.4", + "vitest": "2.1.9" + }, + "dependencies": { + "openai": "4.65.0", + "undici": "6.19.8", + "yaml": "2.5.1" + } +} diff --git a/tests/e2e/src/cases/smoke.test.ts b/tests/e2e/src/cases/smoke.test.ts new file mode 100644 index 00000000..bb2f7962 --- /dev/null +++ b/tests/e2e/src/cases/smoke.test.ts @@ -0,0 +1,97 @@ +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + EtcdClient, + ProxyClient, + spawnApp, + startOpenAiUpstream, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +describe("smoke: admin write → proxy read", () => { + let app: SpawnedApp | undefined; + let upstream: OpenAiUpstream | undefined; + let admin: AdminClient | undefined; + let etcdReachable = false; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + upstream = await startOpenAiUpstream(); + app = await spawnApp(); + admin = new AdminClient(app.adminUrl, app.adminKey); + }); + + afterAll(async () => { + await app?.exit(); + await upstream?.close(); + }); + + test("a Model + ApiKey written via Admin API are visible to /v1/models", async (ctx) => { + if (!etcdReachable || !app || !admin || !upstream) { + ctx.skip(); + return; + } + + await admin.createModel({ + name: "smoke-gpt", + model: "openai/gpt-4o-mini", + // The OpenAI bridge appends `/chat/completions`, so the api_base + // already needs the `/v1` segment to land on `/v1/chat/completions`. + provider_config: { api_key: "sk-mock", api_base: `${upstream.baseUrl}/v1` }, + }); + await admin.createApiKey({ + key: "sk-smoke-caller", + allowed_models: ["smoke-gpt"], + }); + + await waitConfigPropagation(); + + const proxy = new ProxyClient(app.proxyUrl, "sk-smoke-caller"); + const { status, body } = await proxy.listModels(); + + expect(status).toBe(200); + expect(body).toMatchObject({ + object: "list", + data: expect.arrayContaining([expect.objectContaining({ id: "smoke-gpt" })]), + }); + }); + + test("a chat completion forwards to the mock upstream", async (ctx) => { + if (!etcdReachable || !app || !upstream) { + ctx.skip(); + return; + } + + const proxy = new ProxyClient(app.proxyUrl, "sk-smoke-caller"); + const { status, body } = await proxy.chat({ + model: "smoke-gpt", + messages: [{ role: "user", content: "hello" }], + }); + + if (status !== 200) { + throw new Error( + `chat returned ${status}: ${JSON.stringify(body)}\n upstream paths: ${JSON.stringify(upstream.receivedRequests.map((r) => r.path))}`, + ); + } + expect(body).toMatchObject({ + object: "chat.completion", + choices: expect.arrayContaining([ + expect.objectContaining({ + message: expect.objectContaining({ role: "assistant" }), + }), + ]), + }); + + const seen = upstream.receivedRequests.some((r) => + r.path.startsWith("/v1/chat/completions"), + ); + if (!seen) { + throw new Error( + `upstream did not receive /v1/chat/completions; saw paths: ${JSON.stringify(upstream.receivedRequests.map((r) => `${r.method} ${r.path}`))}`, + ); + } + }); +}); diff --git a/tests/e2e/src/harness/admin.ts b/tests/e2e/src/harness/admin.ts new file mode 100644 index 00000000..2c04a379 --- /dev/null +++ b/tests/e2e/src/harness/admin.ts @@ -0,0 +1,59 @@ +import { harnessRequest } from "./http.js"; + +/** + * Thin typed wrapper over the Admin API. Keeps the test surface readable + * — `await admin.createModel({...})` instead of inlined fetch boilerplate. + */ +export class AdminClient { + constructor( + private readonly baseUrl: string, + private readonly adminKey: string, + ) {} + + async createModel( + model: Record, + ): Promise<{ id: string; value: Record }> { + return this.json("POST", "/admin/v1/models", model); + } + + async createApiKey( + key: Record, + ): Promise<{ id: string; value: Record }> { + return this.json("POST", "/admin/v1/apikeys", key); + } + + async listModels(): Promise>> { + const res = await this.json<{ items?: Array<{ value: Record }> }>( + "GET", + "/admin/v1/models", + ); + return (res.items ?? []).map((entry) => entry.value); + } + + async json>( + method: string, + path: string, + body?: unknown, + ): Promise { + const res = await harnessRequest(`${this.baseUrl}${path}`, { + method, + headers: { + authorization: `Bearer ${this.adminKey}`, + "content-type": "application/json", + }, + body: body === undefined ? undefined : JSON.stringify(body), + }); + const text = await res.body.text(); + if (res.statusCode >= 300) { + throw new Error( + `admin ${method} ${path} → ${res.statusCode}: ${text.slice(0, 512)}`, + ); + } + return text ? (JSON.parse(text) as T) : ({} as T); + } +} + +/** Convenience: wait the spec-mandated 500ms for snapshot propagation. */ +export function waitConfigPropagation(): Promise { + return new Promise((r) => setTimeout(r, 500)); +} diff --git a/tests/e2e/src/harness/app.ts b/tests/e2e/src/harness/app.ts new file mode 100644 index 00000000..95317219 --- /dev/null +++ b/tests/e2e/src/harness/app.ts @@ -0,0 +1,180 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { stringify as yamlStringify } from "yaml"; + +import { pickFreePorts } from "./ports.js"; +import { EtcdClient } from "./etcd.js"; +import { harnessRequest } from "./http.js"; + +export interface AppOverrides { + /** Inserted into `admin.admin_keys`. Defaults to a fresh random key. */ + adminKey?: string; + /** Whether to enable Prometheus on `/metrics`. Defaults to true. */ + prometheus?: boolean; + /** Extra raw config keys merged into the YAML at the top level. */ + extra?: Record; +} + +export interface SpawnedApp { + proxyUrl: string; + adminUrl: string; + adminKey: string; + etcdPrefix: string; + exit(): Promise; +} + +const BIN_PATH = + process.env.AISIX_BIN ?? join(process.cwd(), "..", "..", "target", "debug", "aisix"); +const READY_TIMEOUT_MS = 10_000; +const SHUTDOWN_GRACE_MS = 3_000; + +/** + * Per-test handle to a spawned `aisix` binary. Each call writes a fresh + * config YAML into a tmp dir, picks two free ports, picks a unique etcd + * prefix, and waits up to 10s for `/health` on both ports to respond + * 200. `exit()` issues SIGTERM and waits up to 3s, escalating to SIGKILL. + */ +export async function spawnApp(overrides: AppOverrides = {}): Promise { + const etcd = new EtcdClient(); + if (!(await etcd.ping())) { + throw new Error( + `etcd not reachable at ${process.env.AISIX_E2E_ETCD ?? "http://127.0.0.1:2379"} ` + + "(set AISIX_E2E_ETCD or run `docker run --rm -p 2379:2379 quay.io/coreos/etcd:v3.5.15`)", + ); + } + + const [proxyPort, adminPort] = await pickFreePorts(2); + const adminKey = overrides.adminKey ?? `admin-${randomUUID()}`; + const etcdPrefix = `/aisix-e2e-${randomUUID()}`; + + const cfg = { + etcd: { + endpoints: [process.env.AISIX_E2E_ETCD ?? "http://127.0.0.1:2379"], + prefix: etcdPrefix, + dial_timeout_ms: 5000, + request_timeout_ms: 5000, + }, + proxy: { addr: `127.0.0.1:${proxyPort}`, request_body_limit_bytes: 10485760 }, + admin: { addr: `127.0.0.1:${adminPort}`, admin_keys: [adminKey] }, + observability: { + service_name: "aisix-e2e", + log_level: "warn", + access_log: false, + metrics: { + prometheus: { enabled: overrides.prometheus ?? true, path: "/metrics" }, + otlp: { enabled: false, endpoint: "http://127.0.0.1:4317" }, + }, + tracing: { otlp: { enabled: false, endpoint: "http://127.0.0.1:4317", sample_ratio: 1 } }, + langfuse: { enabled: false }, + }, + cache: { backend: "memory" }, + ...(overrides.extra ?? {}), + }; + + const dir = await mkdtemp(join(tmpdir(), "aisix-e2e-")); + const cfgPath = join(dir, "config.yaml"); + await writeFile(cfgPath, yamlStringify(cfg), "utf8"); + + const child = spawn(BIN_PATH, ["--config", cfgPath], { + stdio: ["ignore", "pipe", "pipe"], + env: { ...process.env, RUST_LOG: process.env.RUST_LOG ?? "warn" }, + }); + + let stderrBuf = ""; + child.stderr?.on("data", (c: Buffer) => { + stderrBuf += c.toString("utf8"); + }); + child.stdout?.on("data", (c: Buffer) => { + stderrBuf += c.toString("utf8"); + }); + let exitErr: string | undefined; + child.once("exit", (code, signal) => { + if (code !== 0 && code !== null) { + exitErr = `aisix exited early with code=${code} signal=${signal}`; + } + }); + + const proxyUrl = `http://127.0.0.1:${proxyPort}`; + const adminUrl = `http://127.0.0.1:${adminPort}`; + + try { + await Promise.all([ + waitForReady(`${proxyUrl}/health`, READY_TIMEOUT_MS), + waitForReady(`${adminUrl}/admin/v1/health`, READY_TIMEOUT_MS, adminKey), + ]); + } catch (err) { + const detail = exitErr ?? "still running"; + const stderr = stderrBuf.slice(-2000); + await terminate(child); + await cleanup(etcd, etcdPrefix, dir); + throw new Error( + `${(err as Error).message}\n binary state: ${detail}\n stderr tail:\n${stderr}`, + ); + } + + return { + proxyUrl, + adminUrl, + adminKey, + etcdPrefix, + async exit() { + await terminate(child); + await cleanup(etcd, etcdPrefix, dir); + }, + }; +} + +async function waitForReady(url: string, timeoutMs: number, bearer?: string): Promise { + const deadline = Date.now() + timeoutMs; + let lastErr: unknown; + let lastStatus: number | undefined; + let attempts = 0; + while (Date.now() < deadline) { + attempts++; + try { + const headers: Record = {}; + if (bearer) headers.authorization = `Bearer ${bearer}`; + const res = await harnessRequest(url, { method: "GET", headers }); + lastStatus = res.statusCode; + if (res.statusCode === 200) { + await res.body.dump(); + return; + } + await res.body.dump(); + } catch (err) { + lastErr = err; + } + await sleep(100); + } + throw new Error( + `timed out waiting for ${url} after ${attempts} attempts (lastStatus=${lastStatus ?? "n/a"}): ${lastErr ?? "no response"}`, + ); +} + +async function terminate(child: ChildProcess): Promise { + if (child.exitCode !== null) return; + child.kill("SIGTERM"); + const exited = await Promise.race([ + new Promise((r) => child.once("exit", () => r(true))), + sleep(SHUTDOWN_GRACE_MS).then(() => false), + ]); + if (!exited && child.exitCode === null) { + child.kill("SIGKILL"); + await new Promise((r) => child.once("exit", () => r())); + } +} + +async function cleanup(etcd: EtcdClient, prefix: string, dir: string): Promise { + // Best-effort — never throw from cleanup. + await Promise.allSettled([ + etcd.deletePrefix(prefix), + rm(dir, { recursive: true, force: true }), + ]); +} + +function sleep(ms: number): Promise { + return new Promise((r) => setTimeout(r, ms)); +} diff --git a/tests/e2e/src/harness/etcd.ts b/tests/e2e/src/harness/etcd.ts new file mode 100644 index 00000000..97dd077c --- /dev/null +++ b/tests/e2e/src/harness/etcd.ts @@ -0,0 +1,78 @@ +import { harnessRequest } from "./http.js"; + +/** + * Minimal etcd v3 helper that talks to the JSON gRPC-gateway + * (`/v3/kv/*` endpoints). Avoids pulling a heavy etcd npm dependency. + */ +export class EtcdClient { + constructor( + private readonly endpoint: string = process.env.AISIX_E2E_ETCD ?? + "http://127.0.0.1:2379", + ) {} + + /** + * Best-effort connectivity probe — returns false if etcd isn't reachable. + * + * Beyond a 200 status we also confirm the response is JSON containing the + * expected `header.cluster_id` field. A stray Docker port-mapping or a + * dev-tool's "service unavailable" HTML page can return 200 to anything + * on port 2379 and we don't want to misidentify those as etcd. + */ + async ping(timeoutMs = 1000): Promise { + try { + const ctrl = new AbortController(); + const t = setTimeout(() => ctrl.abort(), timeoutMs); + const res = await harnessRequest(`${this.endpoint}/v3/maintenance/status`, { + method: "POST", + body: "{}", + headers: { "content-type": "application/json" }, + signal: ctrl.signal, + }); + clearTimeout(t); + if (res.statusCode !== 200) { + await res.body.dump(); + return false; + } + const text = await res.body.text(); + try { + const parsed = JSON.parse(text) as { header?: { cluster_id?: string } }; + return typeof parsed.header?.cluster_id === "string"; + } catch { + return false; + } + } catch { + return false; + } + } + + /** Delete every key under `prefix` (range delete in etcd v3 semantics). */ + async deletePrefix(prefix: string): Promise { + const key = Buffer.from(prefix, "utf8").toString("base64"); + const rangeEnd = prefixRangeEnd(prefix).toString("base64"); + const res = await harnessRequest(`${this.endpoint}/v3/kv/deleterange`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ key, range_end: rangeEnd }), + }); + if (res.statusCode >= 300) { + const body = await res.body.text(); + throw new Error(`etcd deleterange failed (${res.statusCode}): ${body}`); + } + } +} + +/** + * Calculate the etcd "range end" for a prefix scan: the prefix with its + * last byte incremented by one. Returned as a Buffer because the + * incremented byte may not be valid UTF-8. + */ +function prefixRangeEnd(prefix: string): Buffer { + const bytes = Array.from(Buffer.from(prefix, "utf8")); + for (let i = bytes.length - 1; i >= 0; i--) { + if (bytes[i] < 0xff) { + const head = bytes.slice(0, i); + return Buffer.from([...head, bytes[i] + 1]); + } + } + return Buffer.from([0]); +} diff --git a/tests/e2e/src/harness/http.ts b/tests/e2e/src/harness/http.ts new file mode 100644 index 00000000..29eee706 --- /dev/null +++ b/tests/e2e/src/harness/http.ts @@ -0,0 +1,56 @@ +import { Agent, request as undiciRequest, type Dispatcher } from "undici"; + +/** + * Build a fresh undici Agent for each request. We deliberately do NOT + * cache a module-level agent: a stuck/aborted ECONNREFUSED on the + * shared agent has been observed to poison subsequent polls during + * waitForReady (the agent caches the failure even after the listener + * comes up). One agent per request is cheap for the volumes involved + * and avoids the failure mode. + * + * The custom agent also bypasses any `HTTP_PROXY`/`ALL_PROXY` env vars + * that local dev tools (ClashX etc.) may set. + */ +function freshAgent(): Agent { + return new Agent({ connectTimeout: 2000, keepAliveTimeout: 1000 }); +} + +export type HttpMethod = Dispatcher.HttpMethod; + +const METHODS = new Set([ + "GET", + "POST", + "PUT", + "DELETE", + "PATCH", + "HEAD", + "OPTIONS", + "TRACE", + "CONNECT", +]); + +export function asMethod(m: string): HttpMethod { + const upper = m.toUpperCase() as HttpMethod; + if (!METHODS.has(upper)) throw new Error(`unsupported HTTP method: ${m}`); + return upper; +} + +export interface HarnessRequestOptions { + method?: HttpMethod | string; + headers?: Record; + body?: string | Buffer; + signal?: AbortSignal; +} + +export async function harnessRequest( + url: string, + opts: HarnessRequestOptions = {}, +): Promise { + return undiciRequest(url, { + method: asMethod(opts.method ?? "GET"), + headers: opts.headers, + body: opts.body, + signal: opts.signal, + dispatcher: freshAgent(), + }); +} diff --git a/tests/e2e/src/harness/index.ts b/tests/e2e/src/harness/index.ts new file mode 100644 index 00000000..d43529d2 --- /dev/null +++ b/tests/e2e/src/harness/index.ts @@ -0,0 +1,6 @@ +export { spawnApp, type SpawnedApp, type AppOverrides } from "./app.js"; +export { AdminClient, waitConfigPropagation } from "./admin.js"; +export { ProxyClient } from "./proxy.js"; +export { EtcdClient } from "./etcd.js"; +export { startOpenAiUpstream, type OpenAiUpstream, type ReceivedRequest } from "./upstream-openai.js"; +export { pickFreePort, pickFreePorts } from "./ports.js"; diff --git a/tests/e2e/src/harness/ports.ts b/tests/e2e/src/harness/ports.ts new file mode 100644 index 00000000..c2188097 --- /dev/null +++ b/tests/e2e/src/harness/ports.ts @@ -0,0 +1,30 @@ +import { createServer } from "node:net"; + +/** + * Ask the OS for a free TCP port by binding to :0, reading the assigned + * port, and closing the socket. There is a tiny race between close and + * the caller binding, but in practice the test process grabs the port + * fast enough that collisions are extremely rare. + */ +export async function pickFreePort(): Promise { + return new Promise((resolve, reject) => { + const srv = createServer(); + srv.once("error", reject); + srv.listen(0, "127.0.0.1", () => { + const address = srv.address(); + if (!address || typeof address === "string") { + reject(new Error("unexpected unix-socket address")); + return; + } + const port = address.port; + srv.close((err) => (err ? reject(err) : resolve(port))); + }); + }); +} + +/** Pick N free ports in sequence. */ +export async function pickFreePorts(n: number): Promise { + const out: number[] = []; + for (let i = 0; i < n; i++) out.push(await pickFreePort()); + return out; +} diff --git a/tests/e2e/src/harness/proxy.ts b/tests/e2e/src/harness/proxy.ts new file mode 100644 index 00000000..bc5cfbee --- /dev/null +++ b/tests/e2e/src/harness/proxy.ts @@ -0,0 +1,51 @@ +import { harnessRequest } from "./http.js"; + +/** + * Thin typed wrapper over the proxy surface. Tests that want full SDK + * compatibility can use the `openai` npm package directly with + * `{ baseURL: app.proxyUrl + "/v1" }` — this client is for the cases + * where we want to inspect raw status codes, headers, or non-OpenAI + * endpoints (e.g. `/v1/messages`, `/passthrough/...`). + */ +export class ProxyClient { + constructor( + private readonly baseUrl: string, + private readonly apiKey: string, + ) {} + + async listModels(): Promise<{ status: number; body: unknown }> { + return this.json("GET", "/v1/models"); + } + + async chat(body: unknown): Promise<{ status: number; body: unknown }> { + return this.json("POST", "/v1/chat/completions", body); + } + + private async json( + method: string, + path: string, + body?: unknown, + ): Promise<{ status: number; body: unknown }> { + const res = await harnessRequest(`${this.baseUrl}${path}`, { + method, + headers: { + authorization: `Bearer ${this.apiKey}`, + "content-type": "application/json", + }, + body: body === undefined ? undefined : JSON.stringify(body), + }); + const text = await res.body.text(); + return { + status: res.statusCode, + body: text ? safeParse(text) : null, + }; + } +} + +function safeParse(text: string): unknown { + try { + return JSON.parse(text); + } catch { + return text; + } +} diff --git a/tests/e2e/src/harness/upstream-openai.ts b/tests/e2e/src/harness/upstream-openai.ts new file mode 100644 index 00000000..3d385378 --- /dev/null +++ b/tests/e2e/src/harness/upstream-openai.ts @@ -0,0 +1,130 @@ +import { createServer, type Server } from "node:http"; +import { pickFreePort } from "./ports.js"; + +export interface OpenAiUpstreamOptions { + /** Returned for non-streaming chat/completions. */ + nonStreamBody?: unknown; + /** Sequence of SSE event payloads (already-stringified JSON or `[DONE]`). */ + streamEvents?: string[]; + /** Inserted before the response is written. */ + responseDelayMs?: number; + /** Inserted between SSE events. */ + eventDelayMs?: number; + /** Status code to return (default 200). */ + status?: number; + /** Body to return when `status` >= 400. */ + errorBody?: unknown; + /** Drop the connection after writing this many SSE events. */ + disconnectAfterEvents?: number; +} + +export interface OpenAiUpstream { + baseUrl: string; + receivedRequests: ReceivedRequest[]; + close(): Promise; +} + +export interface ReceivedRequest { + method: string; + path: string; + headers: Record; + body: string; +} + +/** + * Spins a node http server that mimics the OpenAI surface tightly enough + * for our tests: `/v1/chat/completions`, `/v1/completions`, `/v1/embeddings`, + * `/v1/models`, `/v1/responses`, `/v1/rerank`. All routes echo the same + * canned response, so a single mock can serve any endpoint the test cares + * about. + */ +export async function startOpenAiUpstream( + opts: OpenAiUpstreamOptions = {}, +): Promise { + const received: ReceivedRequest[] = []; + + const server: Server = createServer((req, res) => { + let raw = ""; + req.on("data", (c: Buffer) => (raw += c.toString("utf8"))); + req.on("end", async () => { + received.push({ + method: req.method ?? "GET", + path: req.url ?? "/", + headers: Object.fromEntries( + Object.entries(req.headers).map(([k, v]) => [k, Array.isArray(v) ? v.join(",") : (v ?? "")]), + ), + body: raw, + }); + + if (opts.responseDelayMs) await sleep(opts.responseDelayMs); + + const status = opts.status ?? 200; + if (status >= 400) { + res.statusCode = status; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify(opts.errorBody ?? { error: { message: "mock error" } })); + return; + } + + const isStream = !!opts.streamEvents; + if (isStream) { + res.statusCode = 200; + res.setHeader("content-type", "text/event-stream"); + res.setHeader("cache-control", "no-cache"); + const events = opts.streamEvents ?? []; + for (let i = 0; i < events.length; i++) { + if ( + opts.disconnectAfterEvents !== undefined && + i >= opts.disconnectAfterEvents + ) { + res.destroy(); + return; + } + res.write(`data: ${events[i]}\n\n`); + if (opts.eventDelayMs) await sleep(opts.eventDelayMs); + } + res.end(); + return; + } + + res.statusCode = 200; + res.setHeader("content-type", "application/json"); + res.end( + JSON.stringify( + opts.nonStreamBody ?? { + id: "mock-1", + object: "chat.completion", + created: Math.floor(Date.now() / 1000), + model: "mock-model", + choices: [ + { + index: 0, + message: { role: "assistant", content: "mock reply" }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 5, completion_tokens: 3, total_tokens: 8 }, + }, + ), + ); + }); + }); + + const port = await pickFreePort(); + await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve)); + const baseUrl = `http://127.0.0.1:${port}`; + + return { + baseUrl, + receivedRequests: received, + async close() { + await new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + }, + }; +} + +function sleep(ms: number): Promise { + return new Promise((r) => setTimeout(r, ms)); +} diff --git a/tests/e2e/tsconfig.json b/tests/e2e/tsconfig.json new file mode 100644 index 00000000..73e047ee --- /dev/null +++ b/tests/e2e/tsconfig.json @@ -0,0 +1,14 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "Bundler", + "strict": true, + "esModuleInterop": true, + "resolveJsonModule": true, + "skipLibCheck": true, + "forceConsistentCasingInFileNames": true, + "types": ["node", "vitest/globals"] + }, + "include": ["src/**/*.ts"] +} diff --git a/tests/e2e/vitest.config.ts b/tests/e2e/vitest.config.ts new file mode 100644 index 00000000..9f8737d5 --- /dev/null +++ b/tests/e2e/vitest.config.ts @@ -0,0 +1,21 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + // E2E tests spin up the aisix binary; keep concurrency low so distinct + // test files don't contend for ports. Each file picks a random port + // internally, but lowering parallelism bounds the blast radius. + pool: "forks", + poolOptions: { + forks: { singleFork: false, minForks: 1, maxForks: 4 }, + }, + testTimeout: 60_000, + hookTimeout: 60_000, + globals: false, + coverage: { + provider: "v8", + reporter: ["text", "lcov"], + reportsDirectory: "coverage", + }, + }, +});