test_xros.py

"""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())