#!/usr/bin/env python3 """Tests for Lazarus Pit cell isolation and registry.""" import os import sys import tempfile import unittest from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from lazarus.cell import Cell, CellMember, CellPackager, CellRegistry, CellRole, CellState from lazarus.operator_ctl import OperatorCtl from lazarus.backends.process_backend import ProcessBackend class TestCellPackager(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.TemporaryDirectory() self.packager = CellPackager(root=self.tmpdir.name) def tearDown(self): self.tmpdir.cleanup() def test_create_isolated_paths(self): cell = self.packager.create("laz-test-001", backend="process") self.assertTrue(os.path.isdir(cell.hermes_home)) self.assertTrue(os.path.isdir(cell.workspace_path)) self.assertTrue(os.path.isfile(cell.shared_notes_path)) self.assertTrue(os.path.isdir(cell.shared_artifacts_path)) self.assertTrue(os.path.isfile(cell.decisions_path)) def test_destroy_cleans_paths(self): cell = self.packager.create("laz-test-002", backend="process") self.assertTrue(os.path.isdir(cell.hermes_home)) self.packager.destroy("laz-test-002") self.assertFalse(os.path.exists(cell.hermes_home)) def test_cells_are_separate(self): c1 = self.packager.create("laz-a", backend="process") c2 = self.packager.create("laz-b", backend="process") self.assertNotEqual(c1.hermes_home, c2.hermes_home) self.assertNotEqual(c1.workspace_path, c2.workspace_path) class TestCellRegistry(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.TemporaryDirectory() self.registry = CellRegistry(db_path=os.path.join(self.tmpdir.name, "registry.sqlite")) def tearDown(self): self.tmpdir.cleanup() def test_save_and_load(self): cell = Cell( cell_id="laz-reg-001", project="Timmy_Foundation/test#1", owner="human:Alexander", backend="process", state=CellState.ACTIVE, created_at=datetime.now().astimezone().isoformat(), ttl_minutes=60, members=[CellMember("agent:allegro", CellRole.EXECUTOR, datetime.now().astimezone().isoformat())], ) self.registry.save(cell) loaded = self.registry.load("laz-reg-001") self.assertEqual(loaded.cell_id, "laz-reg-001") self.assertEqual(loaded.members[0].role, CellRole.EXECUTOR) def test_list_active(self): for sid, state in [("a", CellState.ACTIVE), ("b", CellState.DESTROYED)]: cell = Cell( cell_id=f"laz-{sid}", project="test", owner="human:Alexander", backend="process", state=state, created_at=datetime.now().astimezone().isoformat(), ttl_minutes=60, ) self.registry.save(cell) active = self.registry.list_active() ids = {c.cell_id for c in active} self.assertIn("laz-a", ids) self.assertNotIn("laz-b", ids) def test_ttl_expiry(self): created = (datetime.now().astimezone() - timedelta(minutes=61)).isoformat() cell = Cell( cell_id="laz-exp", project="test", owner="human:Alexander", backend="process", state=CellState.ACTIVE, created_at=created, ttl_minutes=60, ) self.assertTrue(cell.is_expired()) def test_ttl_not_expired(self): cell = Cell( cell_id="laz-ok", project="test", owner="human:Alexander", backend="process", state=CellState.ACTIVE, created_at=datetime.now().astimezone().isoformat(), ttl_minutes=60, ) self.assertFalse(cell.is_expired()) class TestOperatorCtl(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.TemporaryDirectory() self.ctl = OperatorCtl( registry=CellRegistry(db_path=os.path.join(self.tmpdir.name, "registry.sqlite")), packager=CellPackager(root=self.tmpdir.name), ) def tearDown(self): # Destroy any cells we created for cell in self.ctl.registry.list_active(): self.ctl.destroy(cell.cell_id) self.tmpdir.cleanup() def test_summon_creates_cell(self): res = self.ctl.summon("allegro", "Timmy_Foundation/test#1") self.assertTrue(res["success"]) self.assertIn("cell_id", res) cell = self.ctl.registry.load(res["cell_id"]) self.assertEqual(cell.members[0].identity, "agent:allegro") self.assertEqual(cell.state, CellState.ACTIVE) # Clean up self.ctl.destroy(res["cell_id"]) def test_team_creates_multi_agent_cell(self): res = self.ctl.team(["allegro", "ezra"], "Timmy_Foundation/test#2") self.assertTrue(res["success"]) cell = self.ctl.registry.load(res["cell_id"]) self.assertEqual(len(cell.members), 2) identities = {m.identity for m in cell.members} self.assertEqual(identities, {"agent:allegro", "agent:ezra"}) self.ctl.destroy(res["cell_id"]) def test_invite_adds_guest(self): res = self.ctl.summon("allegro", "Timmy_Foundation/test#3") cell_id = res["cell_id"] invite = self.ctl.invite(cell_id, "bot:kimi", "observer") self.assertTrue(invite["success"]) cell = self.ctl.registry.load(cell_id) roles = {m.identity: m.role for m in cell.members} self.assertEqual(roles["bot:kimi"], CellRole.OBSERVER) self.ctl.destroy(cell_id) def test_status_shows_active_cells(self): res = self.ctl.summon("allegro", "Timmy_Foundation/test#4") status = self.ctl.status() self.assertTrue(status["success"]) ids = {c["cell_id"] for c in status["active_cells"]} self.assertIn(res["cell_id"], ids) self.ctl.destroy(res["cell_id"]) def test_close_then_destroy(self): res = self.ctl.summon("allegro", "Timmy_Foundation/test#5") cell_id = res["cell_id"] close_res = self.ctl.close(cell_id) self.assertTrue(close_res["success"]) destroy_res = self.ctl.destroy(cell_id) self.assertTrue(destroy_res["success"]) cell = self.ctl.registry.load(cell_id) self.assertEqual(cell.state, CellState.DESTROYED) self.assertFalse(os.path.exists(cell.hermes_home)) def test_destroy_scrubs_filesystem(self): res = self.ctl.summon("allegro", "Timmy_Foundation/test#6") cell_id = res["cell_id"] cell = self.ctl.registry.load(cell_id) hermes = cell.hermes_home self.assertTrue(os.path.isdir(hermes)) self.ctl.destroy(cell_id) self.assertFalse(os.path.exists(hermes)) if __name__ == "__main__": unittest.main()