Open History to see the last commit
1import * as NodeEvents from "node:events";
2import * as NodeNet from "node:net";
3
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { assert, it } from "@effect/vitest";
6import * as HostProcess from "@t3tools/shared/HostProcess";
7import * as Effect from "effect/Effect";
8import * as Fiber from "effect/Fiber";
9import * as Layer from "effect/Layer";
10import * as Scheduler from "effect/Scheduler";
11import { vi } from "vite-plus/test";
12
13import * as NodePtyAdapter from "./NodePtyAdapter.ts";
14import * as PtyAdapter from "@t3tools/shared/PtyAdapter";
15
16function makeNativeProcess(pid = 42) {
17 const events = new NodeEvents.EventEmitter();
18 return {
19 pid,
20 _socket: new NodeNet.Socket(),
21 _agent: { kill: vi.fn() },
22 write: vi.fn(),
23 resize: vi.fn(),
24 kill: vi.fn(),
25 onData: vi.fn((callback: (data: string) => void) => {
26 events.on("data", callback);
27 return {
28 dispose: () => {
29 events.off("data", callback);
30 },
31 };
32 }),
33 onExit: vi.fn((callback: (event: { exitCode: number; signal?: number }) => void) => {
34 events.on("exit", callback);
35 return {
36 dispose: () => {
37 events.off("exit", callback);
38 },
39 };
40 }),
41 events,
42 };
43}
44
45const spawn = vi.fn(() => makeNativeProcess());
46
47function preparePendingProcess() {
48 const nativeProcess = makeNativeProcess(0);
49 const subscribed = Promise.withResolvers<void>();
50 nativeProcess._socket.on("newListener", (event) => {
51 if (event === "ready_datapipe") queueMicrotask(() => subscribed.resolve());
52 });
53 spawn.mockReturnValueOnce(nativeProcess);
54 return { nativeProcess, subscribed: Effect.promise(() => subscribed.promise) };
55}
56
57const spawnInput = { shell: "powershell.exe", cwd: ".", cols: 80, rows: 24, env: {} };
58
59const fakeNodePty = { spawn } as unknown as typeof import("node-pty");
60
61const layerTestFor = (platform: NodeJS.Platform = "win32") =>
62 NodePtyAdapter.layer.pipe(
63 Layer.provide(
64 Layer.mergeAll(
65 NodeServices.layer,
66 Layer.succeed(HostProcess.Platform, platform),
67 Layer.succeed(HostProcess.Architecture, "x64"),
68 Layer.succeed(NodePtyAdapter.NodePtyModuleLoaderRef, () => Promise.resolve(fakeNodePty)),
69 ),
70 ),
71 );
72
73const layerTest = layerTestFor();
74
75it.effect("waits for the Windows PID without requiring output", () =>
76 Effect.gen(function* () {
77 const { nativeProcess, subscribed } = preparePendingProcess();
78 const adapter = yield* PtyAdapter.PtyAdapter;
79 let completed = false;
80 const fiber = yield* adapter.spawn(spawnInput).pipe(
81 Effect.tap(() =>
82 Effect.sync(() => {
83 completed = true;
84 }),
85 ),
86 Effect.forkChild,
87 );
88 yield* subscribed;
89 assert.isFalse(completed);
90 nativeProcess.pid = 12345;
91 nativeProcess._socket.emit("ready_datapipe");
92 const process = yield* Fiber.join(fiber);
93 assert.equal(process.pid, 12345);
94 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
95 assert.equal(nativeProcess.events.listenerCount("exit"), 1);
96
97 const output: string[] = [];
98 const exits: PtyAdapter.PtyExitEvent[] = [];
99 const stopData = process.onData((data) => output.push(data));
100 const stopExit = process.onExit((event) => exits.push(event));
101 nativeProcess.events.emit("data", "first output");
102 nativeProcess.events.emit("exit", { exitCode: 0 });
103 assert.deepEqual(output, ["first output"]);
104 assert.deepEqual(exits, [{ exitCode: 0, signal: null }]);
105 stopData();
106 stopExit();
107 }).pipe(Effect.provide(layerTest)),
108);
109
110it.effect.each(["exit", "close", "error", "invalid-pid"] as const)(
111 "fails Windows startup on %s and cleans up",
112 (failure) =>
113 Effect.gen(function* () {
114 const { nativeProcess, subscribed } = preparePendingProcess();
115 const adapter = yield* PtyAdapter.PtyAdapter;
116 const fiber = yield* adapter.spawn(spawnInput).pipe(Effect.result, Effect.forkChild);
117 yield* subscribed;
118 if (failure === "exit") nativeProcess.events.emit("exit", { exitCode: 1 });
119 else if (failure === "error") nativeProcess._socket.emit("error", new Error("pipe failed"));
120 else nativeProcess._socket.emit(failure === "close" ? "close" : "ready_datapipe");
121 const result = yield* Fiber.join(fiber);
122 assert.equal(result._tag, "Failure");
123 if (result._tag === "Failure") assert.instanceOf(result.failure, PtyAdapter.PtySpawnError);
124 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
125 assert.equal(nativeProcess._socket.listenerCount("error"), 0);
126 assert.equal(nativeProcess._socket.listenerCount("close"), 0);
127 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
128 assert.equal(nativeProcess._agent.kill.mock.calls.length, 1);
129 }).pipe(Effect.provide(layerTest)),
130);
131
132it.effect("cancels the Windows connection without waiting for output", () =>
133 Effect.gen(function* () {
134 const { nativeProcess, subscribed } = preparePendingProcess();
135 const adapter = yield* PtyAdapter.PtyAdapter;
136 const fiber = yield* adapter.spawn(spawnInput).pipe(Effect.forkChild);
137 yield* subscribed;
138 yield* Fiber.interrupt(fiber);
139 assert.equal(nativeProcess._agent.kill.mock.calls.length, 1);
140 assert.equal(nativeProcess.kill.mock.calls.length, 0);
141 assert.equal(nativeProcess._socket.listenerCount("ready_datapipe"), 0);
142 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
143 }).pipe(Effect.provide(layerTest)),
144);
145
146it.effect("reports an incompatible Windows readiness API instead of hanging", () =>
147 Effect.gen(function* () {
148 const nativeProcess = makeNativeProcess(0);
149 Reflect.deleteProperty(nativeProcess, "_socket");
150 spawn.mockReturnValueOnce(nativeProcess);
151 const adapter = yield* PtyAdapter.PtyAdapter;
152 const error = yield* adapter.spawn(spawnInput).pipe(Effect.flip);
153 assert.instanceOf(error, PtyAdapter.PtySpawnError);
154 assert.instanceOf(error.cause, Error);
155 assert.equal(error.cause.message, "Windows PTY readiness socket is unavailable.");
156 assert.equal(nativeProcess._agent.kill.mock.calls.length, 1);
157 }).pipe(Effect.provide(layerTest)),
158);
159
160it.effect.each([2048, 8])(
161 "preserves an exit during readiness handoff with scheduler budget %s",
162 (budget) =>
163 Effect.gen(function* () {
164 const { nativeProcess, subscribed } = preparePendingProcess();
165 const adapter = yield* PtyAdapter.PtyAdapter;
166 const exits: PtyAdapter.PtyExitEvent[] = [];
167 const fiber = yield* Effect.gen(function* () {
168 const process = yield* adapter.spawn(spawnInput);
169 process.onExit((event) => exits.push(event));
170 }).pipe(
171 Effect.provideService(Scheduler.MaxOpsBeforeYield, budget),
172 Effect.provideService(Scheduler.PreventSchedulerYield, false),
173 Effect.forkChild,
174 );
175 yield* subscribed;
176 nativeProcess.pid = 12345;
177 nativeProcess._socket.emit("ready_datapipe");
178 nativeProcess.events.emit("exit", { exitCode: 0 });
179 yield* Fiber.join(fiber);
180 assert.equal(exits.length, 1);
181 }).pipe(Effect.provide(layerTest)),
182);
183
184it.effect("replays an exit to late subscribers and respects unsubscription", () =>
185 Effect.gen(function* () {
186 const adapter = yield* PtyAdapter.PtyAdapter;
187 const process = yield* adapter.spawn(spawnInput);
188 const nativeProcess = spawn.mock.results.at(-1)!.value;
189 const removed = vi.fn();
190 process.onExit(removed)();
191 nativeProcess.events.emit("exit", { exitCode: 7, signal: 2 });
192 const late = vi.fn();
193 process.onExit(late);
194 nativeProcess.events.emit("exit", { exitCode: 9 });
195 assert.equal(removed.mock.calls.length, 0);
196 assert.deepEqual(late.mock.calls, [[{ exitCode: 7, signal: 2 }]]);
197 assert.equal(nativeProcess.events.listenerCount("exit"), 0);
198 }).pipe(Effect.provide(layerTest)),
199);