Skip to content

Commit 76e0e8a

Browse files
gorillamoegorillamoe
committed
fix(websockets): tests
1 parent ed198d8 commit 76e0e8a

1 file changed

Lines changed: 115 additions & 28 deletions

File tree

‎packages/core/src/lib/websocket/websocket-session.test.ts‎

Lines changed: 115 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -10,27 +10,72 @@ type OutboundMessage = {
1010
code?: number;
1111
};
1212

13-
async function readAllStdout(
14-
stream: ReadableStream<Uint8Array>,
15-
): Promise<OutboundMessage[]> {
16-
const events: OutboundMessage[] = [];
17-
const reader = stream.getReader();
18-
const decoder = new TextDecoder();
19-
let buf = "";
20-
21-
while (true) {
22-
const { done, value } = await reader.read();
23-
if (done) break;
24-
buf += decoder.decode(value, { stream: true });
25-
let nl: number;
26-
while ((nl = buf.indexOf("\n")) >= 0) {
27-
const line = buf.slice(0, nl).trim();
28-
buf = buf.slice(nl + 1);
29-
if (line) events.push(JSON.parse(line) as OutboundMessage);
13+
class StdoutCollector {
14+
readonly events: OutboundMessage[] = [];
15+
private buf = "";
16+
private readonly waiters: Array<() => void> = [];
17+
private done = false;
18+
private readonly pump: Promise<void>;
19+
20+
constructor(stream: ReadableStream<Uint8Array>) {
21+
this.pump = this.read(stream);
22+
}
23+
24+
private notify(): void {
25+
const waiters = this.waiters.splice(0);
26+
for (const wake of waiters) wake();
27+
}
28+
29+
private async read(stream: ReadableStream<Uint8Array>): Promise<void> {
30+
const reader = stream.getReader();
31+
const decoder = new TextDecoder();
32+
try {
33+
while (true) {
34+
const { done, value } = await reader.read();
35+
if (done) break;
36+
this.buf += decoder.decode(value, { stream: true });
37+
let nl: number;
38+
while ((nl = this.buf.indexOf("\n")) >= 0) {
39+
const line = this.buf.slice(0, nl).trim();
40+
this.buf = this.buf.slice(nl + 1);
41+
if (line) this.events.push(JSON.parse(line) as OutboundMessage);
42+
}
43+
this.notify();
44+
}
45+
} finally {
46+
this.done = true;
47+
this.notify();
3048
}
3149
}
3250

33-
return events;
51+
async waitUntil(
52+
predicate: (events: OutboundMessage[]) => boolean,
53+
label: string,
54+
timeoutMs = 10_000,
55+
): Promise<void> {
56+
const deadline = Date.now() + timeoutMs;
57+
while (!predicate(this.events)) {
58+
if (this.done) {
59+
throw new Error(
60+
`stdout ended before ${label}; events=${JSON.stringify(this.events)}`,
61+
);
62+
}
63+
if (Date.now() >= deadline) {
64+
throw new Error(
65+
`timed out waiting for ${label}; events=${JSON.stringify(this.events)}`,
66+
);
67+
}
68+
await new Promise<void>((resolve) => {
69+
this.waiters.push(resolve);
70+
setTimeout(resolve, 50);
71+
});
72+
}
73+
}
74+
75+
async finish(): Promise<OutboundMessage[]> {
76+
await this.pump;
77+
return this.events;
78+
}
3479
}
3580

3681
describe("runWebSocketSession", () => {
@@ -43,6 +88,7 @@ describe("runWebSocketSession", () => {
4388

4489
test("emits sent for initial body and stdin send", async () => {
4590
server = Bun.serve({
91+
hostname: "127.0.0.1",
4692
port: 0,
4793
fetch(req, srv) {
4894
if (srv.upgrade(req, { data: undefined })) return;
@@ -68,25 +114,66 @@ describe("runWebSocketSession", () => {
68114

69115
const cliPath = join(import.meta.dir, "../../cli.ts");
70116
const child = Bun.spawn(
71-
["bun", cliPath, "--websocket", "-i", connectFile],
117+
["bun", "run", cliPath, "--websocket", "-i", connectFile],
72118
{
73-
cwd: join(import.meta.dir, "../../../.."),
119+
cwd: join(import.meta.dir, "../../../../../"),
74120
stdin: "pipe",
75121
stdout: "pipe",
76122
stderr: "pipe",
77123
},
78124
);
79125

80-
const eventsPromise = readAllStdout(child.stdout);
126+
const stderrChunks: string[] = [];
127+
void child.stderr.pipeTo(
128+
new WritableStream({
129+
write(chunk) {
130+
stderrChunks.push(new TextDecoder().decode(chunk));
131+
},
132+
}),
133+
);
134+
135+
const collector = new StdoutCollector(child.stdout);
136+
137+
try {
138+
await collector.waitUntil(
139+
(events) =>
140+
events.some((e) => e.type === "ready") &&
141+
events.filter((e) => e.type === "sent").length >= 1 &&
142+
events.filter((e) => e.type === "message").length >= 1,
143+
"initial sent+echo",
144+
);
81145

82-
await Bun.sleep(200);
83-
child.stdin.write(`${JSON.stringify({ op: "send", data: '{"foo":1}' })}\n`);
84-
await Bun.sleep(200);
85-
child.stdin.write(`${JSON.stringify({ op: "close" })}\n`);
86-
child.stdin.end();
146+
child.stdin.write(
147+
`${JSON.stringify({ op: "send", data: '{"foo":1}' })}\n`,
148+
);
149+
150+
await collector.waitUntil(
151+
(events) =>
152+
events.filter((e) => e.type === "sent").length >= 2 &&
153+
events.filter((e) => e.type === "message").length >= 2,
154+
"stdin sent+echo",
155+
);
156+
157+
child.stdin.write(`${JSON.stringify({ op: "close" })}\n`);
158+
child.stdin.end();
159+
160+
await collector.waitUntil(
161+
(events) => events.some((e) => e.type === "closed"),
162+
"closed",
163+
);
164+
} catch (error) {
165+
child.kill();
166+
const stderr = stderrChunks.join("").trim();
167+
throw new Error(
168+
`${error instanceof Error ? error.message : String(error)}${
169+
stderr ? `\nstderr: ${stderr}` : ""
170+
}`,
171+
);
172+
}
87173

88-
const events = await eventsPromise;
89-
expect(await child.exited).toBe(0);
174+
const events = await collector.finish();
175+
const exitCode = await child.exited;
176+
expect(exitCode).toBe(0);
90177

91178
expect(events.map((e) => e.type)).toEqual([
92179
"ready",

0 commit comments

Comments
 (0)