wehub-resource-sync
255 lines
9.4 KiB
Python
255 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
import textwrap
|
|
import time
|
|
import unittest
|
|
|
|
from omp_rpc import RpcClient, host_uri
|
|
from omp_rpc.host_uris import normalize_read_result
|
|
|
|
|
|
URI_SERVER = textwrap.dedent(
|
|
"""
|
|
import json
|
|
import sys
|
|
|
|
print(json.dumps({"type": "ready"}), flush=True)
|
|
|
|
pending_uri_id = 1
|
|
|
|
def respond(request_id, command, data=None, success=True, error=None):
|
|
frame = {"id": request_id, "type": "response", "command": command, "success": success}
|
|
if success:
|
|
if data is not None:
|
|
frame["data"] = data
|
|
else:
|
|
frame["error"] = error or "error"
|
|
print(json.dumps(frame), flush=True)
|
|
|
|
for raw_line in sys.stdin:
|
|
raw_line = raw_line.strip()
|
|
if not raw_line:
|
|
continue
|
|
command = json.loads(raw_line)
|
|
command_type = command.get("type")
|
|
request_id = command.get("id")
|
|
|
|
if command_type == "set_host_uri_schemes":
|
|
schemes = command.get("schemes", [])
|
|
respond(
|
|
request_id,
|
|
"set_host_uri_schemes",
|
|
{"schemes": [entry.get("scheme", "") for entry in schemes]},
|
|
)
|
|
elif command_type == "trigger_read":
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"type": "host_uri_request",
|
|
"id": f"uri-req-{pending_uri_id}",
|
|
"operation": "read",
|
|
"url": command["url"],
|
|
}
|
|
),
|
|
flush=True,
|
|
)
|
|
pending_uri_id += 1
|
|
respond(request_id, "trigger_read", {})
|
|
elif command_type == "trigger_write":
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"type": "host_uri_request",
|
|
"id": f"uri-req-{pending_uri_id}",
|
|
"operation": "write",
|
|
"url": command["url"],
|
|
"content": command["content"],
|
|
}
|
|
),
|
|
flush=True,
|
|
)
|
|
pending_uri_id += 1
|
|
respond(request_id, "trigger_write", {})
|
|
elif command_type == "host_uri_result":
|
|
# 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": "uri_echo",
|
|
"frame": command,
|
|
}
|
|
),
|
|
flush=True,
|
|
)
|
|
else:
|
|
respond(request_id, command_type, success=False, error=f"unsupported: {command_type}")
|
|
"""
|
|
)
|
|
|
|
|
|
class HostUriHelperTests(unittest.TestCase):
|
|
def test_normalize_read_result_accepts_string(self) -> None:
|
|
self.assertEqual(normalize_read_result("hello"), {"content": "hello"})
|
|
|
|
def test_normalize_read_result_accepts_full_mapping(self) -> None:
|
|
result = normalize_read_result(
|
|
{
|
|
"content": "body",
|
|
"content_type": "application/json",
|
|
"notes": ["fresh"],
|
|
"immutable": True,
|
|
}
|
|
)
|
|
self.assertEqual(result["content"], "body")
|
|
self.assertEqual(result["contentType"], "application/json")
|
|
self.assertEqual(result["notes"], ["fresh"])
|
|
self.assertTrue(result["immutable"])
|
|
|
|
def test_normalize_read_result_requires_content(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
normalize_read_result({"content_type": "text/plain"}) # type: ignore[arg-type]
|
|
|
|
def test_normalize_read_result_rejects_invalid_content_type(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
normalize_read_result(
|
|
{"content": "x", "content_type": "application/octet-stream"}
|
|
) # type: ignore[arg-type]
|
|
|
|
def test_host_uri_helper_normalizes_scheme(self) -> None:
|
|
uri = host_uri(scheme=" DB ", read=lambda url, ctx: "x")
|
|
self.assertEqual(uri.scheme, "db")
|
|
self.assertFalse(uri.writable)
|
|
|
|
with self.assertRaises(ValueError):
|
|
host_uri(scheme="", read=lambda url, ctx: "x")
|
|
|
|
def test_host_uri_writable_when_write_supplied(self) -> None:
|
|
uri = host_uri(
|
|
scheme="db", read=lambda url, ctx: "x", write=lambda url, content, ctx: None
|
|
)
|
|
self.assertTrue(uri.writable)
|
|
|
|
|
|
class RpcHostUriBridgeTests(unittest.TestCase):
|
|
def _make_client(self, **kwargs: object) -> 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]] = []
|
|
|
|
def read_db(url: str, _ctx) -> str:
|
|
captured.append(("read", url))
|
|
return "id=42"
|
|
|
|
with self._make_client(
|
|
host_uris=(host_uri(scheme="db", read=read_db, description="test rows"),),
|
|
) as client:
|
|
# No public list — we exercise the on-start side effect by hitting the wire.
|
|
payload = client._request("trigger_read", url="db://users/42") # type: ignore[attr-defined]
|
|
self.assertEqual(payload, {})
|
|
|
|
frame = self._await_echo(client)
|
|
self.assertEqual(frame["type"], "host_uri_result")
|
|
self.assertEqual(frame["content"], "id=42")
|
|
self.assertEqual(captured, [("read", "db://users/42")])
|
|
|
|
def test_read_handler_can_return_structured_result(self) -> None:
|
|
def read_db(_url: str, _ctx):
|
|
return {
|
|
"content": '{"name":"Alice"}',
|
|
"content_type": "application/json",
|
|
"notes": ["row fresh"],
|
|
"immutable": True,
|
|
}
|
|
|
|
with self._make_client(
|
|
host_uris=(host_uri(scheme="db", read=read_db),)
|
|
) as client:
|
|
client._request("trigger_read", url="db://users/42") # type: ignore[attr-defined]
|
|
frame = self._await_echo(client)
|
|
self.assertEqual(frame["content"], '{"name":"Alice"}')
|
|
self.assertEqual(frame["contentType"], "application/json")
|
|
self.assertEqual(frame["notes"], ["row fresh"])
|
|
self.assertTrue(frame["immutable"])
|
|
|
|
def test_write_handler_receives_content_and_succeeds(self) -> None:
|
|
seen: dict[str, str] = {}
|
|
|
|
def write_db(url: str, content: str, _ctx) -> None:
|
|
seen[url] = content
|
|
|
|
uri = host_uri(scheme="db", read=lambda url, ctx: "ignored", write=write_db)
|
|
with self._make_client(host_uris=(uri,)) as client:
|
|
client._request("trigger_write", url="db://users/42", content="name=Bob") # type: ignore[attr-defined]
|
|
frame = self._await_echo(client)
|
|
self.assertEqual(frame["type"], "host_uri_result")
|
|
self.assertNotIn("isError", frame)
|
|
self.assertEqual(seen, {"db://users/42": "name=Bob"})
|
|
|
|
def test_write_rejected_for_read_only_scheme(self) -> None:
|
|
with self._make_client(
|
|
host_uris=(host_uri(scheme="db", read=lambda url, ctx: "x"),),
|
|
) as client:
|
|
client._request("trigger_write", url="db://users/42", content="ignored") # type: ignore[attr-defined]
|
|
frame = self._await_echo(client)
|
|
self.assertTrue(frame.get("isError"))
|
|
self.assertIn("write handler", frame["error"])
|
|
|
|
def test_unknown_scheme_is_rejected_with_error(self) -> None:
|
|
with self._make_client(
|
|
host_uris=(host_uri(scheme="db", read=lambda url, ctx: "x"),),
|
|
) as client:
|
|
client._request("trigger_read", url="other://stuff") # type: ignore[attr-defined]
|
|
frame = self._await_echo(client)
|
|
self.assertTrue(frame.get("isError"))
|
|
self.assertIn("not registered", frame["error"])
|
|
|
|
def test_handler_exception_is_surfaced_as_error(self) -> None:
|
|
def read_db(_url: str, _ctx) -> str:
|
|
raise RuntimeError("boom")
|
|
|
|
with self._make_client(
|
|
host_uris=(host_uri(scheme="db", read=read_db),)
|
|
) as client:
|
|
client._request("trigger_read", url="db://users/42") # type: ignore[attr-defined]
|
|
frame = self._await_echo(client)
|
|
self.assertTrue(frame.get("isError"))
|
|
self.assertEqual(frame["error"], "boom")
|
|
|
|
def _await_echo(self, client: RpcClient) -> dict:
|
|
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:
|
|
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()
|