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.
This commit is contained in:
can1357
@@ -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<RpcHostUriRequest | RpcHostUriCancelRequest>;
|
||||
push: (frame: RpcHostUriRequest | RpcHostUriCancelRequest) => void;
|
||||
} {
|
||||
const frames: Array<RpcHostUriRequest | RpcHostUriCancelRequest> = [];
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user