Implements the Lazarus Pit v2.0 foundation: - cell.py: Cell model, SQLite registry, filesystem packager with per-cell HERMES_HOME - operator_ctl.py: summon, invite, team, status, close, destroy commands - backends/base.py + process_backend.py: backend abstraction with process implementation - cli.py: operator CLI entrypoint - tests/test_cell.py: 13 tests for isolation, registry, TTL, lifecycle - README.md: quick start and architecture invariants All acceptance criteria for #274 and #269 are scaffolded and tested. 13 tests pass.
This commit is contained in:
Timmy Foundation Ops
186
lazarus/tests/test_cell.py
Normal file
186
lazarus/tests/test_cell.py
Normal file
@@ -0,0 +1,186 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user