test: migrate file watcher test to Effect (#27346)
This commit is contained in:
@@ -1,15 +1,13 @@
|
|||||||
import { $ } from "bun"
|
import { describe, expect } from "bun:test"
|
||||||
import { afterEach, describe, expect, test } from "bun:test"
|
|
||||||
import fs from "fs/promises"
|
|
||||||
import path from "path"
|
import path from "path"
|
||||||
import { ConfigProvider, Deferred, Effect, Layer, ManagedRuntime, Option } from "effect"
|
import { AppFileSystem } from "@opencode-ai/core/filesystem"
|
||||||
import { disposeAllInstances, tmpdir } from "../fixture/fixture"
|
import { ConfigProvider, Deferred, Effect, Layer, Option } from "effect"
|
||||||
import { Bus } from "../../src/bus"
|
import { TestInstance, provideInstance } from "../fixture/fixture"
|
||||||
|
import { testEffect } from "../lib/effect"
|
||||||
|
import { GlobalBus, type GlobalEvent } from "../../src/bus/global"
|
||||||
import { Config } from "@/config/config"
|
import { Config } from "@/config/config"
|
||||||
import { FileWatcher } from "../../src/file/watcher"
|
import { FileWatcher } from "../../src/file/watcher"
|
||||||
import { Git } from "../../src/git"
|
import { Git } from "../../src/git"
|
||||||
import { Instance } from "../../src/project/instance"
|
|
||||||
import { WithInstance } from "../../src/project/with-instance"
|
|
||||||
|
|
||||||
// Native @parcel/watcher bindings aren't reliably available in CI (missing on Linux, flaky on Windows)
|
// Native @parcel/watcher bindings aren't reliably available in CI (missing on Linux, flaky on Windows)
|
||||||
const describeWatcher = FileWatcher.hasNativeBinding() && !process.env.CI ? describe : describe.skip
|
const describeWatcher = FileWatcher.hasNativeBinding() && !process.env.CI ? describe : describe.skip
|
||||||
@@ -25,43 +23,43 @@ const watcherConfigLayer = ConfigProvider.layer(
|
|||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
const watcherLayer = FileWatcher.layer.pipe(
|
||||||
|
Layer.provide(Config.defaultLayer),
|
||||||
|
Layer.provide(Git.defaultLayer),
|
||||||
|
Layer.provide(watcherConfigLayer),
|
||||||
|
)
|
||||||
|
|
||||||
|
const it = testEffect(Layer.mergeAll(AppFileSystem.defaultLayer, Git.defaultLayer))
|
||||||
|
|
||||||
type WatcherEvent = { file: string; event: "add" | "change" | "unlink" }
|
type WatcherEvent = { file: string; event: "add" | "change" | "unlink" }
|
||||||
|
|
||||||
/** Run `body` with a live FileWatcher service. */
|
/** Run `body` with a live FileWatcher service. */
|
||||||
function withWatcher<E>(directory: string, body: Effect.Effect<void, E>) {
|
function withWatcher<A, E, R>(directory: string, body: Effect.Effect<A, E, R>) {
|
||||||
return WithInstance.provide({
|
return Effect.gen(function* () {
|
||||||
directory,
|
const watcher = yield* FileWatcher.Service
|
||||||
fn: async () => {
|
yield* watcher.init()
|
||||||
const layer: Layer.Layer<FileWatcher.Service, never, never> = FileWatcher.layer.pipe(
|
yield* ready(directory)
|
||||||
Layer.provide(Config.defaultLayer),
|
return yield* body
|
||||||
Layer.provide(Git.defaultLayer),
|
}).pipe(Effect.provide(watcherLayer), provideInstance(directory), Effect.scoped)
|
||||||
Layer.provide(watcherConfigLayer),
|
|
||||||
)
|
|
||||||
const rt = ManagedRuntime.make(layer)
|
|
||||||
try {
|
|
||||||
await rt.runPromise(FileWatcher.Service.use((s) => s.init()))
|
|
||||||
await Effect.runPromise(ready(directory))
|
|
||||||
await Effect.runPromise(body)
|
|
||||||
} finally {
|
|
||||||
await rt.dispose()
|
|
||||||
}
|
|
||||||
},
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (evt: WatcherEvent) => void) {
|
function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (evt: WatcherEvent) => void) {
|
||||||
let done = false
|
let done = false
|
||||||
|
|
||||||
const unsub = Bus.subscribe(FileWatcher.Event.Updated, (evt) => {
|
const on = (evt: GlobalEvent) => {
|
||||||
if (done) return
|
if (done) return
|
||||||
if (!check(evt.properties)) return
|
if (evt.directory !== directory) return
|
||||||
hit(evt.properties)
|
if (evt.payload.type !== FileWatcher.Event.Updated.type) return
|
||||||
})
|
if (!check(evt.payload.properties)) return
|
||||||
|
hit(evt.payload.properties)
|
||||||
|
}
|
||||||
|
|
||||||
|
GlobalBus.on("event", on)
|
||||||
|
|
||||||
return () => {
|
return () => {
|
||||||
if (done) return
|
if (done) return
|
||||||
done = true
|
done = true
|
||||||
unsub()
|
GlobalBus.off("event", on)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -72,7 +70,7 @@ function wait(directory: string, check: (evt: WatcherEvent) => boolean) {
|
|||||||
let off = () => {}
|
let off = () => {}
|
||||||
off = listen(directory, check, (evt) => {
|
off = listen(directory, check, (evt) => {
|
||||||
off()
|
off()
|
||||||
Deferred.doneUnsafe(deferred, Effect.succeed(evt))
|
Effect.runFork(Deferred.succeed(deferred, evt))
|
||||||
})
|
})
|
||||||
return off
|
return off
|
||||||
})
|
})
|
||||||
@@ -86,7 +84,12 @@ function nextUpdate<E>(directory: string, check: (evt: WatcherEvent) => boolean,
|
|||||||
({ deferred }) =>
|
({ deferred }) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
yield* trigger
|
yield* trigger
|
||||||
return yield* Deferred.await(deferred).pipe(Effect.timeout("5 seconds"))
|
return yield* Deferred.await(deferred).pipe(
|
||||||
|
Effect.timeoutOrElse({
|
||||||
|
duration: "5 seconds",
|
||||||
|
orElse: () => Effect.fail(new Error("timed out waiting for file watcher update")),
|
||||||
|
}),
|
||||||
|
)
|
||||||
}),
|
}),
|
||||||
({ cleanup }) => Effect.sync(cleanup),
|
({ cleanup }) => Effect.sync(cleanup),
|
||||||
)
|
)
|
||||||
@@ -104,7 +107,11 @@ function noUpdate<E>(
|
|||||||
({ deferred }) =>
|
({ deferred }) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
yield* trigger
|
yield* trigger
|
||||||
expect(yield* Deferred.await(deferred).pipe(Effect.timeoutOption(`${ms} millis`))).toEqual(Option.none())
|
const result = yield* Deferred.await(deferred).pipe(
|
||||||
|
Effect.map((evt) => Option.some(evt)),
|
||||||
|
Effect.timeoutOrElse({ duration: `${ms} millis`, orElse: () => Effect.succeed(Option.none()) }),
|
||||||
|
)
|
||||||
|
expect(result).toEqual(Option.none())
|
||||||
}),
|
}),
|
||||||
({ cleanup }) => Effect.sync(cleanup),
|
({ cleanup }) => Effect.sync(cleanup),
|
||||||
)
|
)
|
||||||
@@ -115,29 +122,25 @@ function ready(directory: string) {
|
|||||||
const head = path.join(directory, ".git", "HEAD")
|
const head = path.join(directory, ".git", "HEAD")
|
||||||
|
|
||||||
return Effect.gen(function* () {
|
return Effect.gen(function* () {
|
||||||
|
const fs = yield* AppFileSystem.Service
|
||||||
|
const git = yield* Git.Service
|
||||||
|
|
||||||
yield* nextUpdate(
|
yield* nextUpdate(
|
||||||
directory,
|
directory,
|
||||||
(evt) => evt.file === file && evt.event === "add",
|
(evt) => evt.file === file && evt.event === "add",
|
||||||
Effect.promise(() => fs.writeFile(file, "ready")),
|
fs.writeFileString(file, "ready"),
|
||||||
).pipe(Effect.ensuring(Effect.promise(() => fs.rm(file, { force: true }).catch(() => undefined))), Effect.asVoid)
|
).pipe(Effect.ensuring(fs.remove(file, { force: true }).pipe(Effect.ignore)), Effect.asVoid)
|
||||||
|
|
||||||
const git = yield* Effect.promise(() =>
|
if (!(yield* fs.existsSafe(head))) return
|
||||||
fs
|
|
||||||
.stat(head)
|
|
||||||
.then(() => true)
|
|
||||||
.catch(() => false),
|
|
||||||
)
|
|
||||||
if (!git) return
|
|
||||||
|
|
||||||
const branch = `watch-${Math.random().toString(36).slice(2)}`
|
const branch = `watch-${Math.random().toString(36).slice(2)}`
|
||||||
const hash = yield* Effect.promise(() => $`git rev-parse HEAD`.cwd(directory).quiet().text())
|
const hash = (yield* git.run(["rev-parse", "HEAD"], { cwd: directory })).text()
|
||||||
yield* nextUpdate(
|
yield* nextUpdate(
|
||||||
directory,
|
directory,
|
||||||
(evt) => evt.file === head && evt.event !== "unlink",
|
(evt) => evt.file === head && evt.event !== "unlink",
|
||||||
Effect.promise(async () => {
|
fs
|
||||||
await fs.writeFile(path.join(directory, ".git", "refs", "heads", branch), hash.trim() + "\n")
|
.writeFileString(path.join(directory, ".git", "refs", "heads", branch), hash.trim() + "\n")
|
||||||
await fs.writeFile(head, `ref: refs/heads/${branch}\n`)
|
.pipe(Effect.andThen(fs.writeFileString(head, `ref: refs/heads/${branch}\n`))),
|
||||||
}),
|
|
||||||
).pipe(Effect.asVoid)
|
).pipe(Effect.asVoid)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -147,104 +150,108 @@ function ready(directory: string) {
|
|||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
describeWatcher("FileWatcher", () => {
|
describeWatcher("FileWatcher", () => {
|
||||||
afterEach(async () => {
|
it.instance("publishes root create, update, and delete events", () =>
|
||||||
await disposeAllInstances()
|
Effect.gen(function* () {
|
||||||
})
|
const test = yield* TestInstance
|
||||||
|
const fs = yield* AppFileSystem.Service
|
||||||
|
const file = path.join(test.directory, "watch.txt")
|
||||||
|
const cases = [
|
||||||
|
{ event: "add" as const, trigger: fs.writeFileString(file, "a") },
|
||||||
|
{ event: "change" as const, trigger: fs.writeFileString(file, "b") },
|
||||||
|
{ event: "unlink" as const, trigger: fs.remove(file) },
|
||||||
|
]
|
||||||
|
|
||||||
test("publishes root create, update, and delete events", async () => {
|
yield* withWatcher(
|
||||||
await using tmp = await tmpdir({ git: true })
|
test.directory,
|
||||||
const file = path.join(tmp.path, "watch.txt")
|
Effect.forEach(cases, ({ event, trigger }) =>
|
||||||
const dir = tmp.path
|
nextUpdate(test.directory, (evt) => evt.file === file && evt.event === event, trigger).pipe(
|
||||||
const cases = [
|
Effect.tap((evt) => Effect.sync(() => expect(evt).toEqual({ file, event }))),
|
||||||
{ event: "add" as const, trigger: Effect.promise(() => fs.writeFile(file, "a")) },
|
|
||||||
{ event: "change" as const, trigger: Effect.promise(() => fs.writeFile(file, "b")) },
|
|
||||||
{ event: "unlink" as const, trigger: Effect.promise(() => fs.unlink(file)) },
|
|
||||||
]
|
|
||||||
|
|
||||||
await withWatcher(
|
|
||||||
dir,
|
|
||||||
Effect.forEach(cases, ({ event, trigger }) =>
|
|
||||||
nextUpdate(dir, (evt) => evt.file === file && evt.event === event, trigger).pipe(
|
|
||||||
Effect.tap((evt) => Effect.sync(() => expect(evt).toEqual({ file, event }))),
|
|
||||||
),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("watches non-git roots", async () => {
|
|
||||||
await using tmp = await tmpdir()
|
|
||||||
const file = path.join(tmp.path, "plain.txt")
|
|
||||||
const dir = tmp.path
|
|
||||||
|
|
||||||
await withWatcher(
|
|
||||||
dir,
|
|
||||||
nextUpdate(
|
|
||||||
dir,
|
|
||||||
(e) => e.file === file && e.event === "add",
|
|
||||||
Effect.promise(() => fs.writeFile(file, "plain")),
|
|
||||||
).pipe(Effect.tap((evt) => Effect.sync(() => expect(evt).toEqual({ file, event: "add" })))),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("cleanup stops publishing events", async () => {
|
|
||||||
await using tmp = await tmpdir({ git: true })
|
|
||||||
const file = path.join(tmp.path, "after-dispose.txt")
|
|
||||||
|
|
||||||
// Start and immediately stop the watcher (withWatcher disposes on exit)
|
|
||||||
await withWatcher(tmp.path, Effect.void)
|
|
||||||
|
|
||||||
// Now write a file — no watcher should be listening
|
|
||||||
await WithInstance.provide({
|
|
||||||
directory: tmp.path,
|
|
||||||
fn: () =>
|
|
||||||
Effect.runPromise(
|
|
||||||
noUpdate(
|
|
||||||
tmp.path,
|
|
||||||
(e) => e.file === file,
|
|
||||||
Effect.promise(() => fs.writeFile(file, "gone")),
|
|
||||||
),
|
),
|
||||||
),
|
),
|
||||||
})
|
)
|
||||||
})
|
}),
|
||||||
|
{ git: true },
|
||||||
|
)
|
||||||
|
|
||||||
test("ignores .git/index changes", async () => {
|
it.instance("watches non-git roots", () =>
|
||||||
await using tmp = await tmpdir({ git: true })
|
Effect.gen(function* () {
|
||||||
const gitIndex = path.join(tmp.path, ".git", "index")
|
const test = yield* TestInstance
|
||||||
const edit = path.join(tmp.path, "tracked.txt")
|
const fs = yield* AppFileSystem.Service
|
||||||
|
const file = path.join(test.directory, "plain.txt")
|
||||||
|
|
||||||
await withWatcher(
|
yield* withWatcher(
|
||||||
tmp.path,
|
test.directory,
|
||||||
noUpdate(
|
nextUpdate(
|
||||||
tmp.path,
|
test.directory,
|
||||||
(e) => e.file === gitIndex,
|
(e) => e.file === file && e.event === "add",
|
||||||
Effect.promise(async () => {
|
fs.writeFileString(file, "plain"),
|
||||||
await fs.writeFile(edit, "a")
|
).pipe(Effect.tap((evt) => Effect.sync(() => expect(evt).toEqual({ file, event: "add" })))),
|
||||||
await $`git add .`.cwd(tmp.path).quiet().nothrow()
|
)
|
||||||
}),
|
}),
|
||||||
),
|
)
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("publishes .git/HEAD events", async () => {
|
it.instance("cleanup stops publishing events", () =>
|
||||||
await using tmp = await tmpdir({ git: true })
|
Effect.gen(function* () {
|
||||||
const head = path.join(tmp.path, ".git", "HEAD")
|
const test = yield* TestInstance
|
||||||
const branch = `watch-${Math.random().toString(36).slice(2)}`
|
const fs = yield* AppFileSystem.Service
|
||||||
await $`git branch ${branch}`.cwd(tmp.path).quiet()
|
const file = path.join(test.directory, "after-dispose.txt")
|
||||||
|
|
||||||
await withWatcher(
|
// Start and immediately stop the watcher (withWatcher disposes on exit).
|
||||||
tmp.path,
|
yield* withWatcher(test.directory, Effect.void)
|
||||||
nextUpdate(
|
|
||||||
tmp.path,
|
// Now write a file - no watcher should be listening.
|
||||||
(evt) => evt.file === head && evt.event !== "unlink",
|
yield* noUpdate(test.directory, (e) => e.file === file, fs.writeFileString(file, "gone")).pipe(
|
||||||
Effect.promise(() => fs.writeFile(head, `ref: refs/heads/${branch}\n`)),
|
provideInstance(test.directory),
|
||||||
).pipe(
|
)
|
||||||
Effect.tap((evt) =>
|
}),
|
||||||
Effect.sync(() => {
|
{ git: true },
|
||||||
expect(evt.file).toBe(head)
|
)
|
||||||
expect(["add", "change"]).toContain(evt.event)
|
|
||||||
}),
|
it.instance("ignores .git/index changes", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const test = yield* TestInstance
|
||||||
|
const fs = yield* AppFileSystem.Service
|
||||||
|
const git = yield* Git.Service
|
||||||
|
const gitIndex = path.join(test.directory, ".git", "index")
|
||||||
|
const edit = path.join(test.directory, "tracked.txt")
|
||||||
|
|
||||||
|
yield* withWatcher(
|
||||||
|
test.directory,
|
||||||
|
noUpdate(
|
||||||
|
test.directory,
|
||||||
|
(e) => e.file === gitIndex,
|
||||||
|
fs.writeFileString(edit, "a").pipe(Effect.andThen(git.run(["add", "."], { cwd: test.directory }))),
|
||||||
),
|
),
|
||||||
),
|
)
|
||||||
)
|
}),
|
||||||
})
|
{ git: true },
|
||||||
|
)
|
||||||
|
|
||||||
|
it.instance("publishes .git/HEAD events", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const test = yield* TestInstance
|
||||||
|
const fs = yield* AppFileSystem.Service
|
||||||
|
const git = yield* Git.Service
|
||||||
|
const head = path.join(test.directory, ".git", "HEAD")
|
||||||
|
const branch = `watch-${Math.random().toString(36).slice(2)}`
|
||||||
|
yield* git.run(["branch", branch], { cwd: test.directory })
|
||||||
|
|
||||||
|
yield* withWatcher(
|
||||||
|
test.directory,
|
||||||
|
nextUpdate(
|
||||||
|
test.directory,
|
||||||
|
(evt) => evt.file === head && evt.event !== "unlink",
|
||||||
|
fs.writeFileString(head, `ref: refs/heads/${branch}\n`),
|
||||||
|
).pipe(
|
||||||
|
Effect.tap((evt) =>
|
||||||
|
Effect.sync(() => {
|
||||||
|
expect(evt.file).toBe(head)
|
||||||
|
expect(["add", "change"]).toContain(evt.event)
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}),
|
||||||
|
{ git: true },
|
||||||
|
)
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user