""" AIVA Voice Command Bridge - Priority Routing System Tests Test Suite for Story 9: Priority routing system """ import pytest import json import time import threading from datetime import datetime, timedelta from unittest.mock import Mock, MagicMock, patch, AsyncMock, call from decimal import Decimal from dataclasses import dataclass, field from typing import Optional, List, Dict, Any from enum import Enum import logging # Import the modules under test import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from priority_router import ( PriorityRouter, PriorityLevel, Directive, DirectiveStatus, get_priority_config, calculate_processing_deadline, determine_queue_position ) from priority_escalation_worker import ( PriorityEscalationWorker, EscalationRule, load_escalation_rules, check_escalation_needed, process_escalations ) from notification_dispatcher import ( NotificationDispatcher, NotificationChannel, NotificationType, NotificationPayload, dispatch_notification, batch_notifications ) # ============================================================================ # FIXTURES # ============================================================================ @pytest.fixture def mock_db_connection(): """Mock PostgreSQL database connection.""" conn = MagicMock() cursor = MagicMock() conn.cursor.return_value = cursor cursor.fetchall.return_value = [] cursor.fetchone.return_value = None return conn, cursor @pytest.fixture def sample_directive(): """Create a sample directive for testing.""" return Directive( id="directive-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Turn on lights", raw_transcript="Turn on the lights now", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) @pytest.fixture def sample_directive_critical(): """Create a CRITICAL priority directive.""" return Directive( id="directive-critical-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Emergency stop", raw_transcript="Emergency stop now", priority=10, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={"urgent_keywords": ["emergency", "now"]} ) @pytest.fixture def sample_directive_low(): """Create a LOW priority directive.""" return Directive( id="directive-low-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Reminder set", raw_transcript="Remind me tomorrow", priority=2, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) @pytest.fixture def priority_router(mock_db_connection): """Create PriorityRouter instance with mocked dependencies.""" conn, cursor = mock_db_connection router = PriorityRouter( db_connection=conn, api_key="test-api-key" ) return router @pytest.fixture def notification_dispatcher(): """Create NotificationDispatcher instance.""" dispatcher = NotificationDispatcher( websocket_url="ws://localhost:8080/ws", enable_terminal_bell=True, enable_desktop_notification=True ) return dispatcher @pytest.fixture def escalation_worker(mock_db_connection): """Create PriorityEscalationWorker instance.""" conn, cursor = mock_db_connection worker = PriorityEscalationWorker( db_connection=conn, check_interval_seconds=60 ) return worker @pytest.fixture def directives_in_queue(): """Create a list of directives at different priority levels.""" now = datetime.utcnow() return [ Directive( id=f"directive-{i:03d}", kinan_id="kinan-123", aiva_id="aiva-456", command=f"Command {i}", raw_transcript=f"Command {i}", priority=p, status=DirectiveStatus.PENDING, created_at=now - timedelta(minutes=m), updated_at=now - timedelta(minutes=m), metadata={} ) for i, p, m in [ (1, 5, 10), (2, 8, 8), (3, 3, 6), (4, 7, 4), (5, 1, 2), (6, 9, 1), (7, 6, 0), ] ] @pytest.fixture def old_pending_directive(): """Create a directive that's been pending for too long.""" return Directive( id="directive-old-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Old command", raw_transcript="Old command", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=16), updated_at=datetime.utcnow() - timedelta(minutes=16), metadata={} ) @pytest.fixture def urgent_keyword_directive(): """Create a directive with urgent keywords.""" return Directive( id="directive-urgent-kw-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Process this now", raw_transcript="Process this now urgent", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={"followup_text": "urgent"} ) # ============================================================================ # PRIORITY ROUTER TESTS # ============================================================================ class TestPriorityLevel: """Tests for PriorityLevel enum and configurations.""" def test_priority_level_critical(self): """Test CRITICAL priority level (10).""" config = get_priority_config(10) assert config["level"] == 10 assert config["name"] == "CRITICAL" assert config["max_queue_wait_seconds"] == 0 assert config["bypass_queue"] is True assert config["notification_channels"] == ["websocket", "bell", "log", "desktop"] def test_priority_level_urgent(self): """Test URGENT priority levels (8-9).""" for priority in [8, 9]: config = get_priority_config(priority) assert config["level"] == priority assert config["name"] == "URGENT" assert config["max_queue_wait_seconds"] == 60 assert config["bypass_queue"] is True assert "websocket" in config["notification_channels"] assert "bell" in config["notification_channels"] def test_priority_level_high(self): """Test HIGH priority levels (6-7).""" for priority in [6, 7]: config = get_priority_config(priority) assert config["level"] == priority assert config["name"] == "HIGH" assert config["max_queue_wait_seconds"] == 300 # 5 minutes def test_priority_level_normal(self): """Test NORMAL priority levels (4-5).""" for priority in [4, 5]: config = get_priority_config(priority) assert config["level"] == priority assert config["name"] == "NORMAL" assert config["max_queue_wait_seconds"] is None def test_priority_level_low(self): """Test LOW priority levels (1-3).""" for priority in [1, 2, 3]: config = get_priority_config(priority) assert config["level"] == priority assert config["name"] == "LOW" assert config["batch_enabled"] is True assert config["notification_channels"] == ["log"] def test_priority_level_invalid(self): """Test invalid priority level.""" with pytest.raises(ValueError): get_priority_config(0) with pytest.raises(ValueError): get_priority_config(11) class TestPriorityRouter: """Tests for PriorityRouter main class.""" def test_router_initialization(self, priority_router): """Test router initializes correctly.""" assert priority_router is not None assert priority_router.api_key == "test-api-key" def test_route_directive_critical(self, priority_router, sample_directive_critical, mock_db_connection): """Test routing a CRITICAL priority directive.""" conn, cursor = mock_db_connection with patch.object(priority_router, '_save_directive', return_value=True): result = priority_router.route_directive(sample_directive_critical) assert result["priority"] == 10 assert result["should_bypass_queue"] is True assert result["notification_channels"] == ["websocket", "bell", "log", "desktop"] def test_route_directive_urgent(self, priority_router, mock_db_connection): """Test routing an URGENT priority directive.""" directive = Directive( id="directive-urgent-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Test urgent", raw_transcript="This is urgent", priority=9, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) result = priority_router.route_directive(directive) assert result["priority"] == 9 assert result["should_bypass_queue"] is True assert "websocket" in result["notification_channels"] def test_route_directive_normal(self, priority_router, sample_directive): """Test routing a NORMAL priority directive.""" result = priority_router.route_directive(sample_directive) assert result["priority"] == 5 assert result["should_bypass_queue"] is False assert "websocket" in result["notification_channels"] def test_route_directive_low(self, priority_router, sample_directive_low): """Test routing a LOW priority directive.""" result = priority_router.route_directive(sample_directive_low) assert result["priority"] == 2 assert result["should_bypass_queue"] is False assert result["batch_enabled"] is True def test_calculate_processing_deadline_critical(self): """Test deadline calculation for CRITICAL priority.""" created_at = datetime(2024, 1, 1, 12, 0, 0) deadline = calculate_processing_deadline(10, created_at) assert deadline is not None assert deadline <= created_at # Should be immediate or before def test_calculate_processing_deadline_urgent(self): """Test deadline calculation for URGENT priority.""" created_at = datetime(2024, 1, 1, 12, 0, 0) deadline = calculate_processing_deadline(9, created_at) assert deadline is not None expected = created_at + timedelta(seconds=60) assert deadline <= expected def test_calculate_processing_deadline_high(self): """Test deadline calculation for HIGH priority.""" created_at = datetime(2024, 1, 1, 12, 0, 0) deadline = calculate_processing_deadline(7, created_at) assert deadline is not None expected = created_at + timedelta(seconds=300) # 5 minutes assert deadline <= expected def test_calculate_processing_deadline_normal_no_deadline(self): """Test deadline calculation for NORMAL priority (no deadline).""" created_at = datetime(2024, 1, 1, 12, 0, 0) deadline = calculate_processing_deadline(5, created_at) assert deadline is None def test_determine_queue_position_critical_first(self, directives_in_queue): """Test CRITICAL priority goes to front of queue.""" critical = Directive( id="directive-critical-new", kinan_id="kinan-123", aiva_id="aiva-456", command="Critical", raw_transcript="Critical", priority=10, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) position = determine_queue_position(critical, directives_in_queue) assert position == 0 def test_determine_queue_position_urgent_front(self, directives_in_queue): """Test URGENT priority goes near front of queue.""" urgent = Directive( id="directive-urgent-new", kinan_id="kinan-123", aiva_id="aiva-456", command="Urgent", raw_transcript="Urgent", priority=9, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) position = determine_queue_position(urgent, directives_in_queue) assert position == 0 def test_determine_queue_position_low_back(self, directives_in_queue): """Test LOW priority goes to back of queue.""" low = Directive( id="directive-low-new", kinan_id="kinan-123", aiva_id="aiva-456", command="Low", raw_transcript="Low", priority=1, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) position = determine_queue_position(low, directives_in_queue) assert position == len(directives_in_queue) def test_determine_queue_position_empty_queue(self, sample_directive): """Test queue position with empty queue.""" position = determine_queue_position(sample_directive, []) assert position == 0 # ============================================================================ # PRIORITY ESCALATION TESTS # ============================================================================ class TestEscalationRules: """Tests for escalation rules.""" def test_escalation_rule_5_minutes_pending(self, mock_db_connection): """Test escalation after 5 minutes pending for priority < 8.""" conn, cursor = mock_db_connection # Directive with priority 5, pending for 6 minutes directive = Directive( id="test-esc-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=6), updated_at=datetime.utcnow() - timedelta(minutes=6), metadata={} ) result = check_escalation_needed(directive) assert result["should_escalate"] is True assert result["new_priority"] == 6 def test_escalation_rule_15_minutes_pending(self, mock_db_connection): """Test escalation to URGENT after 15 minutes pending.""" directive = Directive( id="test-esc-002", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=16), updated_at=datetime.utcnow() - timedelta(minutes=16), metadata={} ) result = check_escalation_needed(directive) assert result["should_escalate"] is True assert result["new_priority"] == 8 def test_escalation_rule_urgent_keywords(self, urgent_keyword_directive): """Test escalation with urgent keywords in follow-up.""" result = check_escalation_needed(urgent_keyword_directive) assert result["should_escalate"] is True assert result["new_priority"] == 10 def test_escalation_rule_no_need_when_critical(self): """Test no escalation needed when already CRITICAL.""" directive = Directive( id="test-esc-003", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=10, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=20), updated_at=datetime.utcnow() - timedelta(minutes=20), metadata={} ) result = check_escalation_needed(directive) assert result["should_escalate"] is False def test_escalation_rule_no_need_recent(self, sample_directive): """Test no escalation needed when recently created.""" # Directive created 1 minute ago sample_directive.created_at = datetime.utcnow() - timedelta(minutes=1) result = check_escalation_needed(sample_directive) assert result["should_escalate"] is False def test_escalation_rule_no_need_high_priority(self): """Test no escalation needed when priority >= 8.""" directive = Directive( id="test-esc-004", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=8, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=10), updated_at=datetime.utcnow() - timedelta(minutes=10), metadata={} ) result = check_escalation_needed(directive) assert result["should_escalate"] is False def test_escalation_rule_max_priority_10(self): """Test escalation doesn't exceed priority 10.""" directive = Directive( id="test-esc-005", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=9, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=16), updated_at=datetime.utcnow() - timedelta(minutes=16), metadata={} ) result = check_escalation_needed(directive) # Should escalate to 10, but not beyond assert result["new_priority"] == 10 class TestPriorityEscalationWorker: """Tests for PriorityEscalationWorker.""" def test_worker_initialization(self, escalation_worker): """Test worker initializes correctly.""" assert escalation_worker is not None assert escalation_worker.check_interval_seconds == 60 def test_load_escalation_rules(self): """Test loading escalation rules.""" rules = load_escalation_rules() assert len(rules) > 0 # Check that rules are sorted by priority priorities = [r.min_priority for r in rules] assert priorities == sorted(priorities) def test_process_escalations_empty(self, escalation_worker, mock_db_connection): """Test processing with no pending directives.""" conn, cursor = mock_db_connection cursor.fetchall.return_value = [] result = process_escalations(conn) assert result["escalated_count"] == 0 def test_process_escalations_with_directives(self, escalation_worker, mock_db_connection): """Test processing with pending directives that need escalation.""" conn, cursor = mock_db_connection # Mock directives that need escalation old_pending = Directive( id="test-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=16), updated_at=datetime.utcnow() - timedelta(minutes=16), metadata={} ) cursor.fetchall.return_value = [ (old_pending.id, old_pending.kinan_id, old_pending.aiva_id, old_pending.command, old_pending.raw_transcript, old_pending.priority, old_pending.status.value, old_pending.created_at.isoformat(), old_pending.updated_at.isoformat(), '{}') ] with patch('priority_escalation_worker.check_escalation_needed', return_value={"should_escalate": True, "new_priority": 8}): with patch.object(escalation_worker, '_update_directive_priority', return_value=True): result = process_escalations(conn) assert result["escalated_count"] >= 0 def test_worker_stop(self, escalation_worker): """Test worker can be stopped.""" escalation_worker.start() assert escalation_worker._running is True escalation_worker.stop() assert escalation_worker._running is False def test_worker_run_cycle(self, escalation_worker, mock_db_connection): """Test worker runs a cycle correctly.""" conn, cursor = mock_db_connection cursor.fetchall.return_value = [] escalation_worker._run_cycle() cursor.execute.assert_called() # ============================================================================ # NOTIFICATION DISPATCHER TESTS # ============================================================================ class TestNotificationDispatcher: """Tests for NotificationDispatcher.""" def test_dispatcher_initialization(self, notification_dispatcher): """Test dispatcher initializes correctly.""" assert notification_dispatcher.websocket_url == "ws://localhost:8080/ws" assert notification_dispatcher.enable_terminal_bell is True assert notification_dispatcher.enable_desktop_notification is True def test_dispatch_notification_critical(self, notification_dispatcher): """Test dispatching CRITICAL notification.""" payload = NotificationPayload( directive_id="test-001", priority=10, message="Critical: Emergency stop", channels=[ NotificationChannel.WEBSOCKET, NotificationChannel.TERMINAL_BELL, NotificationChannel.DESKTOP, NotificationChannel.LOG ] ) with patch.object(notification_dispatcher, '_send_websocket', new_callable=AsyncMock): with patch.object(notification_dispatcher, '_send_terminal_bell'): with patch.object(notification_dispatcher, '_send_desktop_notification'): with patch.object(notification_dispatcher, '_send_log'): # Run sync import asyncio asyncio.run(notification_dispatcher.dispatch(payload)) # If we get here without exception, test passes def test_dispatch_notification_urgent(self, notification_dispatcher): """Test dispatching URGENT notification.""" payload = NotificationPayload( directive_id="test-002", priority=9, message="Urgent: Process now", channels=[ NotificationChannel.WEBSOCKET, NotificationChannel.TERMINAL_BELL, NotificationChannel.LOG ] ) with patch.object(notification_dispatcher, '_send_websocket', new_callable=AsyncMock): with patch.object(notification_dispatcher, '_send_terminal_bell'): with patch.object(notification_dispatcher, '_send_log'): import asyncio asyncio.run(notification_dispatcher.dispatch(payload)) def test_dispatch_notification_normal(self, notification_dispatcher): """Test dispatching NORMAL notification.""" payload = NotificationPayload( directive_id="test-003", priority=5, message="Normal: Standard processing", channels=[NotificationChannel.WEBSOCKET, NotificationChannel.LOG] ) with patch.object(notification_dispatcher, '_send_websocket', new_callable=AsyncMock): with patch.object(notification_dispatcher, '_send_log'): import asyncio asyncio.run(notification_dispatcher.dispatch(payload)) def test_dispatch_notification_low(self, notification_dispatcher): """Test dispatching LOW notification (log only).""" payload = NotificationPayload( directive_id="test-004", priority=2, message="Low: Background processing", channels=[NotificationChannel.LOG] ) with patch.object(notification_dispatcher, '_send_log'): import asyncio asyncio.run(notification_dispatcher.dispatch(payload)) def test_dispatch_notification_channels_disabled(self, notification_dispatcher): """Test dispatching when some channels are disabled.""" notification_dispatcher.enable_terminal_bell = False notification_dispatcher.enable_desktop_notification = False payload = NotificationPayload( directive_id="test-005", priority=10, message="Critical test", channels=[ NotificationChannel.WEBSOCKET, NotificationChannel.TERMINAL_BELL, NotificationChannel.DESKTOP, NotificationChannel.LOG ] ) with patch.object(notification_dispatcher, '_send_websocket', new_callable=AsyncMock): with patch.object(notification_dispatcher, '_send_terminal_bell') as mock_bell: with patch.object(notification_dispatcher, '_send_desktop_notification') as mock_desktop: with patch.object(notification_dispatcher, '_send_log'): import asyncio asyncio.run(notification_dispatcher.dispatch(payload)) # Terminal bell and desktop should not be called mock_bell.assert_not_called() mock_desktop.assert_not_called() def test_batch_notifications(self, notification_dispatcher): """Test batching low priority notifications.""" payloads = [ NotificationPayload( directive_id=f"test-{i:03d}", priority=2, message=f"Low priority {i}", channels=[NotificationChannel.LOG] ) for i in range(5) ] with patch.object(notification_dispatcher, '_send_log') as mock_log: import asyncio asyncio.run(batch_notifications(payloads, notification_dispatcher)) # Should be called once with batched notifications assert mock_log.call_count >= 1 class TestNotificationChannels: """Tests for individual notification channels.""" @pytest.mark.asyncio async def test_send_websocket(self, notification_dispatcher): """Test WebSocket notification sending.""" with patch('notification_dispatcher.websockets.connect', new_callable=AsyncMock) as mock_ws: mock_connection = AsyncMock() mock_ws.return_value = mock_connection await notification_dispatcher._send_websocket( "test-message", {"directive_id": "test-001"} ) def test_send_terminal_bell(self, notification_dispatcher): """Test terminal bell notification.""" with patch('notification_dispatcher.subprocess.run') as mock_subprocess: notification_dispatcher._send_terminal_bell() # Should call subprocess to play bell mock_subprocess.assert_called() def test_send_desktop_notification(self, notification_dispatcher): """Test desktop notification.""" with patch('notification_dispatcher.subprocess.run') as mock_subprocess: notification_dispatcher._send_desktop_notification( "Test Title", "Test Message" ) def test_send_log(self, notification_dispatcher): """Test log notification.""" # Should not raise any exceptions notification_dispatcher._send_log("Test message", {"key": "value"}) class TestDispatchNotificationFunction: """Tests for the dispatch_notification helper function.""" @pytest.mark.asyncio async def test_dispatch_notification_helper(self): """Test the dispatch_notification helper function.""" dispatcher = NotificationDispatcher() with patch.object(dispatcher, 'dispatch', new_callable=AsyncMock) as mock_dispatch: await dispatch_notification( dispatcher=dispatcher, directive_id="test-001", priority=10, message="Test" ) mock_dispatch.assert_called_once() # ============================================================================ # INTEGRATION TESTS # ============================================================================ class TestPriorityRoutingIntegration: """Integration tests for the complete priority routing system.""" def test_full_routing_flow_critical(self, mock_db_connection): """Test complete flow from directive creation to notification for CRITICAL.""" conn, cursor = mock_db_connection router = PriorityRouter(db_connection=conn, api_key="test-key") dispatcher = NotificationDispatcher() # Create critical directive directive = Directive( id="integration-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Emergency", raw_transcript="Emergency stop now", priority=10, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={"urgent_keywords": ["emergency", "now"]} ) # Route the directive routing_result = router.route_directive(directive) # Verify routing assert routing_result["should_bypass_queue"] is True assert routing_result["notification_channels"] == ["websocket", "bell", "log", "desktop"] def test_full_routing_flow_normal(self, mock_db_connection): """Test complete flow for NORMAL priority.""" conn, cursor = mock_db_connection router = PriorityRouter(db_connection=conn, api_key="test-key") directive = Directive( id="integration-002", kinan_id="kinan-123", aiva_id="aiva-456", command="Turn on lights", raw_transcript="Turn on the lights please", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) routing_result = router.route_directive(directive) assert routing_result["should_bypass_queue"] is False assert routing_result["priority"] == 5 def test_escalation_integration(self, mock_db_connection): """Test escalation integration.""" conn, cursor = mock_db_connection worker = PriorityEscalationWorker(db_connection=conn, check_interval_seconds=60) # Directive pending for 16 minutes old_directive = Directive( id="integration-003", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=16), updated_at=datetime.utcnow() - timedelta(minutes=16), metadata={} ) # Check if escalation needed result = check_escalation_needed(old_directive) assert result["should_escalate"] is True assert result["new_priority"] == 8 def test_priority_config_completeness(self): """Test all priority levels have complete configurations.""" for priority in range(1, 11): config = get_priority_config(priority) assert "level" in config assert "name" in config assert "notification_channels" in config if priority >= 8: assert config["bypass_queue"] is True else: # Lower priorities may or may not bypass queue pass # ============================================================================ # EDGE CASE TESTS # ============================================================================ class TestEdgeCases: """Edge case tests for the priority routing system.""" def test_directive_with_no_metadata(self, priority_router): """Test directive with empty metadata.""" directive = Directive( id="edge-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata=None ) result = priority_router.route_directive(directive) assert result is not None def test_directive_priority_boundary_low(self, priority_router): """Test minimum valid priority (1).""" directive = Directive( id="edge-002", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=1, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) result = priority_router.route_directive(directive) assert result["priority"] == 1 def test_directive_priority_boundary_high(self, priority_router): """Test maximum valid priority (10).""" directive = Directive( id="edge-003", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=10, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) result = priority_router.route_directive(directive) assert result["priority"] == 10 assert result["should_bypass_queue"] is True def test_very_old_directive(self, priority_router): """Test directive pending for very long time.""" directive = Directive( id="edge-004", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=3, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(hours=2), updated_at=datetime.utcnow() - timedelta(hours=2), metadata={} ) result = check_escalation_needed(directive) # Should escalate to at least 8 (URGENT) assert result["should_escalate"] is True assert result["new_priority"] >= 8 def test_mixed_case_urgent_keywords(self, priority_router): """Test urgent keywords in different cases.""" directive = Directive( id="edge-005", kinan_id="kinan-123", aiva_id="aiva-456", command="URGENT", raw_transcript="This is URGENT and needs to be done NOW", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={"followup_text": "URGENT NOW"} ) result = check_escalation_needed(directive) assert result["should_escalate"] is True def test_empty_queue(self, priority_router): """Test queue position with empty queue.""" directive = Directive( id="edge-006", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) position = determine_queue_position(directive, []) assert position == 0 def test_all_same_priority(self, priority_router): """Test queue position when all have same priority.""" directives = [ Directive( id=f"edge-007-{i}", kinan_id="kinan-123", aiva_id="aiva-456", command=f"Test {i}", raw_transcript=f"Test {i}", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=i), updated_at=datetime.utcnow() - timedelta(minutes=i), metadata={} ) for i in range(5) ] new_directive = Directive( id="edge-007-new", kinan_id="kinan-123", aiva_id="aiva-456", command="New", raw_transcript="New", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) position = determine_queue_position(new_directive, directives) # Should go to end since same priority and newer assert position == len(directives) def test_notification_with_special_characters(self, notification_dispatcher): """Test notification with special characters in message.""" payload = NotificationPayload( directive_id="edge-008", priority=10, message="Critical: Emergency - ", channels=[NotificationChannel.LOG] ) # Should not raise and should handle properly with patch.object(notification_dispatcher, '_send_log') as mock_log: import asyncio asyncio.run(notification_dispatcher.dispatch(payload)) # Verify log was called (sanitization happens internally) mock_log.assert_called() def test_worker_graceful_shutdown(self, escalation_worker): """Test worker shuts down gracefully.""" escalation_worker.start() time.sleep(0.1) # Let it start escalation_worker.stop() assert escalation_worker._running is False def test_deadline_calculation_all_priorities(self): """Test deadline calculation for all priority levels.""" created_at = datetime(2024, 1, 1, 12, 0, 0) # Critical (10): Immediate deadline = calculate_processing_deadline(10, created_at) assert deadline is not None # Urgent (8-9): Within 60 seconds for p in [8, 9]: deadline = calculate_processing_deadline(p, created_at) assert deadline <= created_at + timedelta(seconds=60) # High (6-7): Within 5 minutes for p in [6, 7]: deadline = calculate_processing_deadline(p, created_at) assert deadline <= created_at + timedelta(seconds=300) # Normal (4-5): No specific deadline for p in [4, 5]: deadline = calculate_processing_deadline(p, created_at) assert deadline is None # Low (1-3): No specific deadline for p in [1, 2, 3]: deadline = calculate_processing_deadline(p, created_at) assert deadline is None # ============================================================================ # ERROR HANDLING TESTS # ============================================================================ class TestErrorHandling: """Error handling and robustness tests.""" def test_router_handles_db_error(self, mock_db_connection): """Test router handles database errors gracefully.""" conn, cursor = mock_db_connection cursor.execute.side_effect = Exception("Database connection lost") router = PriorityRouter(db_connection=conn, api_key="test-key") directive = Directive( id="error-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) # Should handle error and return fallback result = router.route_directive(directive) assert result is not None assert "priority" in result def test_worker_handles_empty_results(self, escalation_worker, mock_db_connection): """Test worker handles empty query results.""" conn, cursor = mock_db_connection cursor.fetchall.return_value = [] result = process_escalations(conn) assert result["escalated_count"] == 0 def test_dispatcher_handles_websocket_error(self, notification_dispatcher): """Test dispatcher handles WebSocket errors.""" payload = NotificationPayload( directive_id="error-002", priority=10, message="Test", channels=[NotificationChannel.WEBSOCKET] ) async def mock_ws_error(*args, **kwargs): raise Exception("WebSocket connection failed") with patch.object(notification_dispatcher, '_send_websocket', side_effect=mock_ws_error): import asyncio # Should not raise - should handle gracefully try: asyncio.run(notification_dispatcher.dispatch(payload)) except Exception: pytest.fail("Dispatcher should handle WebSocket errors gracefully") def test_invalid_priority_in_directive(self, priority_router): """Test handling of invalid priority in directive.""" directive = Directive( id="error-003", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=100, # Invalid status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) # Should either raise or return default with pytest.raises(ValueError): priority_router.route_directive(directive) # ============================================================================ # DATABASE SCHEMA TESTS # ============================================================================ class TestDatabaseSchema: """Tests for database schema and operations.""" def test_directive_table_schema(self, mock_db_connection): """Test directive table has required columns.""" conn, cursor = mock_db_connection # This tests that we're using the correct schema cursor.fetchall.return_value = [ (1, "directive-001", "kinan-123", "aiva-456", "Test", "Test", 5, "pending", datetime.utcnow().isoformat(), datetime.utcnow().isoformat(), '{}') ] cursor.execute("SELECT id, priority, status FROM genesis_bridge.directives") result = cursor.fetchall() assert len(result) > 0 # ============================================================================ # PERFORMANCE TESTS # ============================================================================ class TestPerformance: """Performance-related tests.""" def test_queue_position_calculation_performance(self, directives_in_queue): """Test queue position calculation is efficient.""" new_directive = Directive( id="perf-001", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow(), updated_at=datetime.utcnow(), metadata={} ) # Should complete quickly even with large queue start = time.time() for _ in range(1000): determine_queue_position(new_directive, directives_in_queue) elapsed = time.time() - start assert elapsed < 1.0 # Should be very fast def test_escalation_check_performance(self): """Test escalation check is efficient.""" directive = Directive( id="perf-002", kinan_id="kinan-123", aiva_id="aiva-456", command="Test", raw_transcript="Test", priority=5, status=DirectiveStatus.PENDING, created_at=datetime.utcnow() - timedelta(minutes=16), updated_at=datetime.utcnow() - timedelta(minutes=16), metadata={} ) start = time.time() for _ in range(10000): check_escalation_needed(directive) elapsed = time.time() - start assert elapsed < 1.0 # ============================================================================ # MOCK WEBSOCKET SERVER FOR TESTING # ============================================================================ class MockWebSocketServer: """Mock WebSocket server for testing.""" def __init__(self): self.messages = [] self.connections = [] async def connect(self): self.connections.append(True) return self async def send(self, message): self.messages.append(message) async def close(self): self.connections.pop() # ============================================================================ # MAIN TEST RUNNER CONFIGURATION # ============================================================================ if __name__ == "__main__": pytest.main([__file__, "-v", "--tb=short"])