Spaces:
Sleeping
Sleeping
File size: 2,994 Bytes
8b02e7c | 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 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Tests for delegation engine."""
import pytest
from delegation_mcp.config import DelegationConfig, DelegationRule, OrchestratorConfig
from delegation_mcp.orchestrator import OrchestratorRegistry
from delegation_mcp.delegation import DelegationEngine
@pytest.fixture
def config():
"""Create test configuration."""
return DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(
name="claude",
command="python",
args=["-c", "print('claude output')"],
enabled=True,
),
"gemini": OrchestratorConfig(
name="gemini",
command="python",
args=["-c", "print('gemini output')"],
enabled=True,
),
},
rules=[
DelegationRule(
pattern="security",
delegate_to="gemini",
priority=5,
),
],
)
@pytest.fixture
def registry(config):
"""Create test registry."""
reg = OrchestratorRegistry()
for orch_config in config.orchestrators.values():
reg.register(orch_config)
return reg
@pytest.fixture
def engine(config, registry):
"""Create test engine."""
return DelegationEngine(config, registry)
def test_determine_delegation_with_match(engine):
"""Test delegation determination with rule match."""
target, rule = engine._determine_delegation(
query="Run security audit",
force_delegate=None,
)
assert target == "gemini"
assert rule is not None
assert rule.pattern == "security"
def test_determine_delegation_without_match(engine):
"""Test delegation determination without rule match."""
target, rule = engine._determine_delegation(
query="Explain Python",
force_delegate=None,
)
assert target == "claude"
assert rule is None
def test_determine_delegation_with_force(engine):
"""Test forced delegation."""
target, rule = engine._determine_delegation(
query="Run security audit",
force_delegate="gemini",
)
assert target == "gemini"
assert rule is None # Force overrides rules
@pytest.mark.asyncio
async def test_process_delegation(engine):
"""Test processing a query with delegation."""
result = await engine.process("Run security audit")
assert result.orchestrator == "claude"
assert result.delegated_to == "gemini"
assert result.rule is not None
assert result.success is True
def test_statistics(engine):
"""Test delegation statistics."""
stats = engine.get_statistics()
assert stats["total"] == 0
assert stats["delegations"] == 0
assert stats["by_orchestrator"] == {}
def test_clear_history(engine):
"""Test clearing delegation history."""
engine.history = [object(), object()] # Add dummy history
engine.clear_history()
assert len(engine.history) == 0
|