test_config.py•14.2 kB
"""Tests for configuration module."""
import pytest
from pathlib import Path
import tempfile
from delegation_mcp.config import (
DelegationConfig,
DelegationRule,
OrchestratorConfig,
ConfigValidationError,
)
def test_delegation_rule_creation():
"""Test creating a delegation rule."""
rule = DelegationRule(
pattern="security|audit",
delegate_to="gemini",
priority=5,
requires_approval=False,
description="Security tasks",
)
assert rule.pattern == "security|audit"
assert rule.delegate_to == "gemini"
assert rule.priority == 5
def test_orchestrator_config_creation():
"""Test creating orchestrator configuration."""
config = OrchestratorConfig(
name="claude",
command="claude",
enabled=True,
)
assert config.name == "claude"
assert config.command == "claude"
assert config.enabled is True
assert config.timeout == 300 # default
def test_delegation_config_find_rule():
"""Test finding matching delegation rule."""
config = DelegationConfig(
orchestrator="claude",
rules=[
DelegationRule(
pattern="security|audit",
delegate_to="gemini",
priority=5,
),
DelegationRule(
pattern="refactor",
delegate_to="aider",
priority=3,
),
],
)
# Should match security rule
rule = config.find_delegation_rule("Run a security audit")
assert rule is not None
assert rule.delegate_to == "gemini"
# Should match refactor rule
rule = config.find_delegation_rule("Refactor the code")
assert rule is not None
assert rule.delegate_to == "aider"
# Should not match any rule
rule = config.find_delegation_rule("Explain Python")
assert rule is None
def test_config_save_and_load():
"""Test saving and loading configuration."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = Path(tmpdir) / "test_config.yaml"
# Create and save config
config = DelegationConfig(
orchestrator="claude",
rules=[
DelegationRule(
pattern="test",
delegate_to="gemini",
priority=1,
),
],
)
config.to_yaml(config_path)
# Load config
loaded_config = DelegationConfig.from_yaml(config_path, validate=False)
assert loaded_config.orchestrator == "claude"
assert len(loaded_config.rules) == 1
assert loaded_config.rules[0].pattern == "test"
# ============================================================================
# Validation Tests
# ============================================================================
def test_validate_minimum_agents_success():
"""Test validation passes with 2 or more enabled agents."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
"aider": OrchestratorConfig(name="aider", command="aider", enabled=False),
},
rules=[],
)
# Should not raise exception
config.validate()
def test_validate_minimum_agents_failure_zero():
"""Test validation fails with no enabled agents."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=False),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=False),
},
rules=[],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
assert "At least 2 agents must be enabled" in str(exc_info.value)
assert "only 0 are enabled" in str(exc_info.value)
def test_validate_minimum_agents_failure_one():
"""Test validation fails with only one enabled agent."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=False),
},
rules=[],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
assert "At least 2 agents must be enabled" in str(exc_info.value)
assert "only 1 is enabled" in str(exc_info.value)
def test_validate_regex_patterns_success():
"""Test validation passes with valid regex patterns."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
},
rules=[
DelegationRule(pattern="security|audit", delegate_to="gemini", priority=5),
DelegationRule(pattern="refactor.*code", delegate_to="claude", priority=3),
DelegationRule(pattern="^test", delegate_to="gemini", priority=2),
],
)
# Should not raise exception
config.validate()
def test_validate_regex_patterns_failure():
"""Test validation fails with invalid regex patterns."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
},
rules=[
DelegationRule(
pattern="valid_pattern", delegate_to="gemini", priority=5
),
DelegationRule(
pattern="[invalid(regex", delegate_to="claude", priority=3
), # Invalid regex
DelegationRule(
pattern="(?P<incomplete", delegate_to="gemini", priority=2
), # Invalid regex
],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
error_msg = str(exc_info.value)
assert "Invalid regex pattern" in error_msg
assert "[invalid(regex" in error_msg or "(?P<incomplete" in error_msg
def test_validate_agent_references_success():
"""Test validation passes when all referenced agents exist."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
"aider": OrchestratorConfig(name="aider", command="aider", enabled=True),
},
rules=[
DelegationRule(pattern="security", delegate_to="gemini", priority=5),
DelegationRule(pattern="refactor", delegate_to="claude", priority=3),
DelegationRule(pattern="git", delegate_to="aider", priority=4),
],
)
# Should not raise exception
config.validate()
def test_validate_agent_references_failure_rule_target():
"""Test validation fails when rule references non-existent agent."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
},
rules=[
DelegationRule(pattern="security", delegate_to="gemini", priority=5),
DelegationRule(
pattern="git", delegate_to="nonexistent_agent", priority=3
), # Invalid reference
],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
error_msg = str(exc_info.value)
assert "Target orchestrator 'nonexistent_agent' is not defined" in error_msg
assert "pattern: 'git'" in error_msg
def test_validate_no_circular_delegation_success():
"""Test validation passes with normal delegation rules."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
"aider": OrchestratorConfig(name="aider", command="aider", enabled=True),
},
rules=[
DelegationRule(pattern="security", delegate_to="gemini", priority=5),
DelegationRule(pattern="refactor", delegate_to="claude", priority=3),
DelegationRule(pattern="git", delegate_to="aider", priority=4),
],
)
# Should not raise exception - multiple different patterns to different targets is fine
config.validate()
def test_validate_no_ambiguous_delegation():
"""Test validation detects ambiguous delegation patterns."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
"aider": OrchestratorConfig(name="aider", command="aider", enabled=True),
},
rules=[
# Same pattern, same priority, different targets - ambiguous!
DelegationRule(pattern="security", delegate_to="claude", priority=5),
DelegationRule(pattern="security", delegate_to="gemini", priority=5),
],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
error_msg = str(exc_info.value)
assert "ambiguous delegation" in error_msg.lower()
assert "security" in error_msg.lower()
def test_validate_same_pattern_different_priority_ok():
"""Test validation allows same pattern with different priorities."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
"aider": OrchestratorConfig(name="aider", command="aider", enabled=True),
},
rules=[
# Same pattern but different priorities - OK (higher priority wins)
DelegationRule(pattern="security", delegate_to="gemini", priority=5),
DelegationRule(pattern="security", delegate_to="claude", priority=3),
],
)
# Should not raise exception - different priorities resolve ambiguity
config.validate()
def test_validate_multiple_errors():
"""Test validation reports multiple errors at once."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=False),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=False),
# Error 1: less than 2 enabled agents
},
rules=[
DelegationRule(
pattern="[invalid(", delegate_to="claude", priority=5
), # Error 2: invalid regex
DelegationRule(
pattern="test", delegate_to="missing_agent", priority=3
), # Error 3: invalid reference
],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
error_msg = str(exc_info.value)
# Should contain multiple errors
assert "At least 2 agents must be enabled" in error_msg
assert "Invalid regex pattern" in error_msg
assert "Target orchestrator 'missing_agent' is not defined" in error_msg
def test_from_yaml_with_validation():
"""Test loading config from YAML with validation enabled."""
with tempfile.TemporaryDirectory() as tmpdir:
config_path = Path(tmpdir) / "test_config.yaml"
# Create invalid config
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(
name="claude", command="claude", enabled=False
),
"gemini": OrchestratorConfig(
name="gemini", command="gemini", enabled=False
),
},
rules=[],
)
config.to_yaml(config_path)
# Try to load with validation - should fail
with pytest.raises(ConfigValidationError):
DelegationConfig.from_yaml(config_path, validate=True)
# Load without validation - should succeed
loaded = DelegationConfig.from_yaml(config_path, validate=False)
assert loaded.orchestrator == "claude"
def test_validate_empty_orchestrators():
"""Test validation fails with no orchestrators defined."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={}, # Empty orchestrators
rules=[],
)
with pytest.raises(ConfigValidationError) as exc_info:
config.validate()
error_msg = str(exc_info.value)
assert "At least 2 agents must be enabled" in error_msg
assert "Primary orchestrator 'claude' is not defined" in error_msg
def test_validate_with_three_enabled_agents():
"""Test validation passes with more than minimum enabled agents."""
config = DelegationConfig(
orchestrator="claude",
orchestrators={
"claude": OrchestratorConfig(name="claude", command="claude", enabled=True),
"gemini": OrchestratorConfig(name="gemini", command="gemini", enabled=True),
"aider": OrchestratorConfig(name="aider", command="aider", enabled=True),
"copilot": OrchestratorConfig(
name="copilot", command="copilot", enabled=False
),
},
rules=[
DelegationRule(pattern="security", delegate_to="gemini", priority=5),
],
)
# Should not raise exception
config.validate()