tests

test_xros.py

@@ -0,0 +1,109 @@
+"""test_xros.py — end-to-end test of the spatial shell + world-state service.
+
+Two layers:
+  1. Unit: drive the SpatialShell directly against a WorldState and assert
+     the world mutates correctly.
+  2. Integration: start the real WebSocket server, connect a client, run
+     shell commands over the wire, and confirm a second client observes the
+     world changes (the DiOS "single shared resource" claim).
+
+Run:  python test_xros.py
+"""
+
+import asyncio
+import json
+
+import websockets
+
+from xros_world import WorldState
+from xros_shell import SpatialShell
+from xros_server import handle
+from xros_client import WorldClient
+
+
+async def test_shell_unit() -> None:
+    world = WorldState()
+    shell = SpatialShell(world)
+
+    out = await shell.execute("help")
+    assert "spawn" in out, out
+
+    out = await shell.execute("spawn panel --title Hello --color blue")
+    assert "spawned panel" in out, out
+
+    out = await shell.execute("spawn cube 1 2 0")
+    assert "spawned cube" in out, out
+
+    out = await shell.execute("ls")
+    assert "panel" in out and "cube" in out, out
+
+    objs = await world.snapshot()
+    assert len(objs) == 2, objs
+    cube = next(o for o in objs if o["type"] == "cube")
+    assert cube["pose"]["position"] == [1.0, 2.0, 0.0], cube
+
+    out = await shell.execute(f"move {cube['id']} 5 5 5")
+    assert "moved" in out, out
+    moved = await world.get(cube["id"])
+    assert moved.pose.position == [5.0, 5.0, 5.0]
+
+    out = await shell.execute("gen a clock for the wall")
+    assert "generated clock panel" in out, out
+    assert any(o["properties"].get("panel_type") == "clock" for o in await world.snapshot())
+
+    out = await shell.execute("clear")
+    assert "cleared 3" in out, out
+    assert await world.snapshot() == []
+
+    out = await shell.execute("bogus")
+    assert out.startswith("error: unknown command"), out
+
+    print("unit tests: OK")
+
+
+async def test_integration() -> None:
+    world = WorldState()
+    shell = SpatialShell(world)
+
+    async def server():
+        async with websockets.serve(lambda ws: handle(ws, world, shell), "127.0.0.1", 8766):
+            await asyncio.Future()
+
+    srv = asyncio.create_task(server())
+    await asyncio.sleep(0.2)
+
+    uri = "ws://127.0.0.1:8766"
+
+    # Client A: the "terminal" — runs shell commands (filters world events).
+    a = WorldClient(uri)
+    await a.connect()
+    # Client B: the "world renderer" — observes every world event.
+    b = await websockets.connect(uri)
+
+    # A spawns a panel via the shell.
+    out = await a.command("spawn panel --title Hello")
+    assert "spawned panel" in out, out
+
+    # B should receive the spawn event (the shared-world claim).
+    event = json.loads(await asyncio.wait_for(b.recv(), timeout=5))
+    assert event["op"] == "spawn", event
+    assert event["object"]["type"] == "panel", event
+
+    # A lists the world.
+    out = await a.command("ls")
+    assert "panel" in out, out
+
+    await a.close()
+    await b.close()
+    srv.cancel()
+    print("integration test: OK (client B observed client A's spawn)")
+
+
+async def main() -> None:
+    await test_shell_unit()
+    await test_integration()
+    print("\nALL TESTS PASSED")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())