From 05117a025849d64c6ef4aff8dc8edf33e56aba1e Mon Sep 17 00:00:00 2001 From: can1357 Date: Fri, 15 May 2026 00:55:56 +0200 Subject: [PATCH] test(python/omp-rpc): reworked host URI tests to capture uri_echo notifications - Adjusted the test server in test_host_uris.py to re-emit host_uri_result frames as uri_echo notifications. - Updated test client setup to attach an unknown-notification capture hook for uri_echo frames. - Changed _await_echo to read captured uri_echo frames from that hook and fail fast if capture is missing. --- .../coding-agent/test/rpc-host-uris.test.ts | 144 ++++++++++++++++++ python/omp-rpc/tests/test_host_uris.py | 38 +++-- 2 files changed, 169 insertions(+), 13 deletions(-) create mode 100644 packages/coding-agent/test/rpc-host-uris.test.ts diff --git a/packages/coding-agent/test/rpc-host-uris.test.ts b/packages/coding-agent/test/rpc-host-uris.test.ts new file mode 100644 index 000000000..a762c58b6 --- /dev/null +++ b/packages/coding-agent/test/rpc-host-uris.test.ts @@ -0,0 +1,144 @@ +import { afterEach, describe, expect, it } from "bun:test"; +import { InternalUrlRouter } from "@oh-my-pi/pi-coding-agent/internal-urls"; +import { parseInternalUrl } from "@oh-my-pi/pi-coding-agent/internal-urls/parse"; +import { RpcHostUriBridge } from "@oh-my-pi/pi-coding-agent/modes/rpc/host-uris"; +import type { RpcHostUriCancelRequest, RpcHostUriRequest } from "@oh-my-pi/pi-coding-agent/modes/rpc/rpc-types"; + +const router = InternalUrlRouter.instance(); + +afterEach(() => { + // Tests register transient schemes on the global router; clean them up + // between cases so neighboring suites observe the default registration. + router.unregister("db"); + router.unregister("notes"); + router.unregister("Db"); // case-insensitive guard +}); + +function recordOutput(): { + frames: Array; + push: (frame: RpcHostUriRequest | RpcHostUriCancelRequest) => void; +} { + const frames: Array = []; + return { frames, push: frame => frames.push(frame) }; +} + +describe("RpcHostUriBridge", () => { + it("registers schemes against the router and surfaces read results", async () => { + const out = recordOutput(); + const bridge = new RpcHostUriBridge(out.push); + + bridge.setSchemes([{ scheme: "db", description: "rows", writable: false }]); + expect(router.canHandle("db://users/42")).toBe(true); + + const pending = bridge.requestRead("db", parseInternalUrl("db://users/42")); + expect(out.frames).toHaveLength(1); + const request = out.frames[0]; + if (!request || request.type !== "host_uri_request") { + throw new Error("Expected host_uri_request frame"); + } + expect(request.operation).toBe("read"); + expect(request.url).toBe("db://users/42"); + + bridge.handleResult({ + type: "host_uri_result", + id: request.id, + content: "id=42", + contentType: "application/json", + notes: ["fresh"], + }); + + const resource = await pending; + expect(resource.content).toBe("id=42"); + expect(resource.contentType).toBe("application/json"); + expect(resource.notes).toEqual(["fresh"]); + bridge.clear("test cleanup"); + }); + + it("attaches a write hook only for writable schemes", async () => { + const out = recordOutput(); + const bridge = new RpcHostUriBridge(out.push); + bridge.setSchemes([ + { scheme: "db", writable: true }, + { scheme: "notes", writable: false }, + ]); + + const dbHandler = router.getHandler("db"); + const notesHandler = router.getHandler("notes"); + expect(typeof dbHandler?.write).toBe("function"); + expect(notesHandler?.write).toBeUndefined(); + + const url = parseInternalUrl("db://users/42"); + const pending = bridge.requestWrite("db", url, "new content"); + expect(out.frames).toHaveLength(1); + const request = out.frames[0]; + if (!request || request.type !== "host_uri_request") { + throw new Error("Expected host_uri_request frame"); + } + expect(request.operation).toBe("write"); + expect(request.content).toBe("new content"); + + bridge.handleResult({ type: "host_uri_result", id: request.id }); + await expect(pending).resolves.toBeUndefined(); + bridge.clear("test cleanup"); + }); + + it("propagates host-reported errors as exceptions", async () => { + const out = recordOutput(); + const bridge = new RpcHostUriBridge(out.push); + bridge.setSchemes([{ scheme: "db", writable: true }]); + + const url = parseInternalUrl("db://users/42"); + const pending = bridge.requestRead("db", url); + const request = out.frames[0]; + if (!request || request.type !== "host_uri_request") { + throw new Error("Expected host_uri_request frame"); + } + bridge.handleResult({ + type: "host_uri_result", + id: request.id, + isError: true, + error: "row not found", + }); + + await expect(pending).rejects.toThrow("row not found"); + bridge.clear("test cleanup"); + }); + + it("emits a cancel frame when the read signal aborts", async () => { + const out = recordOutput(); + const bridge = new RpcHostUriBridge(out.push); + bridge.setSchemes([{ scheme: "db" }]); + + const controller = new AbortController(); + const url = parseInternalUrl("db://users/42"); + const pending = bridge.requestRead("db", url, { signal: controller.signal }); + expect(out.frames).toHaveLength(1); + + controller.abort(); + await expect(pending).rejects.toThrow(/aborted/); + const cancel = out.frames[1]; + expect(cancel?.type).toBe("host_uri_cancel"); + bridge.clear("test cleanup"); + }); + + it("normalizes scheme casing and rejects invalid characters", () => { + const bridge = new RpcHostUriBridge(() => {}); + const accepted = bridge.setSchemes([{ scheme: " DB " }]); + expect(accepted).toEqual(["db"]); + expect(router.canHandle("db://x")).toBe(true); + + expect(() => bridge.setSchemes([{ scheme: "1bad" }])).toThrow(); + bridge.clear("test cleanup"); + }); + + it("replaces the registered set and unregisters schemes that drop off", () => { + const bridge = new RpcHostUriBridge(() => {}); + bridge.setSchemes([{ scheme: "db" }, { scheme: "notes" }]); + expect(router.canHandle("notes://idx")).toBe(true); + + bridge.setSchemes([{ scheme: "db" }]); + expect(router.canHandle("notes://idx")).toBe(false); + expect(router.canHandle("db://idx")).toBe(true); + bridge.clear("test cleanup"); + }); +}); diff --git a/python/omp-rpc/tests/test_host_uris.py b/python/omp-rpc/tests/test_host_uris.py index 632d47c13..486d96a11 100644 --- a/python/omp-rpc/tests/test_host_uris.py +++ b/python/omp-rpc/tests/test_host_uris.py @@ -73,14 +73,13 @@ URI_SERVER = textwrap.dedent( pending_uri_id += 1 respond(request_id, "trigger_write", {}) elif command_type == "host_uri_result": - # Echo back as response so the test can assert on the wire frame + # Re-emit the host_uri_result frame as an unknown notification + # so the test can capture it through on_unknown_notification. print( json.dumps( { - "type": "response", - "command": "uri_echo", - "success": True, - "data": {"frame": command}, + "type": "uri_echo", + "frame": command, } ), flush=True, @@ -132,12 +131,14 @@ class HostUriHelperTests(unittest.TestCase): class RpcHostUriBridgeTests(unittest.TestCase): def _make_client(self, **kwargs: object) -> RpcClient: - return RpcClient( + client = RpcClient( command=[sys.executable, "-u", "-c", URI_SERVER], startup_timeout=2.0, request_timeout=2.0, **kwargs, ) + self._attach_capture(client) + return client def test_set_host_uris_registers_schemes_on_start(self) -> None: captured: list[tuple[str, str]] = [] @@ -218,18 +219,29 @@ class RpcHostUriBridgeTests(unittest.TestCase): self.assertEqual(frame["error"], "boom") def _await_echo(self, client: RpcClient) -> dict: - # The fake server echoes the host_uri_result frame back as an - # `uri_echo` response. We poll the events history to surface it. + captured = getattr(client, "_test_uri_echos", None) + if captured is None: + self.fail("_capture was not called for this client") deadline = time.time() + 2.0 while time.time() < deadline: - with client._state_lock: # type: ignore[attr-defined] - events = client._events.snapshot() # type: ignore[attr-defined] - for event in events: - if event.get("command") == "uri_echo" and event.get("data"): - return event["data"]["frame"] + if captured: + return captured.pop(0) time.sleep(0.02) self.fail("Timed out waiting for host_uri_result echo") + def _attach_capture(self, client: RpcClient) -> None: + captured: list[dict] = [] + client._test_uri_echos = captured # type: ignore[attr-defined] + + def on_notification(notification) -> None: + payload = notification.payload + if payload.get("type") == "uri_echo": + frame = payload.get("frame") + if isinstance(frame, dict): + captured.append(frame) + + client.on_unknown_notification(on_notification) + if __name__ == "__main__": unittest.main()