From 11556aaf4cb6cc7ee3aead83458f3c055e4d6d6d Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 2 Nov 2026 06:12:33 +0100 Subject: [PATCH] test(plugins): add the plain subscription unit test and the browser spec --- .../plugin-subscription-examples.spec.ts | 79 ++++++ .../app/src/plugins/subscriptions.test.ts | 265 +++++------------- 2 files changed, 154 insertions(+), 190 deletions(-) create mode 100644 packages/app/e2e/browser/plugin-subscription-examples.spec.ts diff --git a/packages/app/e2e/browser/plugin-subscription-examples.spec.ts b/packages/app/e2e/browser/plugin-subscription-examples.spec.ts new file mode 100644 index 0000000000..8cb25576bc --- /dev/null +++ b/packages/app/e2e/browser/plugin-subscription-examples.spec.ts @@ -0,0 +1,79 @@ +import { writeFile } from "node:fs/promises"; +import path from "node:path"; +import { expect, test, type Page } from "../support/fixtures"; +import { gotoAppShell } from "../support/helpers/app"; +import { openCommandCenter } from "../support/helpers/command-center"; +import { connectNewWorkspaceDaemonClient } from "../support/helpers/new-workspace"; +import { copyPluginExample } from "../support/helpers/plugin-fixture"; +import { seedWorkspace, type SeededWorkspace } from "../support/helpers/seed-client"; +import { getServerId } from "../support/helpers/server-id"; +import { + switchWorkspaceViaSidebar, + waitForWorkspaceInSidebar, +} from "../support/helpers/workspace-ui"; + +async function runCommand(page: Page, title: string): Promise { + const panel = await openCommandCenter(page); + await panel.getByTestId("command-center-input").fill(title); + await panel.getByRole("button", { name: title, exact: true }).click(); + await expect(panel).not.toBeVisible(); +} + +/** Installs a plugin example, opens a seeded workspace, and runs `exercise` against them. */ +async function withExample( + page: Page, + example: { name: string; id: string }, + exercise: (workspace: SeededWorkspace) => Promise, +): Promise { + const client = await connectNewWorkspaceDaemonClient({ ownProjects: false }); + const previous = await client.getDaemonConfig(); + const copy = await copyPluginExample(example.name); + const workspace = await seedWorkspace({ repoPrefix: `${example.name}-` }); + try { + await client.patchDaemonConfig({ pluginsEnabled: true }); + await client.installDirectoryPlugin(copy.directory); + await page.setViewportSize({ width: 1280, height: 800 }); + await gotoAppShell(page); + const target = { serverId: getServerId(), workspaceId: workspace.workspaceId }; + await waitForWorkspaceInSidebar(page, target); + await switchWorkspaceViaSidebar({ page, ...target }); + await exercise(workspace); + } finally { + try { + await client.removePlugin(example.id); + } finally { + await client.patchDaemonConfig({ pluginsEnabled: previous.config.pluginsEnabled }); + await workspace.cleanup(); + await client.close(); + await copy.cleanup(); + } + } +} + +test.describe("plugin subscriptions", () => { + test.describe.configure({ timeout: 180_000 }); + + test("a panel follows events the daemon pushes as files change", async ({ page }) => { + await withExample(page, { name: "diff-watch", id: "diff-watch" }, async (workspace) => { + await runCommand(page, "Open live changes"); + await expect(page.getByText("No changes", { exact: true })).toBeVisible(); + + await writeFile(path.join(workspace.repoPath, "pushed.txt"), "from the test\n"); + await expect(page.getByText("1 file changed", { exact: true })).toBeVisible(); + await expect(page.getByText("pushed.txt", { exact: true })).toBeVisible(); + }); + }); + + test("a panel plays streamed bytes and sends bytes back", async ({ page }) => { + await withExample(page, { name: "media-stream", id: "media-stream" }, async () => { + await runCommand(page, "Open media stream"); + await expect(page.getByText("live", { exact: true })).toBeVisible(); + await expect(page.getByText(/^[1-9]\d* video frames/)).toBeVisible(); + + await page.getByRole("button", { name: "Send this frame back", exact: true }).click(); + await expect( + page.getByText(/^The daemon received [1-9]\d* bytes, sha256 [0-9a-f]{12}$/), + ).toBeVisible(); + }); + }); +}); diff --git a/packages/app/src/plugins/subscriptions.test.ts b/packages/app/src/plugins/subscriptions.test.ts index 971e841fbf..927a54c178 100644 --- a/packages/app/src/plugins/subscriptions.test.ts +++ b/packages/app/src/plugins/subscriptions.test.ts @@ -1,21 +1,12 @@ -/** - * @vitest-environment jsdom - */ -import { act, renderHook, waitFor } from "@testing-library/react"; -import { defineRpc, defineSubscription } from "@getpaseo/plugin"; -import { useRpc, useSubscription } from "@getpaseo/plugin/client"; -import { PluginTranslationRuntime, subscribePluginRpc } from "@getpaseo/plugin/client/host"; -import type { PaseoApi } from "@getpaseo/client"; +import { defineSubscription } from "@getpaseo/plugin"; +import type { PluginSubscriptionClosed } from "@getpaseo/plugin/client"; +import { subscribePluginRpc } from "@getpaseo/plugin/client/host"; import type { DaemonClient } from "@getpaseo/client/internal/daemon-client"; import type { SubscriptionObserver } from "@getpaseo/client"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; -import { QueryClient } from "@tanstack/react-query"; -import React, { type ReactNode } from "react"; import { expect, it } from "vitest"; import { z } from "zod"; -import { PluginInstallationProvider } from "./installation-provider"; import { createPluginSubscribe } from "./subscriptions"; -import type { InstalledPlugin } from "./types"; const changes = defineSubscription({ name: "diff.watch", @@ -95,7 +86,8 @@ class HostSubscription { } } -function host() { +/** An installation's subscribe on a host the test drives, and what one subscriber observed. */ +function installation() { const subscriptions: HostSubscription[] = []; const client = { observePluginRpc: (...[pluginId, method, input, options]: Observed) => { @@ -105,200 +97,93 @@ function host() { return subscription.handle() as unknown as ReturnType; }, }; - return { client, subscriptions }; -} - -function installation(input: { - client: ReturnType["client"]; - invoke?: InstalledPlugin["invoke"]; -}): InstalledPlugin { const lifetime = new AbortController(); - return { - id: "diffs", - serverId: "host", - clientBundle: "", - lifetime, - locale: "en", - i18n: new PluginTranslationRuntime({ getLocale: () => "en", subscribe: () => () => {} }), - paseo: { dispose: async () => undefined } as PaseoApi, - invoke: input.invoke ?? (async () => undefined), - subscribe: createPluginSubscribe({ - client: input.client, - pluginId: "diffs", - lifetime: lifetime.signal, - }), - queryClient: new QueryClient(), - cleanup: () => undefined, - settingsScreens: [], - surfaces: [], - sidebarItems: { header: [], footer: [] }, - legacySidebarItems: [], - filePanels: [], - workspacePanels: [], - commandCenterItems: [], - clientSlashCommands: [], - attachmentSources: [], - themes: [], - timelineTransformers: [], - timelineRenderers: [], - }; + const subscribe = createPluginSubscribe({ client, pluginId: "diffs", lifetime: lifetime.signal }); + function open(ref: string, signal?: AbortSignal) { + const received: string[] = []; + const closed: PluginSubscriptionClosed[] = []; + const handle = subscribePluginRpc( + changes, + subscribe, + { ref }, + (event) => + received.push(event instanceof Uint8Array ? `bytes:${[...event]}` : `path:${event.path}`), + { signal, onClosed: (result) => closed.push(result) }, + ); + return { handle, received, closed, host: subscriptions[subscriptions.length - 1]! }; + } + return { open, lifetime }; } -function wrapper(plugin: InstalledPlugin) { - return function Wrapper({ children }: { children: ReactNode }) { - return {children}; - }; -} +it("delivers JSON and byte events in order and releases the host subscription", async () => { + const { open } = installation(); + const { handle, received, closed, host } = open("main"); + expect(host.method).toBe("diff.watch"); + expect(host.input).toEqual({ ref: "main" }); + host.acknowledge(); + await handle.ready; -it("useSubscription goes live, delivers JSON and bytes to the latest callback, and releases on unmount", async () => { - const { client, subscriptions } = host(); - const plugin = installation({ client }); - const received: string[] = []; - const { result, rerender, unmount } = renderHook( - ({ label }: { label: string }) => - useSubscription(changes, { ref: "main" }, (event) => { - received.push( - event instanceof Uint8Array ? `${label}:bytes:${[...event]}` : `${label}:${event.path}`, - ); - }), - { wrapper: wrapper(plugin), initialProps: { label: "first" } }, - ); + host.event({ path: "a.ts" }); + host.bytes(new Uint8Array([1, 2])); + expect(received).toEqual(["path:a.ts", "bytes:1,2"]); - expect(result.current).toEqual({ status: "connecting" }); - const [subscription] = subscriptions; - expect(subscription.method).toBe("diff.watch"); - expect(subscription.input).toEqual({ ref: "main" }); - await act(async () => subscription.acknowledge()); - expect(result.current).toEqual({ status: "live" }); - - act(() => subscription.event({ path: "a.ts" })); - rerender({ label: "second" }); - act(() => subscription.bytes(new Uint8Array([1, 2]))); - expect(received).toEqual(["first:a.ts", "second:bytes:1,2"]); - // A new callback is not a new subscription. - expect(subscriptions).toHaveLength(1); - - unmount(); - await subscription.released; + handle(); + await host.released; + expect(closed).toEqual([]); }); -it("useSubscription subscribes again when its input changes shallowly", async () => { - const { client, subscriptions } = host(); - const plugin = installation({ client }); - const { rerender } = renderHook( - ({ ref }: { ref: string }) => useSubscription(changes, { ref }, () => undefined), - { wrapper: wrapper(plugin), initialProps: { ref: "main" } }, - ); - rerender({ ref: "main" }); - expect(subscriptions).toHaveLength(1); - rerender({ ref: "next" }); - expect(subscriptions).toHaveLength(2); - await subscriptions[0].released; - expect(subscriptions[1].input).toEqual({ ref: "next" }); - expect(subscriptions[1].isReleased).toBe(false); -}); +it("reports a refusal, a failure after going live, and a clean close", async () => { + const { open } = installation(); + const refused = open("a"); + refused.host.refuseWith(new Error("no such ref")); + await expect(refused.handle.ready).rejects.toThrow("no such ref"); + expect(refused.closed).toEqual([{ error: new Error("no such ref") }]); -it("useSubscription reports a refusal, a failure after going live, and a clean close", async () => { - const { client, subscriptions } = host(); - const plugin = installation({ client }); - const refused = renderHook(() => useSubscription(changes, { ref: "a" }, () => undefined), { - wrapper: wrapper(plugin), - }); - await act(async () => subscriptions[0].refuseWith(new Error("no such ref"))); - expect(refused.result.current).toMatchObject({ - status: "error", - error: { message: "no such ref" }, - }); - - const failed = renderHook(() => useSubscription(changes, { ref: "b" }, () => undefined), { - wrapper: wrapper(plugin), - }); - await act(async () => subscriptions[1].acknowledge()); - act(() => subscriptions[1].closed("Plugin stopped: diffs")); - expect(failed.result.current).toMatchObject({ - status: "error", - error: { message: "Plugin stopped: diffs" }, - }); - await subscriptions[1].released; - - const finished = renderHook(() => useSubscription(changes, { ref: "c" }, () => undefined), { - wrapper: wrapper(plugin), - }); - await act(async () => subscriptions[2].acknowledge()); - act(() => subscriptions[2].closed()); - expect(finished.result.current).toEqual({ status: "closed" }); + const failed = open("b"); + failed.host.acknowledge(); + await failed.handle.ready; + failed.host.closed("Plugin stopped: diffs"); + expect(failed.closed).toEqual([{ error: new Error("Plugin stopped: diffs") }]); + await failed.host.released; // A handler that returns at once closes in the same tick the host accepts it. - const immediate = renderHook(() => useSubscription(changes, { ref: "d" }, () => undefined), { - wrapper: wrapper(plugin), - }); - await act(async () => { - subscriptions[3].acknowledge(); - subscriptions[3].closed(); - }); - expect(immediate.result.current).toEqual({ status: "closed" }); + const finished = open("c"); + finished.host.acknowledge(); + finished.host.closed(); + await expect(finished.handle.ready).resolves.toBeUndefined(); + expect(finished.closed).toEqual([{}]); }); -it("an event that fails the contract closes the subscription with an error", async () => { - const { client, subscriptions } = host(); - const plugin = installation({ client }); - const received: unknown[] = []; - const { result } = renderHook( - () => useSubscription(changes, { ref: "a" }, (event) => received.push(event)), - { wrapper: wrapper(plugin) }, - ); - await act(async () => subscriptions[0].acknowledge()); - act(() => subscriptions[0].event({ path: 3 })); +it("closes a subscription whose event fails the contract and drops what follows", async () => { + const { open } = installation(); + const { handle, received, closed, host } = open("a"); + host.acknowledge(); + await handle.ready; + host.event({ path: 3 }); + host.event({ path: "late.ts" }); expect(received).toEqual([]); - expect(result.current).toMatchObject({ - status: "error", - error: { message: "Plugin subscription diff.watch sent an invalid event" }, - }); - await subscriptions[0].released; + expect(closed).toEqual([ + { error: new Error("Plugin subscription diff.watch sent an invalid event") }, + ]); + await host.released; }); -it("client.subscribe releases on its signal, on its remover, and when the installation stops", async () => { - const { client, subscriptions } = host(); - const plugin = installation({ client }); - const subscribe = (onEvent: (event: Event) => void, signal?: AbortSignal) => - subscribePluginRpc(changes, plugin.subscribe, { ref: "main" }, onEvent as never, { signal }); - +it("releases on its signal, on its remover, and when the installation stops", async () => { + const { open, lifetime } = installation(); const controller = new AbortController(); - const bySignal = subscribe(() => undefined, controller.signal); - subscriptions[0].acknowledge(); - await bySignal.ready; + const bySignal = open("a", controller.signal); + bySignal.host.acknowledge(); + await bySignal.handle.ready; controller.abort(); - await subscriptions[0].released; + await bySignal.host.released; - const byRemover = subscribe(() => undefined); - byRemover(); - await subscriptions[1].released; + const byRemover = open("b"); + byRemover.handle(); + await byRemover.host.released; - const received: unknown[] = []; - subscribe((event) => received.push(event)); - subscriptions[2].event({ path: "a.ts" }); - expect(received).toEqual([{ path: "a.ts" }]); - plugin.lifetime.abort(); - await subscriptions[2].released; -}); - -it("useRpc resolves a binary output as bytes", async () => { - const thumbnail = defineRpc({ - name: "thumbnail.render", - input: z.object({ path: z.string() }), - output: z.instanceof(Uint8Array), - }); - const plugin = installation({ - client: host().client, - invoke: async (method, input) => { - expect(method).toBe("thumbnail.render"); - expect(input).toEqual({ path: "a.png" }); - return new Uint8Array([137, 80, 78, 71]); - }, - }); - const { result } = renderHook(() => useRpc(thumbnail), { wrapper: wrapper(plugin) }); - await waitFor(async () => { - const bytes = await result.current({ path: "a.png" }); - expect([...bytes]).toEqual([137, 80, 78, 71]); - }); + const byLifetime = open("c"); + byLifetime.host.event({ path: "a.ts" }); + expect(byLifetime.received).toEqual(["path:a.ts"]); + lifetime.abort(); + await byLifetime.host.released; });