test_delegation.py•2.99 kB
"""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