test(pty): migrate session tests to Effect runner (#27222)
This commit is contained in:
@@ -1,103 +1,96 @@
|
|||||||
import { describe, expect, test } from "bun:test"
|
import { describe, expect } from "bun:test"
|
||||||
import { AppRuntime } from "../../src/effect/app-runtime"
|
|
||||||
import { Bus } from "../../src/bus"
|
import { Bus } from "../../src/bus"
|
||||||
import { Effect } from "effect"
|
import { Config } from "../../src/config/config"
|
||||||
import { Instance } from "../../src/project/instance"
|
import { Plugin } from "../../src/plugin"
|
||||||
import { WithInstance } from "../../src/project/with-instance"
|
|
||||||
import { Pty } from "../../src/pty"
|
import { Pty } from "../../src/pty"
|
||||||
import type { PtyID } from "../../src/pty/schema"
|
import type { PtyID } from "../../src/pty/schema"
|
||||||
import { tmpdir } from "../fixture/fixture"
|
import { Effect, Layer, Queue } from "effect"
|
||||||
import { setTimeout as sleep } from "node:timers/promises"
|
import { testEffect } from "../lib/effect"
|
||||||
|
|
||||||
const wait = async (fn: () => boolean, ms = 5000) => {
|
type PtyEvent = { type: "created" | "exited" | "deleted"; id: PtyID }
|
||||||
const end = Date.now() + ms
|
|
||||||
while (Date.now() < end) {
|
|
||||||
if (fn()) return
|
|
||||||
await sleep(25)
|
|
||||||
}
|
|
||||||
throw new Error("timeout waiting for pty events")
|
|
||||||
}
|
|
||||||
|
|
||||||
const pick = (log: Array<{ type: "created" | "exited" | "deleted"; id: PtyID }>, id: PtyID) => {
|
const it = testEffect(
|
||||||
return log.filter((evt) => evt.id === id).map((evt) => evt.type)
|
Pty.layer.pipe(
|
||||||
|
Layer.provideMerge(Bus.layer),
|
||||||
|
Layer.provideMerge(Config.defaultLayer),
|
||||||
|
Layer.provideMerge(Plugin.defaultLayer),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
const ptyTest = process.platform === "win32" ? it.instance.skip : it.instance
|
||||||
|
|
||||||
|
const subscribePtyEvents = Effect.fn("PtySessionTest.subscribePtyEvents")(function* () {
|
||||||
|
const bus = yield* Bus.Service
|
||||||
|
const events = yield* Queue.unbounded<PtyEvent>()
|
||||||
|
|
||||||
|
const subscribe = <A>(effect: Effect.Effect<() => void, never, A>) =>
|
||||||
|
Effect.acquireRelease(effect, (off) => Effect.sync(off))
|
||||||
|
|
||||||
|
yield* subscribe(
|
||||||
|
bus.subscribeCallback(Pty.Event.Created, (evt) => {
|
||||||
|
Queue.offerUnsafe(events, { type: "created", id: evt.properties.info.id })
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
yield* subscribe(
|
||||||
|
bus.subscribeCallback(Pty.Event.Exited, (evt) => {
|
||||||
|
Queue.offerUnsafe(events, { type: "exited", id: evt.properties.id })
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
yield* subscribe(
|
||||||
|
bus.subscribeCallback(Pty.Event.Deleted, (evt) => {
|
||||||
|
Queue.offerUnsafe(events, { type: "deleted", id: evt.properties.id })
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
return events
|
||||||
|
})
|
||||||
|
|
||||||
|
const createPty = Effect.fn("PtySessionTest.createPty")(function* (input: Pty.CreateInput) {
|
||||||
|
const pty = yield* Pty.Service
|
||||||
|
return yield* Effect.acquireRelease(pty.create(input), (info) => pty.remove(info.id).pipe(Effect.ignore))
|
||||||
|
})
|
||||||
|
|
||||||
|
const waitForEvents = (events: Queue.Queue<PtyEvent>, id: PtyID, count: number) => {
|
||||||
|
return Effect.gen(function* () {
|
||||||
|
const picked: Array<PtyEvent["type"]> = []
|
||||||
|
while (picked.length < count) {
|
||||||
|
const evt = yield* Queue.take(events)
|
||||||
|
if (evt.id === id) picked.push(evt.type)
|
||||||
|
}
|
||||||
|
return picked
|
||||||
|
}).pipe(
|
||||||
|
Effect.timeoutOrElse({
|
||||||
|
duration: "5 seconds",
|
||||||
|
orElse: () => Effect.fail(new Error("timeout waiting for pty events")),
|
||||||
|
}),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
describe("pty", () => {
|
describe("pty", () => {
|
||||||
test("publishes created, exited, deleted in order for a short-lived process", async () => {
|
ptyTest("publishes created, exited, deleted in order for a short-lived process", () =>
|
||||||
if (process.platform === "win32") return
|
Effect.gen(function* () {
|
||||||
|
const events = yield* subscribePtyEvents()
|
||||||
|
const info = yield* createPty({
|
||||||
|
command: "/usr/bin/env",
|
||||||
|
args: ["sh", "-c", "sleep 0.1"],
|
||||||
|
title: "sleep",
|
||||||
|
})
|
||||||
|
|
||||||
await using dir = await tmpdir({ git: true })
|
expect(yield* waitForEvents(events, info.id, 3)).toEqual(["created", "exited", "deleted"])
|
||||||
|
}),
|
||||||
|
{ git: true },
|
||||||
|
)
|
||||||
|
|
||||||
await WithInstance.provide({
|
ptyTest("publishes created, exited, deleted in order for /bin/sh + remove", () =>
|
||||||
directory: dir.path,
|
Effect.gen(function* () {
|
||||||
fn: () =>
|
const pty = yield* Pty.Service
|
||||||
AppRuntime.runPromise(
|
const events = yield* subscribePtyEvents()
|
||||||
Effect.gen(function* () {
|
const info = yield* createPty({ command: "/bin/sh", title: "sh" })
|
||||||
const pty = yield* Pty.Service
|
|
||||||
const log: Array<{ type: "created" | "exited" | "deleted"; id: PtyID }> = []
|
|
||||||
const off = [
|
|
||||||
Bus.subscribe(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
|
|
||||||
Bus.subscribe(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
|
|
||||||
Bus.subscribe(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
|
|
||||||
]
|
|
||||||
|
|
||||||
let id: PtyID | undefined
|
expect(yield* waitForEvents(events, info.id, 1)).toEqual(["created"])
|
||||||
try {
|
yield* pty.write(info.id, "exit\n")
|
||||||
const info = yield* pty.create({
|
expect(yield* waitForEvents(events, info.id, 2)).toEqual(["exited", "deleted"])
|
||||||
command: "/usr/bin/env",
|
yield* pty.remove(info.id)
|
||||||
args: ["sh", "-c", "sleep 0.1"],
|
}),
|
||||||
title: "sleep",
|
{ git: true },
|
||||||
})
|
)
|
||||||
id = info.id
|
|
||||||
|
|
||||||
yield* Effect.promise(() => wait(() => pick(log, id!).includes("exited")))
|
|
||||||
|
|
||||||
yield* pty.remove(id)
|
|
||||||
yield* Effect.promise(() => wait(() => pick(log, id!).length >= 3))
|
|
||||||
expect(pick(log, id!)).toEqual(["created", "exited", "deleted"])
|
|
||||||
} finally {
|
|
||||||
off.forEach((x) => x())
|
|
||||||
if (id) yield* pty.remove(id)
|
|
||||||
}
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
})
|
|
||||||
})
|
|
||||||
|
|
||||||
test("publishes created, exited, deleted in order for /bin/sh + remove", async () => {
|
|
||||||
if (process.platform === "win32") return
|
|
||||||
|
|
||||||
await using dir = await tmpdir({ git: true })
|
|
||||||
|
|
||||||
await WithInstance.provide({
|
|
||||||
directory: dir.path,
|
|
||||||
fn: () =>
|
|
||||||
AppRuntime.runPromise(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const pty = yield* Pty.Service
|
|
||||||
const log: Array<{ type: "created" | "exited" | "deleted"; id: PtyID }> = []
|
|
||||||
const off = [
|
|
||||||
Bus.subscribe(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
|
|
||||||
Bus.subscribe(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
|
|
||||||
Bus.subscribe(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
|
|
||||||
]
|
|
||||||
|
|
||||||
let id: PtyID | undefined
|
|
||||||
try {
|
|
||||||
const info = yield* pty.create({ command: "/bin/sh", title: "sh" })
|
|
||||||
id = info.id
|
|
||||||
|
|
||||||
yield* Effect.promise(() => sleep(100))
|
|
||||||
|
|
||||||
yield* pty.remove(id)
|
|
||||||
yield* Effect.promise(() => wait(() => pick(log, id!).length >= 3))
|
|
||||||
expect(pick(log, id!)).toEqual(["created", "exited", "deleted"])
|
|
||||||
} finally {
|
|
||||||
off.forEach((x) => x())
|
|
||||||
if (id) yield* pty.remove(id)
|
|
||||||
}
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
})
|
|
||||||
})
|
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user