"""TDD — I1: Acyclicity invariant for the identity DAG. Every test here drives one behaviour of CycleDetector. No implementation exists yet; all tests must fail red before we write code. """ from __future__ import annotations import pytest from musehub.graph.dag import EdgeType, GraphEdge, IdentityDAG, NodeType from musehub.graph.cycle import CycleDetector, CycleError # ── helpers ────────────────────────────────────────────────────────────────── def dag(*edges: tuple[str, str, EdgeType]) -> IdentityDAG: """Build an IdentityDAG from (from, to, type) triples.""" return IdentityDAG.from_edges( [GraphEdge(from_handle=f, to_handle=t, edge_type=e) for f, t, e in edges] ) S = EdgeType.SPAWNS M = EdgeType.MEMBER_OF # ── SPAWNS edge — valid cases ───────────────────────────────────────────────── class TestSpawnsValid: def test_empty_graph_accepts_any_spawn(self) -> None: d = IdentityDAG.empty() CycleDetector(d).assert_no_cycle("alice", "bot-1", S) def test_human_spawns_agent(self) -> None: d = IdentityDAG.empty() CycleDetector(d).assert_no_cycle("alice", "bot-1", S) def test_agent_spawns_agent(self) -> None: d = dag(("alice", "bot-1", S)) CycleDetector(d).assert_no_cycle("bot-1", "bot-2", S) def test_org_spawns_agent(self) -> None: d = dag(("alice", "acme", M)) CycleDetector(d).assert_no_cycle("acme", "acme-bot", S) def test_deep_spawn_chain_valid(self) -> None: d = dag( ("alice", "a1", S), ("a1", "a2", S), ("a2", "a3", S), ("a3", "a4", S), ) CycleDetector(d).assert_no_cycle("a4", "a5", S) def test_diamond_dag_no_cycle(self) -> None: # alice → a1 and alice → a2, both → a3: valid DAG, no cycle d = dag( ("alice", "a1", S), ("alice", "a2", S), ("a1", "a3", S), ("a2", "a3", S), ) CycleDetector(d).assert_no_cycle("alice", "a4", S) # ── SPAWNS edge — cycle cases ───────────────────────────────────────────────── class TestSpawnsCycle: def test_self_spawn_rejected(self) -> None: d = IdentityDAG.empty() with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("bot-1", "bot-1", S) def test_direct_spawn_cycle(self) -> None: # alice → bot-1; bot-1 tries to spawn alice d = dag(("alice", "bot-1", S)) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("bot-1", "alice", S) def test_indirect_spawn_cycle_three_nodes(self) -> None: # alice → a1 → a2; a2 tries to spawn alice d = dag(("alice", "a1", S), ("a1", "a2", S)) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("a2", "alice", S) def test_indirect_spawn_cycle_five_nodes(self) -> None: d = dag( ("h", "a1", S), ("a1", "a2", S), ("a2", "a3", S), ("a3", "a4", S), ) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("a4", "h", S) # ── MEMBER_OF edge — valid cases ────────────────────────────────────────────── class TestMemberOfValid: def test_human_joins_org(self) -> None: d = IdentityDAG.empty() CycleDetector(d).assert_no_cycle("alice", "acme", M) def test_agent_joins_org(self) -> None: d = dag(("alice", "bot-1", S)) CycleDetector(d).assert_no_cycle("bot-1", "acme", M) def test_org_joins_parent_org(self) -> None: d = dag(("alice", "acme", M)) CycleDetector(d).assert_no_cycle("acme", "conglomerate", M) def test_deep_org_nesting_valid(self) -> None: d = dag( ("alice", "org-a", M), ("org-a", "org-b", M), ("org-b", "org-c", M), ) CycleDetector(d).assert_no_cycle("org-c", "org-d", M) def test_multiple_members_in_org_valid(self) -> None: d = dag( ("alice", "acme", M), ("bob", "acme", M), ("bot-1", "acme", M), ) CycleDetector(d).assert_no_cycle("carol", "acme", M) # ── MEMBER_OF edge — cycle cases ────────────────────────────────────────────── class TestMemberOfCycle: def test_self_membership_rejected(self) -> None: d = IdentityDAG.empty() with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("acme", "acme", M) def test_direct_org_cycle(self) -> None: # acme has sub-org as member; sub-org tries to add acme as member d = dag(("sub-org", "acme", M)) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("acme", "sub-org", M) def test_indirect_org_cycle_three(self) -> None: d = dag(("org-a", "org-b", M), ("org-b", "org-c", M)) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("org-c", "org-a", M) def test_indirect_org_cycle_five(self) -> None: d = dag( ("org-a", "org-b", M), ("org-b", "org-c", M), ("org-c", "org-d", M), ("org-d", "org-e", M), ) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("org-e", "org-a", M) # ── Cross-edge-type cycles ───────────────────────────────────────────────────── class TestCrossEdgeCycle: def test_spawns_then_member_of_cycle(self) -> None: # alice spawns bot-1; bot-1 is member of acme; acme tries to spawn alice d = dag(("alice", "bot-1", S), ("bot-1", "acme", M)) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("acme", "alice", S) def test_member_of_then_spawns_cycle(self) -> None: # alice is member of acme; acme spawns bot-1; bot-1 tries to add alice as spawned d = dag(("alice", "acme", M), ("acme", "bot-1", S)) with pytest.raises(CycleError): CycleDetector(d).assert_no_cycle("bot-1", "alice", S) def test_complex_cross_edge_valid(self) -> None: # Verify a legitimate complex DAG is not falsely rejected # alice → acme (member), alice spawns bot-1, bot-1 → lab (member) d = dag( ("alice", "acme", M), ("alice", "bot-1", S), ("bot-1", "lab", M), ) CycleDetector(d).assert_no_cycle("bob", "acme", M)