timmy-config/lazarus/tests/test_cell.py

#!/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()