File size: 3,222 Bytes
2bc094c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """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())
|