Testing Guide

Testing

The test suite uses pytest with pytest-asyncio in auto mode. Tests live in tests/ and cover the service layer, external world protocol, snapshot/restore, and WebSocket transport.

Running tests

bash
# Install dev dependencies
pip install -e ".[websocket,dev]"

# Run all tests
pytest tests/

# Run with verbose output
pytest tests/ -v

# Run a specific test file
pytest tests/test_service.py

# Run a specific test function
pytest -k test_step_returns_agent_decision

# Short tracebacks
pytest --tb=short

# Type checking
mypy src/

Test files

FileWhat it tests
tests/test_external_world.pyExternalWorld protocol, HostedWorld push/observe/apply/collect, TickSummary, engine_commands flow, metadata-to-brain context
tests/test_obs_registry_reliability.pyObservationRegistry fault isolation (provider exceptions, warning logging, capability filtering), SocialContextProvider fault handling, Brain lifecycle (Closeable protocol, Simulation.close(), context manager)
Test suite is growing

The test suite currently covers the external world protocol and observation registry. Service layer, snapshot, and WebSocket transport tests are not yet present in the repository. See Contributing to add them.

Test patterns

Minimal simulation stub

Most tests build a minimal simulation with a fixed-output brain and a simple handler, rather than using real world implementations or LLMs. This pattern keeps tests fast and deterministic:

python
from src.contracts.action import ActionResult, ActionSchema, Intent
from src.engine.agent import Agent
from src.engine.registry import ActionRegistry
from src.engine.simulation import Simulation, SimulationConfig
from src.plugins.builtin.simple_memory.memory import SimpleMemory
from src.plugins.external.world import HostedWorld

class _FixedBrain:
    async def decide(self, agent, observation, actions, context) -> Intent:
        return Intent(action="act", parameters={"p": 1})

class _EchoHandler:
    def execute(self, agent, intent, context) -> ActionResult:
        return ActionResult(
            success=True,
            outcome_text="done",
            engine_commands=[{"type": "move", "dir": "north"}],
        )

def _make_sim(ticks: int = 5) -> Simulation:
    world = HostedWorld()
    agent = Agent(id="agent_001", name="Alice",
                  brain=_FixedBrain(), memory=SimpleMemory())
    registry = ActionRegistry()
    registry.register(ActionSchema("act", "Test."), _EchoHandler())
    return Simulation(
        agents   = [agent],
        world    = world,
        registry = registry,
        config   = SimulationConfig(ticks=ticks),
    )

Service layer tests (sync + async)

python
import pytest
from src.service.session import SimulationSession, create_session
from src.service.interfaces import SessionState, TickMode
from src.service.dto import StepRequest, AgentObservationDTO

class TestSimulationSessionLifecycle:
    def test_initial_state_is_created(self):
        session = SimulationSession(_make_sim())
        assert session.state == SessionState.CREATED

    async def test_step_transitions_to_running(self):
        session = SimulationSession(_make_sim())
        resp    = await session.step()
        assert session.state == SessionState.RUNNING
        assert resp.tick == 1

    async def test_step_with_observations(self):
        world   = HostedWorld()
        sim     = _make_sim()
        sim.world = world
        session = SimulationSession(sim)

        req = StepRequest(
            agent_observations=[
                AgentObservationDTO("agent_001", {"location": "castle"})
            ],
            world_metadata={"time_of_day": "dusk"},
        )
        resp = await session.step(req)
        assert resp.tick == 1
        assert len(resp.decisions) == 1

External world tests

python
import pytest
from src.contracts.world import ExternalWorld
from src.plugins.external.world import HostedWorld
from src.engine.simulation import TickSummary

def test_hosted_world_satisfies_external_world_protocol():
    assert isinstance(HostedWorld(), ExternalWorld)

def test_observe_returns_copy_not_reference():
    world = HostedWorld()
    obs   = {"location": "forest"}
    world.push_observation("a1", obs)
    world.observe("a1")["location"] = "mutated"
    assert world.observe("a1")["location"] == "forest"

@pytest.mark.asyncio
async def test_run_tick_returns_tick_summary():
    world = HostedWorld()
    world.push_observation("agent_001", {"location": "start"})
    sim = _make_sim_with_world(world)
    summary = await sim.run_tick()
    assert isinstance(summary, TickSummary)
    assert summary.tick == 1
    assert len(summary.agent_results) == 1
    assert summary.engine_commands()[0]["type"] == "move"

WebSocket transport tests

WebSocket tests spin up a real WebSocketServer on a random port (port=0) and connect a real websockets client. No mocking of the transport:

python
import websockets
import json
import pytest
from src.transport.websocket.server import WebSocketServer
from src.service import create_session

@pytest.fixture
async def server_and_session():
    sim     = _make_sim()
    session = create_session(sim)
    server  = WebSocketServer(session, port=0)
    port    = await server.start()
    yield server, session, port
    await server.stop()

async def test_hello_frame_received(server_and_session):
    server, session, port = server_and_session
    async with websockets.connect(f"ws://localhost:{port}") as ws:
        raw  = await ws.recv()
        hlo  = json.loads(raw)
        assert hlo["type"] == "hlo"
        assert hlo["v"] == 1
        assert hlo["tick_mode"] == "host_driven"
        assert "tick" in hlo["capabilities"]

async def test_register_and_tick(server_and_session):
    server, session, port = server_and_session
    async with websockets.connect(f"ws://localhost:{port}") as ws:
        await ws.recv()  # hello
        # Register agent
        await ws.send(json.dumps({"type":"req","v":1,"id":"1",
            "method":"register_agent",
            "params":{"agent_id":"t1","agent_name":"T",
                      "brain_class":"src.plugins.builtin.idle_brain.brain.IdleBrain"}}))
        res = json.loads(await ws.recv())
        assert res["ok"] is True
        # Tick
        await ws.send(json.dumps({"type":"req","v":1,"id":"2","method":"tick","params":{}}))
        res = json.loads(await ws.recv())
        assert res["ok"] is True
        assert res["result"]["tick"] == 1

pytest configuration

From pyproject.toml:

toml
[tool.pytest.ini_options]
asyncio_mode = "auto"   # all async test functions run automatically
testpaths    = ["tests"]

The asyncio_mode = "auto" setting means async test methods and async fixtures work without needing @pytest.mark.asyncio decorators (though they are sometimes added for clarity).

Testing your own code

Testing a custom brain

python
from src.contracts.action import Intent, ActionSchema
from src.contracts.brain import BrainContext
from src.contracts.world import AgentView

async def test_my_brain_returns_valid_action():
    from my_project.brains import MyBrain

    brain   = MyBrain()
    agent   = AgentView(id="a1", name="Test", inventory={})
    actions = [ActionSchema("move", "Move."), ActionSchema("idle", "Do nothing.")]
    context = BrainContext(tick=1, memory="", metadata={}, observation_schemas=[])
    obs     = {"location": "town"}

    intent = await brain.decide(agent, obs, actions, context)

    assert intent.action in {"move", "idle"}
    assert isinstance(intent.reasoning, str)

Testing a custom handler

python
from src.contracts.action import Intent, ActionResult
from src.contracts.world import AgentView
from src.plugins.external.world import HostedWorld

def test_move_handler_returns_engine_command():
    from my_project.handlers import MoveHandler

    handler = MoveHandler()
    agent   = AgentView(id="a1", name="Guard", inventory={})
    intent  = Intent(action="move", parameters={"direction": "north"})
    world   = HostedWorld()

    result = handler.execute(agent, intent, world)

    assert result.success is True
    assert len(result.engine_commands) == 1
    assert result.engine_commands[0]["type"] == "navigate"
    assert result.engine_commands[0]["direction"] == "north"

Testing snapshot round-trip

python
async def test_snapshot_restore_preserves_tick():
    world   = HostedWorld()
    sim     = _make_sim()   # _make_sim() should create sim with this world
    session = SimulationSession(sim)

    # Run 3 ticks
    for _ in range(3):
        world.push_observation("agent_001", {"location": "town"})
        await session.step()

    # Snapshot at tick 3
    snap = session.snapshot()
    assert snap.tick == 3   # SimulationSnapshot.tick

    # Run 2 more ticks
    for _ in range(2):
        await session.step()
    assert session.tick == 5   # session.tick is a property on SimulationSession

    # Restore to tick 3
    session.restore(snap)
    assert session.tick == 3