Spaces:
Paused
Paused
File size: 1,617 Bytes
02ff91f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """Tests for delegation graph DAG enforcement."""
import pytest
from env.delegation_graph import DelegationGraph
def test_no_self_delegation():
g = DelegationGraph(max_depth=2)
g.add_root("orchestrator")
assert not g.can_delegate("orchestrator", "orchestrator")
def test_basic_delegation():
g = DelegationGraph(max_depth=2)
g.add_root("orchestrator")
assert g.can_delegate("orchestrator", "frontend_react")
g.record_delegation("orchestrator", "frontend_react", "sequential")
assert "frontend_react" in g.get_called_specialists()
def test_cycle_prevention():
g = DelegationGraph(max_depth=3)
g.add_root("orchestrator")
g.record_delegation("orchestrator", "a", "sequential")
g.record_delegation("a", "b", "sequential")
# b -> orchestrator should be blocked (cycle)
assert not g.can_delegate("b", "orchestrator")
# b -> a should be blocked (cycle)
assert not g.can_delegate("b", "a")
def test_depth_enforcement():
g = DelegationGraph(max_depth=2)
g.add_root("orchestrator")
g.record_delegation("orchestrator", "a", "sequential")
g.record_delegation("a", "b", "sequential")
# depth 3 would exceed max_depth=2
assert not g.can_delegate("b", "c")
def test_adjacency_vector():
g = DelegationGraph(max_depth=2)
g.add_root("orchestrator")
g.record_delegation("orchestrator", "frontend_react", "parallel")
all_ids = ["orchestrator", "frontend_react", "backend_api"]
vec = g.to_adjacency_vector(all_ids, max_size=3)
assert len(vec) == 9 # 3x3
assert vec[1] == 1.0 # orchestrator->frontend_react edge
|