#!/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())