Trixy3/trixy_core/events/test_basic.py

#!/usr/bin/env python3
"""
Basic tests for the Trixy Event System.

This file provides basic functionality tests to ensure the event system
works correctly in production scenarios.
"""

import sys
import os
import time
from typing import Any

# Add the source directory to path for imports
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))

from trixy_core.events import (
    EventHandler, TrixyEvent, EventType, EventDataFactory,
    SatelliteInfo, SpeakerInfo, SystemStartupEventData
)


class TestEventHandler:
    """Test component for basic event handling."""
    
    def __init__(self):
        self.events_received = []
    
    @TrixyEvent(["system_startup", "wakeword_received"])
    def handle_test_events(self, event_name: str, event_data: Any):
        """Handle test events."""
        self.events_received.append((event_name, event_data))
        print(f"✓ Received event: {event_name}")


def test_basic_functionality():
    """Test basic event system functionality."""
    print("=== Basic Event System Tests ===\n")
    
    # Test 1: EventHandler creation and configuration
    print("1. Testing EventHandler creation...")
    event_handler = EventHandler(max_history_size=100)
    event_handler.enable_debug_mode(False)  # Disable debug for cleaner output
    print("✓ EventHandler created successfully")
    
    # Test 2: Event handler registration
    print("\n2. Testing event handler registration...")
    test_handler = TestEventHandler()
    event_handler.register_handler_object(test_handler)
    print("✓ Event handlers registered successfully")
    
    # Test 3: Basic event triggering
    print("\n3. Testing basic event triggering...")
    event_id = event_handler.trigger_event(
        EventType.SYSTEM_STARTUP,
        mode="test",
        version="1.0.0",
        config_file="test_config.json"
    )
    print(f"✓ Event triggered with ID: {event_id}")
    
    # Test 4: Event data factory
    print("\n4. Testing event data factory...")
    satellite_info = SatelliteInfo(
        satellite_id="test_sat",
        mac_address="00:11:22:33:44:55",
        room_id="test_room",
        alias="Test Satellite",
        version="1.0.0"
    )
    
    speaker_info = SpeakerInfo(
        speaker_id="test_speaker",
        speaker_name="Test User",
        confidence=0.95
    )
    
    wakeword_data = EventDataFactory.create_event_data(
        EventType.WAKEWORD_RECEIVED,
        wakeword_id="trixy",
        speaker_info=speaker_info,
        satellite_info=satellite_info,
        volume=0.8
    )
    
    event_id2 = event_handler.trigger_event(
        EventType.WAKEWORD_RECEIVED,
        wakeword_data
    )
    print(f"✓ Event created with factory, triggered with ID: {event_id2}")
    
    # Test 5: Event history
    print("\n5. Testing event history...")
    history = event_handler.get_event_history(limit=5)
    print(f"✓ Retrieved {len(history)} events from history")
    
    # Test 6: Event statistics
    print("\n6. Testing event statistics...")
    stats = event_handler.get_event_statistics()
    print(f"✓ Retrieved statistics for {len(stats)} event types")
    
    # Test 7: Async events
    print("\n7. Testing async events...")
    async_event_id = event_handler.trigger_event_async(
        EventType.SYSTEM_STARTUP,
        mode="async_test",
        version="1.0.0",
        config_file="async_test_config.json"
    )
    
    # Wait for async event to complete
    result = event_handler.wait_for_event(async_event_id, timeout=5.0)
    if result:
        print(f"✓ Async event completed successfully in {result.execution_time_ms:.1f}ms")
    else:
        print("✗ Async event timed out")
    
    # Test 8: Verify events were received
    print("\n8. Verifying event reception...")
    expected_events = 3  # 2 system_startup + 1 wakeword_received
    if len(test_handler.events_received) == expected_events:
        print(f"✓ All {expected_events} events received correctly")
    else:
        print(f"✗ Expected {expected_events} events, got {len(test_handler.events_received)}")
    
    # Test 9: Event filtering
    print("\n9. Testing event filtering...")
    event_handler.disable_event_type("system_startup")
    
    # This should be ignored
    ignored_event_id = event_handler.trigger_event(
        EventType.SYSTEM_STARTUP,
        mode="ignored",
        version="1.0.0",
        config_file="ignored.json"
    )
    
    if ignored_event_id == "":
        print("✓ Disabled event type correctly ignored")
    else:
        print("✗ Disabled event type was not ignored")
    
    # Re-enable for cleanup
    event_handler.enable_event_type("system_startup")
    
    # Test 10: Cleanup
    print("\n10. Testing cleanup...")
    event_handler.unregister_handler_object(test_handler)
    event_handler.shutdown()
    print("✓ Event system cleaned up successfully")
    
    print("\n=== All Tests Completed ===")
    return True


def test_event_data_structures():
    """Test event data structures."""
    print("\n=== Event Data Structure Tests ===\n")
    
    # Test all supported event types
    supported_events = EventDataFactory.get_supported_events()
    print(f"Testing {len(supported_events)} supported event types...")
    
    success_count = 0
    for event_type in supported_events:
        try:
            # Create minimal event data for each type
            if event_type == EventType.SATELLITE_CONNECTED:
                satellite_info = SatelliteInfo("test", "00:00:00:00:00:00", "test", "test", "1.0")
                data = EventDataFactory.create_event_data(event_type, satellite_info=satellite_info)
            elif event_type == EventType.WAKEWORD_RECEIVED:
                satellite_info = SatelliteInfo("test", "00:00:00:00:00:00", "test", "test", "1.0")
                speaker_info = SpeakerInfo("test", "test")
                data = EventDataFactory.create_event_data(
                    event_type, 
                    satellite_info=satellite_info, 
                    speaker_info=speaker_info
                )
            else:
                # Use default values for other event types
                data = EventDataFactory.create_event_data(event_type)
            
            # Test serialization
            data_dict = data.to_dict()
            
            print(f"✓ {event_type.value}")
            success_count += 1
            
        except Exception as e:
            print(f"✗ {event_type.value}: {e}")
    
    print(f"\n✓ {success_count}/{len(supported_events)} event data structures tested successfully")
    return success_count == len(supported_events)


def main():
    """Run all tests."""
    print("Trixy Event System - Basic Tests")
    print("=" * 40)
    
    try:
        # Run basic functionality tests
        basic_success = test_basic_functionality()
        
        # Run event data structure tests
        data_success = test_event_data_structures()
        
        if basic_success and data_success:
            print("\n🎉 All tests passed successfully!")
            return 0
        else:
            print("\n❌ Some tests failed!")
            return 1
            
    except Exception as e:
        print(f"\n💥 Test execution failed: {e}")
        import traceback
        traceback.print_exc()
        return 1


if __name__ == "__main__":
    exit(main())