root
95 lines
3.9 KiB
TypeScript
95 lines
3.9 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { createServer } from "node:http";
|
|
import { test } from "bun:test";
|
|
|
|
import { handleWorkbenchKafkaSseDebugHttp } from "./workbench-kafka-sse-debug.ts";
|
|
|
|
test("workbench Kafka SSE debug endpoint streams filtered raw HWLAB Kafka events", async () => {
|
|
const fakeKafka = createFakeKafkaFactory();
|
|
const env = {
|
|
HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
|
|
HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
|
|
HWLAB_KAFKA_EVENT_TOPIC: "hwlab.event.v1"
|
|
};
|
|
const server = createServer((request, response) => {
|
|
const url = new URL(request.url ?? "/", "http://127.0.0.1");
|
|
void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, logger: null });
|
|
});
|
|
await listen(server);
|
|
const abort = new AbortController();
|
|
try {
|
|
const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab&sessionId=ses_kafka_sse_debug`, { signal: abort.signal });
|
|
assert.equal(response.status, 200);
|
|
assert.match(response.headers.get("content-type") ?? "", /text\/event-stream/u);
|
|
const reader = response.body?.getReader();
|
|
assert.ok(reader, "SSE response must expose a readable stream");
|
|
const connected = await readUntil(reader, "hwlab.kafka.connected");
|
|
assert.match(connected, /hwlab\.event\.v1/u);
|
|
|
|
await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_other", event: { label: "ignored" } });
|
|
await fakeKafka.emit({ eventType: "hwlab.trace.event.projected", sessionId: "ses_kafka_sse_debug", traceId: "trc_kafka_sse_debug", context: { runId: "run_kafka_sse_debug", commandId: "cmd_kafka_sse_debug" }, event: { label: "agentrun:event:debug", status: "running" } });
|
|
const streamed = await readUntil(reader, "trc_kafka_sse_debug");
|
|
assert.match(streamed, /hwlab\.kafka\.event/u);
|
|
assert.match(streamed, /ses_kafka_sse_debug/u);
|
|
assert.doesNotMatch(streamed, /ses_other/u);
|
|
} finally {
|
|
abort.abort();
|
|
await close(server);
|
|
}
|
|
});
|
|
|
|
function createFakeKafkaFactory() {
|
|
let eachMessage: ((input: any) => Promise<void>) | null = null;
|
|
const consumer = {
|
|
connect: async () => undefined,
|
|
subscribe: async () => undefined,
|
|
run: async (input: any) => { eachMessage = input.eachMessage; },
|
|
stop: async () => undefined,
|
|
disconnect: async () => undefined
|
|
};
|
|
return {
|
|
factory: () => ({ consumer: () => consumer }),
|
|
emit: async (value: Record<string, unknown>) => {
|
|
assert.ok(eachMessage, "consumer.run must be called before emitting fake Kafka events");
|
|
await eachMessage({
|
|
topic: "hwlab.event.v1",
|
|
partition: 0,
|
|
message: {
|
|
offset: String(value.sessionId === "ses_other" ? 1 : 2),
|
|
key: Buffer.from(String(value.sessionId ?? "unknown")),
|
|
timestamp: "2026-07-09T18:30:00.000Z",
|
|
value: Buffer.from(JSON.stringify(value))
|
|
}
|
|
});
|
|
}
|
|
};
|
|
}
|
|
|
|
async function readUntil(reader: ReadableStreamDefaultReader<Uint8Array>, pattern: string): Promise<string> {
|
|
const decoder = new TextDecoder();
|
|
let text = "";
|
|
const deadline = Date.now() + 3000;
|
|
while (!text.includes(pattern)) {
|
|
if (Date.now() > deadline) throw new Error(`SSE stream did not include ${pattern}: ${text}`);
|
|
const next = await reader.read();
|
|
if (next.done) break;
|
|
text += decoder.decode(next.value, { stream: true });
|
|
}
|
|
return text;
|
|
}
|
|
|
|
async function listen(server: ReturnType<typeof createServer>) {
|
|
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
|
}
|
|
|
|
async function close(server: ReturnType<typeof createServer>) {
|
|
await new Promise<void>((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
|
|
}
|
|
|
|
function serverUrl(server: ReturnType<typeof createServer>): string {
|
|
const address = server.address();
|
|
assert.equal(typeof address, "object");
|
|
assert.ok(address && typeof address.port === "number");
|
|
return `http://127.0.0.1:${address.port}`;
|
|
}
|