diff --git a/tests/test_mission_bus.py b/tests/test_mission_bus.py new file mode 100644 index 00000000..22bc3873 --- /dev/null +++ b/tests/test_mission_bus.py @@ -0,0 +1,105 @@ +from importlib import util +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parent.parent +MODULE_PATH = ROOT / "nexus" / "mission_bus.py" +CONFIG_PATH = ROOT / "config" / "mission_bus_profiles.json" + + +def load_module(): + spec = util.spec_from_file_location("mission_bus", MODULE_PATH) + module = util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def build_bus(module): + profiles = module.load_profiles(CONFIG_PATH) + bus = module.MissionBus("mission-883", title="multi-agent teaming", config=profiles) + bus.register_participant("timmy", module.MissionRole.LEAD) + bus.register_participant("ezra", module.MissionRole.WRITE) + bus.register_participant("bezalel", module.MissionRole.READ) + bus.register_participant("allegro", module.MissionRole.AUDIT) + return bus + + +def test_role_permissions_gate_publish_checkpoint_and_handoff(): + module = load_module() + bus = build_bus(module) + + assert bus.allowed("timmy", "publish") is True + assert bus.allowed("ezra", "handoff") is True + assert bus.allowed("allegro", "audit") is True + assert bus.allowed("bezalel", "publish") is False + + with pytest.raises(PermissionError): + bus.publish("bezalel", "mission.notes", {"text": "should fail"}) + + with pytest.raises(PermissionError): + bus.create_checkpoint("allegro", summary="audit cannot checkpoint", state={}) + + +def test_mission_bus_unified_stream_records_messages_checkpoints_and_handoffs(): + module = load_module() + bus = build_bus(module) + + msg = bus.publish("timmy", "mission.start", {"goal": "build the slice"}) + checkpoint = bus.create_checkpoint( + "ezra", + summary="checkpoint before lead review", + state={"branch": "fix/883", "files": ["nexus/mission_bus.py"]}, + artifacts=["docs/mission-bus.md"], + ) + handoff = bus.handoff("ezra", "timmy", checkpoint.checkpoint_id, note="ready for lead review") + + assert [event.event_type for event in bus.events] == ["message", "checkpoint", "handoff"] + assert [event.sequence for event in bus.events] == [1, 2, 3] + assert msg.topic == "mission.start" + assert handoff.recipient == "timmy" + + +def test_handoff_resume_packet_contains_checkpoint_state_and_participants(): + module = load_module() + bus = build_bus(module) + checkpoint = bus.create_checkpoint( + "ezra", + summary="handoff package", + state={"branch": "fix/883", "tests": ["tests/test_mission_bus.py"]}, + artifacts=["config/mission_bus_profiles.json"], + ) + handoff = bus.handoff("ezra", "timmy", checkpoint.checkpoint_id, note="pick up from here") + + packet = bus.build_resume_packet(handoff.handoff_id) + assert packet["recipient"] == "timmy" + assert packet["checkpoint"]["state"]["branch"] == "fix/883" + assert packet["checkpoint"]["artifacts"] == ["config/mission_bus_profiles.json"] + assert packet["participants"]["ezra"]["role"] == "write" + assert packet["handoff_note"] == "pick up from here" + + +def test_profiles_define_level2_mount_namespace_and_level3_rootless_podman(): + module = load_module() + profiles = module.load_profiles(CONFIG_PATH) + + levels = {entry["level"]: entry["mechanism"] for entry in profiles["isolation_profiles"]} + assert levels[2] == "mount_namespace" + assert levels[3] == "rootless_podman" + assert profiles["roles"]["audit"] == ["read", "audit"] + + +def test_mission_bus_roundtrip_preserves_events_and_isolation_profile(): + module = load_module() + bus = build_bus(module) + bus.publish("timmy", "mission.start", {"goal": "roundtrip"}) + checkpoint = bus.create_checkpoint("ezra", summary="save state", state={"count": 1}) + bus.handoff("ezra", "timmy", checkpoint.checkpoint_id, note="resume") + + restored = module.MissionBus.from_dict(bus.to_dict()) + assert restored.mission_id == "mission-883" + assert restored.events[-1].event_type == "handoff" + assert restored.events[-1].note == "resume" + assert restored.isolation_profiles[1].mechanism == "mount_namespace"