General tests for SSE client transport
Since we're here anyway…
This commit is contained in:
Justin Spahr-Summers
parent
e2723d30b8
commit
742b8b43a0
1 changed files with 236 additions and 55 deletions
|
|
@ -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<void>((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<void>((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/);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Reference in a new issue