Add SSE Connection and streamingFetchAdapter tests (#625)

Author: EhabY

src/api/coderApi.ts

@@ -110,8 +110,9 @@ export class CoderApi extends Api {
 		options?: ClientOptions,
 	) => {
 		const searchParams = new URLSearchParams({ follow: "true" });
-		if (logs.length) {
-			searchParams.append("after", logs[logs.length - 1].id.toString());
+		const lastLog = logs.at(-1);
+		if (lastLog) {
+			searchParams.append("after", lastLog.id.toString());
 		}
 
 		return this.createWebSocket<ProvisionerJobLog>({
@@ -311,9 +312,9 @@ function setupInterceptors(
 			output,
 		);
 		// Add headers from the header command.
-		Object.entries(headers).forEach(([key, value]) => {
+		for (const [key, value] of Object.entries(headers)) {
 			config.headers[key] = value;
-		});
+		}
 
 		// Configure proxy and TLS.
 		// Note that by default VS Code overrides the agent. To prevent this, set
@@ -425,7 +426,7 @@ function wrapResponseTransform(
 function getSize(headers: AxiosHeaders, data: unknown): number | undefined {
 	const contentLength = headers["content-length"];
 	if (contentLength !== undefined) {
-		return parseInt(contentLength, 10);
+		return Number.parseInt(contentLength, 10);
 	}
 
 	return sizeOf(data);

src/api/streamingFetchAdapter.ts

@@ -1,6 +1,6 @@
 import { type AxiosInstance } from "axios";
 import { type FetchLikeInit, type FetchLikeResponse } from "eventsource";
-import { type IncomingMessage } from "http";
+import { type IncomingMessage } from "node:http";
 
 /**
  * Creates a fetch adapter using an Axios instance that returns streaming responses.

src/websocket/sseConnection.ts

@@ -109,11 +109,10 @@ export class SseConnection implements UnidirectionalStream<ServerSentEvent> {
 	}
 
 	private createErrorEvent(event: Event | ErrorEvent): WsErrorEvent {
-		const errorMessage =
-			event instanceof ErrorEvent && event.message
-				? event.message
-				: "SSE connection error";
-		const error = event instanceof ErrorEvent ? event.error : undefined;
+		// Check for properties instead of instanceof to avoid browser-only ErrorEvent global
+		const eventWithMessage = event as { message?: string; error?: unknown };
+		const errorMessage = eventWithMessage.message || "SSE connection error";
+		const error = eventWithMessage.error;
 
 		return {
 			error: error,

test/unit/api/coderApi.test.ts

@@ -125,7 +125,7 @@ describe("CoderApi", () => {
 			expect(thrownError.x509Err).toBeDefined();
 		});
 
-		it("applies headers in correct precedence order (command > config > axios default)", async () => {
+		it("applies headers in correct precedence order (command overrides config overrides axios default)", async () => {
 			const api = createApi(CODER_URL, AXIOS_TOKEN);
 
 			// Test 1: Headers from config, default token from API creation
@@ -225,7 +225,7 @@ describe("CoderApi", () => {
 			});
 		});
 
-		it("applies headers in correct precedence order (command > config > axios default)", async () => {
+		it("applies headers in correct precedence order (command overrides config overrides axios default)", async () => {
 			// Test 1: Default token from API creation
 			await api.watchBuildLogsByBuildId(BUILD_ID, []);
 

test/unit/api/streamingFetchAdapter.test.ts

@@ -0,0 +1,220 @@
+import { type AxiosInstance, type AxiosResponse } from "axios";
+import { type ReaderLike } from "eventsource";
+import { EventEmitter } from "node:events";
+import { type IncomingMessage } from "node:http";
+import { describe, it, expect, vi } from "vitest";
+
+import { createStreamingFetchAdapter } from "@/api/streamingFetchAdapter";
+
+const TEST_URL = "https://example.com/api";
+
+describe("createStreamingFetchAdapter", () => {
+	describe("Request Handling", () => {
+		it("passes URL, signal, and responseType to axios", async () => {
+			const mockAxios = createAxiosMock();
+			const mockStream = createMockStream();
+			setupAxiosResponse(mockAxios, 200, {}, mockStream);
+
+			const adapter = createStreamingFetchAdapter(mockAxios);
+			const signal = new AbortController().signal;
+
+			await adapter(TEST_URL, { signal });
+
+			expect(mockAxios.request).toHaveBeenCalledWith({
+				url: TEST_URL,
+				signal, // correctly passes signal
+				headers: {},
+				responseType: "stream",
+				validateStatus: expect.any(Function),
+			});
+		});
+
+		it("applies headers in correct precedence order (config overrides init)", async () => {
+			const mockAxios = createAxiosMock();
+			const mockStream = createMockStream();
+			setupAxiosResponse(mockAxios, 200, {}, mockStream);
+
+			// Test 1: No config headers, only init headers
+			const adapter1 = createStreamingFetchAdapter(mockAxios);
+			await adapter1(TEST_URL, {
+				headers: { "X-Init": "init-value" },
+			});
+
+			expect(mockAxios.request).toHaveBeenCalledWith(
+				expect.objectContaining({
+					headers: { "X-Init": "init-value" },
+				}),
+			);
+
+			// Test 2: Config headers merge with init headers
+			const adapter2 = createStreamingFetchAdapter(mockAxios, {
+				"X-Config": "config-value",
+			});
+			await adapter2(TEST_URL, {
+				headers: { "X-Init": "init-value" },
+			});
+
+			expect(mockAxios.request).toHaveBeenCalledWith(
+				expect.objectContaining({
+					headers: {
+						"X-Init": "init-value",
+						"X-Config": "config-value",
+					},
+				}),
+			);
+
+			// Test 3: Config headers override init headers
+			const adapter3 = createStreamingFetchAdapter(mockAxios, {
+				"X-Header": "config-value",
+			});
+			await adapter3(TEST_URL, {
+				headers: { "X-Header": "init-value" },
+			});
+
+			expect(mockAxios.request).toHaveBeenCalledWith(
+				expect.objectContaining({
+					headers: { "X-Header": "config-value" },
+				}),
+			);
+		});
+	});
+
+	describe("Response Properties", () => {
+		it("returns response with correct properties", async () => {
+			const mockAxios = createAxiosMock();
+			const mockStream = createMockStream();
+			setupAxiosResponse(
+				mockAxios,
+				200,
+				{ "content-type": "text/event-stream" },
+				mockStream,
+			);
+
+			const adapter = createStreamingFetchAdapter(mockAxios);
+			const response = await adapter(TEST_URL);
+
+			expect(response.url).toBe(TEST_URL);
+			expect(response.status).toBe(200);
+			expect(response.headers.get("content-type")).toBe("text/event-stream");
+			// Headers are lowercased when we retrieve them
+			expect(response.headers.get("CoNtEnT-TyPe")).toBe("text/event-stream");
+			expect(response.body?.getReader).toBeDefined();
+		});
+
+		it("detects redirected requests", async () => {
+			const mockAxios = createAxiosMock();
+			const mockStream = createMockStream();
+			const mockResponse = {
+				status: 200,
+				headers: {},
+				data: mockStream,
+				request: {
+					res: {
+						responseUrl: "https://redirect.com/api",
+					},
+				},
+			} as AxiosResponse<IncomingMessage>;
+			vi.mocked(mockAxios.request).mockResolvedValue(mockResponse);
+
+			const adapter = createStreamingFetchAdapter(mockAxios);
+			const response = await adapter(TEST_URL);
+
+			expect(response.redirected).toBe(true);
+		});
+	});
+
+	describe("Stream Handling", () => {
+		it("enqueues data chunks from stream", async () => {
+			const { mockStream, reader } = await setupReaderTest();
+
+			const chunk1 = Buffer.from("data1");
+			const chunk2 = Buffer.from("data2");
+			mockStream.emit("data", chunk1);
+			mockStream.emit("data", chunk2);
+			mockStream.emit("end");
+
+			const result1 = await reader.read();
+			expect(result1.value).toEqual(chunk1);
+			expect(result1.done).toBe(false);
+
+			const result2 = await reader.read();
+			expect(result2.value).toEqual(chunk2);
+			expect(result2.done).toBe(false);
+
+			const result3 = await reader.read();
+			// Closed after end
+			expect(result3.done).toBe(true);
+		});
+
+		it("propagates stream errors", async () => {
+			const { mockStream, reader } = await setupReaderTest();
+
+			const error = new Error("Stream error");
+			mockStream.emit("error", error);
+
+			await expect(reader.read()).rejects.toThrow("Stream error");
+		});
+
+		it("handles errors after stream is closed", async () => {
+			const { mockStream, reader } = await setupReaderTest();
+
+			mockStream.emit("end");
+			await reader.read();
+
+			// Emit events after stream is closed - should not throw
+			expect(() => mockStream.emit("data", Buffer.from("late"))).not.toThrow();
+			expect(() => mockStream.emit("end")).not.toThrow();
+		});
+
+		it("destroys stream on cancel", async () => {
+			const { mockStream, reader } = await setupReaderTest();
+
+			await reader.cancel();
+
+			expect(mockStream.destroy).toHaveBeenCalled();
+		});
+	});
+});
+
+function createAxiosMock(): AxiosInstance {
+	return {
+		request: vi.fn(),
+	} as unknown as AxiosInstance;
+}
+
+function createMockStream(): IncomingMessage {
+	const stream = new EventEmitter() as IncomingMessage;
+	stream.destroy = vi.fn();
+	return stream;
+}
+
+function setupAxiosResponse(
+	axios: AxiosInstance,
+	status: number,
+	headers: Record<string, string>,
+	stream: IncomingMessage,
+): void {
+	vi.mocked(axios.request).mockResolvedValue({
+		status,
+		headers,
+		data: stream,
+	});
+}
+
+async function setupReaderTest(): Promise<{
+	mockStream: IncomingMessage;
+	reader: ReaderLike | ReadableStreamDefaultReader<Uint8Array<ArrayBuffer>>;
+}> {
+	const mockAxios = createAxiosMock();
+	const mockStream = createMockStream();
+	setupAxiosResponse(mockAxios, 200, {}, mockStream);
+
+	const adapter = createStreamingFetchAdapter(mockAxios);
+	const response = await adapter(TEST_URL);
+	const reader = response.body?.getReader();
+	if (reader === undefined) {
+		throw new Error("Reader is undefined");
+	}
+
+	return { mockStream, reader };
+}

test/unit/core/cliManager.test.ts

@@ -546,7 +546,8 @@ describe("CliManager", () => {
 			expect(files.find((file) => file.includes(".asc"))).toBeUndefined();
 		});
 
-		it.each([
+		type SignatureErrorTestCase = [status: number, message: string];
+		it.each<SignatureErrorTestCase>([
 			[404, "Signature not found"],
 			[500, "Failed to download signature"],
 		])("allows skipping verification on %i", async (status, message) => {
@@ -558,7 +559,7 @@ describe("CliManager", () => {
 			expect(pgp.verifySignature).not.toHaveBeenCalled();
 		});
 
-		it.each([
+		it.each<SignatureErrorTestCase>([
 			[404, "Signature not found"],
 			[500, "Failed to download signature"],
 		])(

test/unit/logging/utils.test.ts

@@ -23,7 +23,8 @@ describe("Logging utils", () => {
 	});
 
 	describe("sizeOf", () => {
-		it.each([
+		type SizeOfTestCase = [data: unknown, bytes: number | undefined];
+		it.each<SizeOfTestCase>([
 			// Primitives return a fixed value
 			[null, 0],
 			[undefined, 0],