diff --git a/src/client/sse.test.ts b/src/client/sse.test.ts index 02dfac43..c1777ada 100644 --- a/src/client/sse.test.ts +++ b/src/client/sse.test.ts @@ -1,5 +1,5 @@ import { SSEClientTransport } from "./sse.js"; -import { createServer, type Server } from "http"; +import { createServer, type Server, type IncomingMessage } from "http"; import { JSONRPCMessage } from "../types.js"; import { AddressInfo } from "net"; @@ -7,23 +7,45 @@ describe("SSEClientTransport", () => { let server: Server; let transport: SSEClientTransport; let baseUrl: URL; + let lastServerRequest: IncomingMessage; + let sendServerMessage: ((message: string) => void) | null = null; beforeEach((done) => { + // Reset state + lastServerRequest = null as unknown as IncomingMessage; + sendServerMessage = null; + // Create a test server that will receive the EventSource connection server = createServer((req, res) => { - // Store the received headers for verification - (server as any).lastRequest = req; + lastServerRequest = req; // Send SSE headers res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", - "Connection": "keep-alive" + Connection: "keep-alive", }); // Send the endpoint event res.write("event: endpoint\n"); res.write(`data: ${baseUrl.href}\n\n`); + + // Store reference to send function for tests + sendServerMessage = (message: string) => { + res.write(`data: ${message}\n\n`); + }; + + // Handle request body for POST endpoints + if (req.method === "POST") { + let body = ""; + req.on("data", (chunk) => { + body += chunk; + }); + req.on("end", () => { + (req as IncomingMessage & { body: string }).body = body; + res.end(); + }); + } }); // Start server on random port @@ -40,68 +62,227 @@ describe("SSEClientTransport", () => { }); }); - it("uses custom fetch implementation from EventSourceInit to add auth headers", async () => { - const authToken = "Bearer test-token"; + describe("connection handling", () => { + it("establishes SSE connection and receives endpoint", async () => { + transport = new SSEClientTransport(baseUrl); + await transport.start(); - // Create a fetch wrapper that adds auth header - const fetchWithAuth = (url: string | URL, init?: RequestInit) => { - const headers = new Headers(init?.headers); - headers.set("Authorization", authToken); - return fetch(url.toString(), { ...init, headers }); - }; - - transport = new SSEClientTransport(baseUrl, { - eventSourceInit: { - fetch: fetchWithAuth - } + expect(lastServerRequest.headers.accept).toBe("text/event-stream"); + expect(lastServerRequest.method).toBe("GET"); }); - await transport.start(); + it("rejects if server returns non-200 status", async () => { + // Create a server that returns 403 + server.close(); + await new Promise((resolve) => server.on("close", resolve)); - // Verify the auth header was received by the server - const headers = (server as any).lastRequest.headers; - expect(headers.authorization).toBe(authToken); + server = createServer((req, res) => { + res.writeHead(403); + res.end(); + }); + + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + const addr = server.address() as AddressInfo; + baseUrl = new URL(`http://127.0.0.1:${addr.port}`); + resolve(); + }); + }); + + transport = new SSEClientTransport(baseUrl); + await expect(transport.start()).rejects.toThrow(); + }); + + it("closes EventSource connection on close()", async () => { + transport = new SSEClientTransport(baseUrl); + await transport.start(); + + const closePromise = new Promise((resolve) => { + lastServerRequest.on("close", resolve); + }); + + await transport.close(); + await closePromise; + }); }); - it("passes custom headers to fetch requests", async () => { - const customHeaders = { - Authorization: "Bearer test-token", - "X-Custom-Header": "custom-value" - }; + describe("message handling", () => { + it("receives and parses JSON-RPC messages", async () => { + const receivedMessages: JSONRPCMessage[] = []; + transport = new SSEClientTransport(baseUrl); + transport.onmessage = (msg) => receivedMessages.push(msg); - transport = new SSEClientTransport(baseUrl, { - requestInit: { - headers: customHeaders - } + await transport.start(); + + const testMessage: JSONRPCMessage = { + jsonrpc: "2.0", + id: "test-1", + method: "test", + params: { foo: "bar" }, + }; + + sendServerMessage!(JSON.stringify(testMessage)); + + // Wait for message processing + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(receivedMessages).toHaveLength(1); + expect(receivedMessages[0]).toEqual(testMessage); }); - await transport.start(); + it("handles malformed JSON messages", async () => { + const errors: Error[] = []; + transport = new SSEClientTransport(baseUrl); + transport.onerror = (err) => errors.push(err); - // Mock fetch for the message sending test - global.fetch = jest.fn().mockResolvedValue({ - ok: true + await transport.start(); + + sendServerMessage!("invalid json"); + + // Wait for message processing + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(errors).toHaveLength(1); + expect(errors[0].message).toMatch(/JSON/); }); - const message: JSONRPCMessage = { - jsonrpc: "2.0", - id: "1", - method: "test", - params: {} - }; + it("handles messages via POST requests", async () => { + transport = new SSEClientTransport(baseUrl); + await transport.start(); - await transport.send(message); + const testMessage: JSONRPCMessage = { + jsonrpc: "2.0", + id: "test-1", + method: "test", + params: { foo: "bar" }, + }; - // Verify fetch was called with correct headers - expect(global.fetch).toHaveBeenCalledWith( - expect.any(URL), - expect.objectContaining({ - headers: expect.any(Headers) - }) - ); + await transport.send(testMessage); - const calledHeaders = (global.fetch as jest.Mock).mock.calls[0][1].headers; - expect(calledHeaders.get("Authorization")).toBe(customHeaders.Authorization); - expect(calledHeaders.get("X-Custom-Header")).toBe(customHeaders["X-Custom-Header"]); - expect(calledHeaders.get("content-type")).toBe("application/json"); + // Wait for request processing + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(lastServerRequest.method).toBe("POST"); + expect(lastServerRequest.headers["content-type"]).toBe( + "application/json", + ); + expect( + JSON.parse( + (lastServerRequest as IncomingMessage & { body: string }).body, + ), + ).toEqual(testMessage); + }); + + it("handles POST request failures", async () => { + // Create a server that returns 500 for POST + server.close(); + await new Promise((resolve) => server.on("close", resolve)); + + server = createServer((req, res) => { + if (req.method === "GET") { + res.writeHead(200, { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + }); + res.write("event: endpoint\n"); + res.write(`data: ${baseUrl.href}\n\n`); + } else { + res.writeHead(500); + res.end("Internal error"); + } + }); + + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + const addr = server.address() as AddressInfo; + baseUrl = new URL(`http://127.0.0.1:${addr.port}`); + resolve(); + }); + }); + + transport = new SSEClientTransport(baseUrl); + await transport.start(); + + const testMessage: JSONRPCMessage = { + jsonrpc: "2.0", + id: "test-1", + method: "test", + params: {}, + }; + + await expect(transport.send(testMessage)).rejects.toThrow(/500/); + }); }); -}); \ No newline at end of file + + describe("header handling", () => { + it("uses custom fetch implementation from EventSourceInit to add auth headers", async () => { + const authToken = "Bearer test-token"; + + // Create a fetch wrapper that adds auth header + const fetchWithAuth = (url: string | URL, init?: RequestInit) => { + const headers = new Headers(init?.headers); + headers.set("Authorization", authToken); + return fetch(url.toString(), { ...init, headers }); + }; + + transport = new SSEClientTransport(baseUrl, { + eventSourceInit: { + fetch: fetchWithAuth, + }, + }); + + await transport.start(); + + // Verify the auth header was received by the server + expect(lastServerRequest.headers.authorization).toBe(authToken); + }); + + it("passes custom headers to fetch requests", async () => { + const customHeaders = { + Authorization: "Bearer test-token", + "X-Custom-Header": "custom-value", + }; + + transport = new SSEClientTransport(baseUrl, { + requestInit: { + headers: customHeaders, + }, + }); + + await transport.start(); + + // Mock fetch for the message sending test + global.fetch = jest.fn().mockResolvedValue({ + ok: true, + }); + + const message: JSONRPCMessage = { + jsonrpc: "2.0", + id: "1", + method: "test", + params: {}, + }; + + await transport.send(message); + + // Verify fetch was called with correct headers + expect(global.fetch).toHaveBeenCalledWith( + expect.any(URL), + expect.objectContaining({ + headers: expect.any(Headers), + }), + ); + + const calledHeaders = (global.fetch as jest.Mock).mock.calls[0][1] + .headers; + expect(calledHeaders.get("Authorization")).toBe( + customHeaders.Authorization, + ); + expect(calledHeaders.get("X-Custom-Header")).toBe( + customHeaders["X-Custom-Header"], + ); + expect(calledHeaders.get("content-type")).toBe("application/json"); + }); + }); +});