#!/usr/bin/env python3 """ Standalone test runner for trigger_interceptor - no pytest dependencies Manual validation of core logic """ import sys import os sys.path.insert(0, '.') os.environ['ENVIRONMENT'] = 'development' # Import the module from core.trigger_interceptor import ( TriggerInterceptor, TriggerDecision, MaturityLevel, RoutingDecision, ) from unittest.mock import MagicMock, AsyncMock, patch import asyncio def test_maturity_levels(): """Test maturity level determination""" print("Testing maturity level determination...") mock_db = MagicMock() interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") # Test explicit status mapping assert interceptor._determine_maturity_level("student", 0.3) == MaturityLevel.STUDENT assert interceptor._determine_maturity_level("intern", 0.6) == MaturityLevel.INTERN assert interceptor._determine_maturity_level("supervised", 0.8) == MaturityLevel.SUPERVISED assert interceptor._determine_maturity_level("autonomous", 0.95) == MaturityLevel.AUTONOMOUS # Test confidence-based fallback assert interceptor._determine_maturity_level("unknown", 0.4) == MaturityLevel.STUDENT assert interceptor._determine_maturity_level("unknown", 0.55) == MaturityLevel.INTERN assert interceptor._determine_maturity_level("unknown", 0.75) == MaturityLevel.SUPERVISED assert interceptor._determine_maturity_level("unknown", 0.92) == MaturityLevel.AUTONOMOUS # Test boundaries assert interceptor._determine_maturity_level("unknown", 0.5) == MaturityLevel.INTERN assert interceptor._determine_maturity_level("unknown", 0.7) == MaturityLevel.SUPERVISED assert interceptor._determine_maturity_level("unknown", 0.9) == MaturityLevel.AUTONOMOUS print("✓ All maturity level tests passed") def test_trigger_decision(): """Test TriggerDecision creation""" print("Testing TriggerDecision creation...") decision = TriggerDecision( routing_decision=RoutingDecision.EXECUTION, execute=True, agent_id="agent-001", agent_maturity="autonomous", confidence_score=0.95, trigger_source="manual" ) assert decision.execute is True assert decision.agent_id == "agent-001" assert decision.routing_decision == RoutingDecision.EXECUTION assert decision.confidence_score == 0.95 assert decision.blocked_context is None assert decision.proposal is None print("✓ TriggerDecision tests passed") async def test_cache_integration(): """Test cache integration""" print("Testing cache integration...") mock_db = MagicMock() interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") # Test cache hit with patch('core.trigger_interceptor.get_async_governance_cache') as mock_get: mock_cache = AsyncMock() mock_cache.get = AsyncMock(return_value={"status": "autonomous", "confidence": 0.95}) mock_get.return_value = mock_cache status, confidence = await interceptor._get_agent_maturity_cached("agent-001") assert status == "autonomous" assert confidence == 0.95 # DB should NOT be queried on cache hit mock_db.query.assert_not_called() print("✓ Cache integration tests passed") async def test_manual_trigger(): """Test manual trigger handling""" print("Testing manual trigger handling...") mock_db = MagicMock() interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") # Test with STUDENT agent decision = await interceptor._handle_manual_trigger( agent_id="agent-student-001", maturity_level=MaturityLevel.STUDENT, confidence_score=0.3, trigger_context={"action": "test"}, user_id="user-001" ) assert decision.execute is True assert decision.routing_decision == RoutingDecision.EXECUTION assert "STUDENT mode" in decision.reason assert "user-001" in decision.reason # Test with INTERN agent decision = await interceptor._handle_manual_trigger( agent_id="agent-intern-001", maturity_level=MaturityLevel.INTERN, confidence_score=0.6, trigger_context={"action": "test"}, user_id="user-002" ) assert decision.execute is True assert "INTERN learning mode" in decision.reason # Test with SUPERVISED agent decision = await interceptor._handle_manual_trigger( agent_id="agent-supervised-001", maturity_level=MaturityLevel.SUPERVISED, confidence_score=0.8, trigger_context={"action": "test"}, user_id="user-003" ) assert decision.execute is True assert "SUPERVISION mode" in decision.reason # Test with AUTONOMOUS agent decision = await interceptor._handle_manual_trigger( agent_id="agent-autonomous-001", maturity_level=MaturityLevel.AUTONOMOUS, confidence_score=0.95, trigger_context={"action": "test"}, user_id="user-004" ) assert decision.execute is True assert "user-004" in decision.reason # Test with unknown user decision = await interceptor._handle_manual_trigger( agent_id="agent-autonomous-001", maturity_level=MaturityLevel.AUTONOMOUS, confidence_score=0.95, trigger_context={"action": "test"}, user_id=None ) assert decision.execute is True assert "unknown" in decision.reason print("✓ Manual trigger tests passed") async def test_student_agent_routing(): """Test STUDENT agent routing""" print("Testing STUDENT agent routing...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-student-001" mock_agent.name = "Student Agent" mock_agent.status = "student" mock_agent.confidence_score = 0.3 mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") # Mock training service mock_proposal = MagicMock() mock_proposal.id = "proposal-001" interceptor.training_service.create_training_proposal = AsyncMock( return_value=mock_proposal ) # Mock BlockedTriggerContext to avoid SQLAlchemy initialization with patch('core.trigger_interceptor.BlockedTriggerContext') as mock_blocked_ctx_class: mock_blocked_ctx = MagicMock() mock_blocked_ctx_class.return_value = mock_blocked_ctx from core.models import TriggerSource decision = await interceptor._route_student_agent( agent_id="agent-student-001", trigger_source=TriggerSource.WORKFLOW_ENGINE, trigger_context={"action_type": "send_email"}, confidence_score=0.3 ) assert decision.execute is False assert decision.routing_decision == RoutingDecision.TRAINING assert "STUDENT agent blocked" in decision.reason assert decision.proposal.id == "proposal-001" # Verify blocked context was created assert mock_db.add.called assert mock_db.commit.called print("✓ STUDENT agent routing tests passed") async def test_intern_agent_routing(): """Test INTERN agent routing""" print("Testing INTERN agent routing...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-intern-001" mock_agent.name = "Intern Agent" mock_agent.status = "intern" mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") # Mock BlockedTriggerContext with patch('core.trigger_interceptor.BlockedTriggerContext') as mock_blocked_ctx_class: mock_blocked_ctx = MagicMock() mock_blocked_ctx_class.return_value = mock_blocked_ctx from core.models import TriggerSource decision = await interceptor._route_intern_agent( agent_id="agent-intern-001", trigger_source=TriggerSource.DATA_SYNC, trigger_context={"action_type": "create_record"}, confidence_score=0.6 ) assert decision.execute is False assert decision.routing_decision == RoutingDecision.PROPOSAL assert "INTERN agent" in decision.reason assert "proposal" in decision.reason.lower() # Verify blocked context was created assert mock_db.add.called print("✓ INTERN agent routing tests passed") async def test_supervised_agent_routing(): """Test SUPERVISED agent routing""" print("Testing SUPERVISED agent routing...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-supervised-001" mock_agent.name = "Supervised Agent" mock_agent.status = "supervised" mock_agent.user_id = "user-001" mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") from core.models import TriggerSource # Test with available user with patch('core.user_activity_service.UserActivityService') as mock_service_class: mock_service = AsyncMock() mock_service.get_user_state = AsyncMock(return_value="active") mock_service.should_supervise = MagicMock(return_value=True) mock_service_class.return_value = mock_service decision = await interceptor._route_supervised_agent( agent_id="agent-supervised-001", trigger_source=TriggerSource.WORKFLOW_ENGINE, trigger_context={"action_type": "execute_workflow"}, confidence_score=0.8 ) assert decision.execute is True assert decision.routing_decision == RoutingDecision.SUPERVISION assert "real-time monitoring" in decision.reason # Test with unavailable user with patch('core.user_activity_service.UserActivityService') as mock_service_class: mock_service = AsyncMock() mock_service.get_user_state = AsyncMock(return_value="away") mock_service.should_supervise = MagicMock(return_value=False) mock_service_class.return_value = mock_service # Mock queue service with patch('core.supervised_queue_service.SupervisedQueueService') as mock_queue_class: mock_queue = AsyncMock() mock_queue.enqueue_execution = AsyncMock() mock_queue_class.return_value = mock_queue decision = await interceptor._route_supervised_agent( agent_id="agent-supervised-001", trigger_source=TriggerSource.DATA_SYNC, trigger_context={"action_type": "execute_workflow"}, confidence_score=0.75 ) assert decision.execute is False assert "queued" in decision.reason assert "user unavailable" in decision.reason mock_queue.enqueue_execution.assert_called_once() # Test agent not found mock_query.filter.return_value.first.return_value = None decision = await interceptor._route_supervised_agent( agent_id="nonexistent-agent", trigger_source=TriggerSource.WORKFLOW_ENGINE, trigger_context={}, confidence_score=0.8 ) assert decision.execute is False assert decision.routing_decision == RoutingDecision.SUPERVISION assert "not found" in decision.reason print("✓ SUPERVISED agent routing tests passed") async def test_autonomous_agent_execution(): """Test AUTONOMOUS agent execution""" print("Testing AUTONOMOUS agent execution...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-autonomous-001" mock_agent.name = "Autonomous Agent" mock_agent.status = "autonomous" mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") from core.models import TriggerSource decision = await interceptor._allow_execution( agent_id="agent-autonomous-001", trigger_source=TriggerSource.WORKFLOW_ENGINE, trigger_context={"action_type": "critical_operation"}, confidence_score=0.95 ) assert decision.execute is True assert decision.routing_decision == RoutingDecision.EXECUTION assert "AUTONOMOUS agent" in decision.reason assert "approved for full execution" in decision.reason # Note: _allow_execution doesn't validate agent existence (it's called after validation) # The public allow_execution method does validation, but _allow_execution is an optimization # that assumes the agent was already validated upstream print("✓ AUTONOMOUS agent execution tests passed") async def test_intercept_trigger_main(): """Test main intercept_trigger method""" print("Testing main intercept_trigger method...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-001" mock_agent.status = "autonomous" mock_agent.confidence_score = 0.95 mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") from core.models import TriggerSource # Mock cache with patch('core.trigger_interceptor.get_async_governance_cache') as mock_get: mock_cache = AsyncMock() mock_cache.get = AsyncMock(return_value=None) mock_get.return_value = mock_cache # Test MANUAL trigger decision = await interceptor.intercept_trigger( agent_id="agent-001", trigger_source=TriggerSource.MANUAL, trigger_context={"action": "test"}, user_id="user-001" ) assert decision.execute is True assert "Manual trigger" in decision.reason print("✓ Main intercept_trigger tests passed") async def test_route_to_training(): """Test route_to_training method""" print("Testing route_to_training method...") mock_db = MagicMock() interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") mock_blocked_trigger = MagicMock() mock_blocked_trigger.agent_id = "agent-student-001" mock_blocked_trigger.trigger_type = "send_email" mock_proposal = MagicMock() mock_proposal.id = "proposal-training-001" interceptor.training_service.create_training_proposal = AsyncMock( return_value=mock_proposal ) result = await interceptor.route_to_training(mock_blocked_trigger) assert result.id == "proposal-training-001" interceptor.training_service.create_training_proposal.assert_called_once_with( mock_blocked_trigger ) print("✓ route_to_training tests passed") async def test_create_proposal(): """Test create_proposal method""" print("Testing create_proposal method...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-intern-001" mock_agent.name = "Intern Agent" mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") proposed_action = {"action_type": "send_email", "to": "test@example.com"} reasoning = "User requested email" # Mock AgentProposal to avoid SQLAlchemy initialization issues with patch('core.trigger_interceptor.AgentProposal') as mock_proposal_class: mock_proposal = MagicMock() mock_proposal.agent_id = "agent-intern-001" mock_proposal.description = "Agent is proposing an action...send_email..." mock_proposal_class.return_value = mock_proposal result = await interceptor.create_proposal( intern_agent_id="agent-intern-001", trigger_context={"action": "test"}, proposed_action=proposed_action, reasoning=reasoning ) # Verify proposal was created assert result.agent_id == "agent-intern-001" assert "send_email" in result.description # Verify database operations assert mock_db.add.called assert mock_db.commit.called assert mock_db.refresh.called # Test agent not found mock_query.filter.return_value.first.return_value = None try: await interceptor.create_proposal( intern_agent_id="nonexistent-agent", trigger_context={}, proposed_action={}, reasoning="test" ) assert False, "Should have raised ValueError" except ValueError as e: assert "not found" in str(e) print("✓ create_proposal tests passed") async def test_execute_with_supervision(): """Test execute_with_supervision method""" print("Testing execute_with_supervision method...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.id = "agent-supervised-001" mock_agent.name = "Supervised Agent" mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") trigger_context = {"action": "test_action"} # Mock SupervisionSession to avoid SQLAlchemy initialization issues with patch('core.trigger_interceptor.SupervisionSession') as mock_session_class: mock_session = MagicMock() mock_session.agent_id = "agent-supervised-001" mock_session.supervisor_id = "supervisor-001" mock_session_class.return_value = mock_session result = await interceptor.execute_with_supervision( trigger_context=trigger_context, agent_id="agent-supervised-001", user_id="supervisor-001" ) # Verify supervision session was created assert result.agent_id == "agent-supervised-001" assert result.supervisor_id == "supervisor-001" # Verify database operations assert mock_db.add.called assert mock_db.commit.called assert mock_db.refresh.called # Test agent not found mock_query.filter.return_value.first.return_value = None try: await interceptor.execute_with_supervision( trigger_context={}, agent_id="nonexistent-agent", user_id="supervisor-001" ) assert False, "Should have raised ValueError" except ValueError as e: assert "not found" in str(e) print("✓ execute_with_supervision tests passed") async def test_cache_miss(): """Test cache miss path""" print("Testing cache miss path...") mock_db = MagicMock() mock_query = MagicMock() mock_agent = MagicMock() mock_agent.status = "autonomous" mock_agent.confidence_score = 0.95 mock_query.filter.return_value.first.return_value = mock_agent mock_db.query.return_value = mock_query interceptor = TriggerInterceptor(db=mock_db, workspace_id="ws-001") with patch('core.trigger_interceptor.get_async_governance_cache') as mock_get: mock_cache = AsyncMock() mock_cache.get = AsyncMock(return_value=None) # Cache miss mock_cache.set = AsyncMock() mock_get.return_value = mock_cache status, confidence = await interceptor._get_agent_maturity_cached("agent-001") assert status == "autonomous" assert confidence == 0.95 # Verify DB was queried mock_db.query.assert_called_once() # Verify cache was set mock_cache.set.assert_called_once() print("✓ Cache miss tests passed") async def main(): """Run all tests""" print("=" * 60) print("Running Trigger Interceptor Tests (Standalone)") print("=" * 60) try: # Sync tests test_maturity_levels() test_trigger_decision() # Async tests await test_cache_integration() await test_manual_trigger() await test_student_agent_routing() await test_intern_agent_routing() await test_supervised_agent_routing() await test_autonomous_agent_execution() await test_intercept_trigger_main() await test_route_to_training() await test_create_proposal() await test_execute_with_supervision() await test_cache_miss() print("=" * 60) print("✓ ALL TESTS PASSED") print("=" * 60) return 0 except AssertionError as e: print(f"✗ TEST FAILED: {e}") import traceback traceback.print_exc() return 1 except Exception as e: print(f"✗ ERROR: {e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": exit_code = asyncio.run(main()) sys.exit(exit_code)