File size: 4,120 Bytes
4440aec | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | // Shared fixture for tests driving createGatewayAuthenticatedRequestDispatcher.
import { vi } from "vitest";
import type { WebSocket } from "ws";
import { createDeferredCore, type Deferred } from "../../../shared/deferred.js";
import type { GatewayWsClient } from "../ws-types.js";
import { createGatewayAuthenticatedRequestDispatcher } from "./authenticated-request-dispatch.js";
import type { GatewayWsMessageHandlerParams } from "./message-handler-types.js";
export type GatewayTestResponseFrame = {
type: "res";
id: string;
ok: boolean;
payload?: unknown;
error?: { code?: string; message?: string } & Record<string, unknown>;
};
export function createOperatorWsClient(
overrides: {
connId?: string;
socket?: unknown;
clientInfo?: { id: string; mode: string };
scopes?: string[];
} = {},
): GatewayWsClient {
return {
socket: (overrides.socket ?? {}) as WebSocket,
connId: overrides.connId ?? "dispatch-test-connection",
usesSharedGatewayAuth: false,
connect: {
minProtocol: 1,
maxProtocol: 1,
client: {
id: overrides.clientInfo?.id ?? "gateway-client",
version: "dev",
platform: "test",
mode: overrides.clientInfo?.mode ?? "backend",
},
role: "operator",
scopes: overrides.scopes ?? ["operator.admin"],
},
} as GatewayWsClient;
}
export function createDispatchTestHarness(
options: {
connId?: string;
extraHandlers?: GatewayWsMessageHandlerParams["extraHandlers"];
buildRequestContext?: () => unknown;
isClosed?: () => boolean;
getRequiredSharedGatewaySessionGeneration?: () => string | undefined;
} = {},
) {
const sentResponses: GatewayTestResponseFrame[] = [];
const responseWaiters: { id: string; deferred: Deferred<GatewayTestResponseFrame> }[] = [];
const send = vi.fn<GatewayWsMessageHandlerParams["send"]>(() => ({ kind: "sent" }));
// Recording lives outside the spy so tests may replace send's implementation
// (to observe call context) without silently breaking awaitResponseFrame.
const sendForDispatcher = (frame: unknown) => {
const response = frame as GatewayTestResponseFrame;
if (response?.type === "res") {
sentResponses.push(response);
for (let index = responseWaiters.length - 1; index >= 0; index -= 1) {
const waiter = responseWaiters[index];
if (waiter && waiter.id === response.id) {
responseWaiters.splice(index, 1);
waiter.deferred.resolve(response);
}
}
}
return send(frame) ?? ({ kind: "sent" } as const);
};
const close = vi.fn();
const setCloseCause = vi.fn();
const logGateway = { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() };
const dispatcher = createGatewayAuthenticatedRequestDispatcher({
handler: {
connId: options.connId ?? "dispatch-test-connection",
extraHandlers: options.extraHandlers ?? {},
buildRequestContext: () => (options.buildRequestContext?.() ?? {}) as never,
send: sendForDispatcher,
close,
isClosed: options.isClosed ?? (() => false),
getRequiredSharedGatewaySessionGeneration: options.getRequiredSharedGatewaySessionGeneration,
setCloseCause,
logGateway,
} as unknown as GatewayWsMessageHandlerParams,
isWebchatConnect: () => false,
});
// A response can precede handler completion. Tests driving ongoing work wait
// for this event, then release their gates and join the original dispatch.
const awaitResponseFrame = (id: string): Promise<GatewayTestResponseFrame> => {
const already = sentResponses.find((frame) => frame.id === id);
if (already) {
return Promise.resolve(already);
}
const deferred = createDeferredCore<GatewayTestResponseFrame>();
responseWaiters.push({ id, deferred });
return deferred.promise;
};
return {
awaitResponseFrame,
close,
dispatcher: {
dispatch: (frame: unknown, client: GatewayWsClient) =>
dispatcher.dispatch(frame, client, Buffer.byteLength(JSON.stringify(frame))),
},
logGateway,
send,
setCloseCause,
};
}
|