Spaces:
Running
Running
| """Tests for the reflexive engine.""" | |
| import pytest | |
| from datetime import datetime | |
| from unittest.mock import Mock, AsyncMock | |
| from fastmcp.reflexive import ReflexiveEngine, ActionContext, DecisionType, RiskLevel | |
| from fastmcp.reflexive.engine import ReflexiveDecision | |
| class TestReflexiveEngine: | |
| """Test the reflexive engine functionality.""" | |
| def reflexive_engine(self): | |
| """Create a reflexive engine for testing.""" | |
| return ReflexiveEngine() | |
| def action_context(self): | |
| """Create an action context for testing.""" | |
| return ActionContext( | |
| action_id="test_action_123", | |
| actor_id="test_user", | |
| action_type="tool_call", | |
| resource_id="test_resource", | |
| metadata={"test": "data"} | |
| ) | |
| def test_engine_initialization(self, reflexive_engine): | |
| """Test reflexive engine initialization.""" | |
| assert reflexive_engine.policy_engine is None | |
| assert reflexive_engine.ledger is None | |
| assert len(reflexive_engine.monitors) == 0 | |
| assert len(reflexive_engine.decision_handlers) == 4 | |
| assert not reflexive_engine.is_running | |
| def test_engine_status(self, reflexive_engine): | |
| """Test getting engine status.""" | |
| status = reflexive_engine.get_engine_status() | |
| assert "is_running" in status | |
| assert "monitor_count" in status | |
| assert "queue_size" in status | |
| assert "decision_handlers" in status | |
| assert status["is_running"] is False | |
| assert status["monitor_count"] == 0 | |
| assert len(status["decision_handlers"]) == 4 | |
| def test_add_remove_monitor(self, reflexive_engine): | |
| """Test adding and removing monitors.""" | |
| def test_monitor(context): | |
| return None | |
| # Add monitor | |
| reflexive_engine.add_monitor(test_monitor) | |
| assert len(reflexive_engine.monitors) == 1 | |
| assert test_monitor in reflexive_engine.monitors | |
| # Remove monitor | |
| reflexive_engine.remove_monitor(test_monitor) | |
| assert len(reflexive_engine.monitors) == 0 | |
| assert test_monitor not in reflexive_engine.monitors | |
| async def test_submit_action(self, reflexive_engine, action_context): | |
| """Test submitting an action for evaluation.""" | |
| await reflexive_engine.submit_action(action_context) | |
| # Check that action was added to queue | |
| assert reflexive_engine.event_queue.qsize() == 1 | |
| async def test_evaluate_action_no_violations(self, reflexive_engine, action_context): | |
| """Test evaluating an action with no violations.""" | |
| decision = await reflexive_engine._evaluate_action(action_context) | |
| assert decision.decision_type == DecisionType.ALLOW | |
| assert decision.risk_level == RiskLevel.LOW | |
| assert decision.reason == "No violations or anomalies detected" | |
| assert decision.action_context == action_context | |
| assert decision.proof_hash is not None | |
| async def test_evaluate_action_with_violations(self, reflexive_engine, action_context): | |
| """Test evaluating an action with violations.""" | |
| # Add a monitor that returns violations | |
| def violation_monitor(context): | |
| return { | |
| "type": "violation", | |
| "severity": "high", | |
| "violations": [{"rule": "test_rule", "message": "Test violation", "severity": "high"}] | |
| } | |
| reflexive_engine.add_monitor(violation_monitor) | |
| decision = await reflexive_engine._evaluate_action(action_context) | |
| assert decision.decision_type == DecisionType.HALT | |
| assert decision.risk_level == RiskLevel.HIGH | |
| assert "violations" in decision.evidence | |
| assert len(decision.evidence["violations"]) == 1 | |
| async def test_evaluate_action_with_anomalies(self, reflexive_engine, action_context): | |
| """Test evaluating an action with anomalies.""" | |
| # Add a monitor that returns anomalies | |
| def anomaly_monitor(context): | |
| return { | |
| "type": "anomaly", | |
| "severity": "medium", | |
| "anomalies": [{"type": "test_anomaly", "message": "Test anomaly", "severity": "medium"}] | |
| } | |
| reflexive_engine.add_monitor(anomaly_monitor) | |
| decision = await reflexive_engine._evaluate_action(action_context) | |
| assert decision.decision_type == DecisionType.ESCALATE | |
| assert decision.risk_level == RiskLevel.MEDIUM | |
| assert "anomalies" in decision.evidence | |
| assert len(decision.evidence["anomalies"]) == 1 | |
| async def test_evaluate_action_critical_violation(self, reflexive_engine, action_context): | |
| """Test evaluating an action with critical violations.""" | |
| # Add a monitor that returns critical violations | |
| def critical_monitor(context): | |
| return { | |
| "type": "violation", | |
| "severity": "critical", | |
| "violations": [{"rule": "critical_rule", "message": "Critical violation", "severity": "critical"}] | |
| } | |
| reflexive_engine.add_monitor(critical_monitor) | |
| decision = await reflexive_engine._evaluate_action(action_context) | |
| assert decision.decision_type == DecisionType.HALT | |
| assert decision.risk_level == RiskLevel.CRITICAL | |
| async def test_simulate_risk(self, reflexive_engine): | |
| """Test risk simulation.""" | |
| risk_scenario = { | |
| "action_context": { | |
| "action_id": "simulation_action", | |
| "actor_id": "test_actor", | |
| "action_type": "admin_access", | |
| "resource_id": "sensitive_data" | |
| }, | |
| "monitors": [ | |
| lambda ctx: { | |
| "type": "violation", | |
| "severity": "high", | |
| "violations": [{"rule": "admin_restriction", "message": "Unauthorized admin access", "severity": "high"}] | |
| } | |
| ] | |
| } | |
| decision = await reflexive_engine.simulate_risk(risk_scenario) | |
| assert decision.decision_type == DecisionType.HALT | |
| assert decision.risk_level == RiskLevel.HIGH | |
| assert decision.action_context.action_id == "simulation_action" | |
| def test_assess_risk_level(self, reflexive_engine): | |
| """Test risk level assessment.""" | |
| # Test critical risk | |
| critical_violations = [{"severity": "critical"}] | |
| assert reflexive_engine._assess_risk_level(critical_violations, []) == RiskLevel.CRITICAL | |
| # Test high risk | |
| high_violations = [{"severity": "high"}] | |
| assert reflexive_engine._assess_risk_level(high_violations, []) == RiskLevel.HIGH | |
| # Test medium risk | |
| medium_violations = [{"severity": "medium"}] | |
| assert reflexive_engine._assess_risk_level(medium_violations, []) == RiskLevel.MEDIUM | |
| # Test low risk | |
| low_violations = [{"severity": "low"}] | |
| assert reflexive_engine._assess_risk_level(low_violations, []) == RiskLevel.LOW | |
| # Test multiple issues | |
| multiple_issues = [{"severity": "low"}, {"severity": "low"}, {"severity": "low"}] | |
| assert reflexive_engine._assess_risk_level(multiple_issues, []) == RiskLevel.MEDIUM | |
| def test_determine_escalation_target(self, reflexive_engine): | |
| """Test escalation target determination.""" | |
| decision = ReflexiveDecision( | |
| decision_type=DecisionType.ESCALATE, | |
| risk_level=RiskLevel.CRITICAL, | |
| action_context=ActionContext(action_id="test", actor_id="test", action_type="test"), | |
| reason="Test" | |
| ) | |
| target = reflexive_engine._determine_escalation_target(decision) | |
| assert target == "security_admin" | |
| decision.risk_level = RiskLevel.HIGH | |
| target = reflexive_engine._determine_escalation_target(decision) | |
| assert target == "system_admin" | |
| decision.risk_level = RiskLevel.MEDIUM | |
| target = reflexive_engine._determine_escalation_target(decision) | |
| assert target == "monitoring_team" | |
| class TestActionContext: | |
| """Test the ActionContext class.""" | |
| def test_action_context_creation(self): | |
| """Test creating an action context.""" | |
| context = ActionContext( | |
| action_id="test_action", | |
| actor_id="test_actor", | |
| action_type="test_type" | |
| ) | |
| assert context.action_id == "test_action" | |
| assert context.actor_id == "test_actor" | |
| assert context.action_type == "test_type" | |
| assert context.resource_id is None | |
| assert context.metadata == {} | |
| assert isinstance(context.timestamp, datetime) | |
| def test_action_context_hash(self): | |
| """Test action context hash generation.""" | |
| context1 = ActionContext( | |
| action_id="test_action", | |
| actor_id="test_actor", | |
| action_type="test_type", | |
| metadata={"key": "value"} | |
| ) | |
| context2 = ActionContext( | |
| action_id="test_action", | |
| actor_id="test_actor", | |
| action_type="test_type", | |
| metadata={"key": "value"} | |
| ) | |
| # Same content should produce same hash | |
| assert context1.get_context_hash() == context2.get_context_hash() | |
| # Different content should produce different hash | |
| context3 = ActionContext( | |
| action_id="different_action", | |
| actor_id="test_actor", | |
| action_type="test_type", | |
| metadata={"key": "value"} | |
| ) | |
| assert context1.get_context_hash() != context3.get_context_hash() | |
| class TestReflexiveDecision: | |
| """Test the ReflexiveDecision class.""" | |
| def test_decision_creation(self): | |
| """Test creating a reflexive decision.""" | |
| action_context = ActionContext( | |
| action_id="test_action", | |
| actor_id="test_actor", | |
| action_type="test_type" | |
| ) | |
| decision = ReflexiveDecision( | |
| decision_type=DecisionType.HALT, | |
| risk_level=RiskLevel.HIGH, | |
| action_context=action_context, | |
| reason="Test reason" | |
| ) | |
| assert decision.decision_type == DecisionType.HALT | |
| assert decision.risk_level == RiskLevel.HIGH | |
| assert decision.action_context == action_context | |
| assert decision.reason == "Test reason" | |
| assert decision.evidence == {} | |
| assert decision.escalated_to is None | |
| assert decision.proof_hash is None | |
| assert isinstance(decision.timestamp, datetime) | |
| def test_decision_hash(self): | |
| """Test decision hash generation.""" | |
| from uuid import uuid4 | |
| action_context = ActionContext( | |
| action_id="test_action", | |
| actor_id="test_actor", | |
| action_type="test_type" | |
| ) | |
| # Use the same decision_id for both decisions | |
| decision_id = uuid4() | |
| decision1 = ReflexiveDecision( | |
| decision_id=decision_id, | |
| decision_type=DecisionType.HALT, | |
| risk_level=RiskLevel.HIGH, | |
| action_context=action_context, | |
| reason="Test reason" | |
| ) | |
| decision2 = ReflexiveDecision( | |
| decision_id=decision_id, | |
| decision_type=DecisionType.HALT, | |
| risk_level=RiskLevel.HIGH, | |
| action_context=action_context, | |
| reason="Test reason" | |
| ) | |
| # Same content should produce same hash | |
| assert decision1.get_decision_hash() == decision2.get_decision_hash() | |
| # Different content should produce different hash | |
| decision3 = ReflexiveDecision( | |
| decision_type=DecisionType.ESCALATE, | |
| risk_level=RiskLevel.HIGH, | |
| action_context=action_context, | |
| reason="Test reason" | |
| ) | |
| assert decision1.get_decision_hash() != decision3.get_decision_hash() | |