#!/usr/bin/env python3 """ Comprehensive Test for Satellite Management System This test script demonstrates all the features of the Trixy satellite management system as specified in CLAUDE.md, including advanced access patterns, bulk operations, registration management, and blacklist functionality. """ import sys import os import tempfile import shutil from pathlib import Path from datetime import datetime, timezone # Add the source directory to the Python path - run this from project root sys.path.insert(0, '.') from trixy_core.satellites import ( SatelliteManager, Satellite, SatelliteInfo, SatelliteCapability, AudioPortInfo, RegistrationManager, BlacklistManager, QueryParser, normalize_mac_address ) from trixy_core.satellites.satellite import create_satellite_info def pprint_test(message: str) -> None: """Test logging function.""" print(f"[TEST] {message}") class MockApplication: """Mock application container for testing.""" def __init__(self): self._event_handler = None self._network_manager = None def get_event_handler(self): return self._event_handler def get_network_manager(self): return self._network_manager def set_event_handler(self, handler): self._event_handler = handler def set_network_manager(self, manager): self._network_manager = manager def create_test_satellite_info( mac_address: str, room_id: str, alias_name: str, capabilities: list = None ) -> SatelliteInfo: """Create test satellite info.""" return create_satellite_info( satellite_id=f"sat_{mac_address.replace(':', '_')}", mac_address=normalize_mac_address(mac_address), room_id=room_id, alias_name=alias_name, ip_address=f"192.168.1.{hash(mac_address) % 200 + 50}", version="1.0.0", capabilities=capabilities or [ SatelliteCapability.WAKEWORD_DETECTION, SatelliteCapability.AUDIO_INPUT, SatelliteCapability.AUDIO_OUTPUT, SatelliteCapability.TEXT_TO_SPEECH ] ) def test_basic_satellite_manager(): """Test basic satellite manager functionality.""" pprint_test("=== Testing Basic Satellite Manager ===") # Create temporary directories for testing temp_dir = tempfile.mkdtemp() try: app = MockApplication() manager = SatelliteManager( app, registration_dir=os.path.join(temp_dir, "registrations"), blacklist_file=os.path.join(temp_dir, "blacklist.json"), max_satellites=10, enable_auto_registration=True ) pprint_test(f"✓ Created satellite manager: {manager}") # Test adding satellites sat1_info = create_test_satellite_info( "aa:bb:cc:dd:ee:01", "kitchen", "Kitchen Assistant" ) sat2_info = create_test_satellite_info( "aa:bb:cc:dd:ee:02", "living_room", "Living Room Speaker" ) sat3_info = create_test_satellite_info( "aa:bb:cc:dd:ee:03", "bedroom", "Bedroom Helper" ) # Add satellites sat1 = manager.add_satellite(sat1_info) sat2 = manager.add_satellite(sat2_info) sat3 = manager.add_satellite(sat3_info) pprint_test(f"✓ Added 3 satellites, total: {len(manager)}") # Test basic properties assert len(manager) == 3 assert sat1_info.mac_address in manager assert sat1 in manager pprint_test("✓ Basic satellite manager tests passed") return manager, [sat1, sat2, sat3] finally: shutil.rmtree(temp_dir, ignore_errors=True) def test_advanced_access_patterns(manager, satellites): """Test advanced access patterns as specified in CLAUDE.md.""" pprint_test("=== Testing Advanced Access Patterns ===") sat1, sat2, sat3 = satellites # Test direct index access: satellite_manager[0] pprint_test("Testing direct index access...") first_satellite = manager[0] assert isinstance(first_satellite, Satellite) pprint_test(f"✓ manager[0] = {first_satellite.alias_name}") # Test index access for all satellites for i in range(len(manager)): satellite = manager[i] pprint_test(f"✓ manager[{i}] = {satellite.alias_name}") # Simulate connections for query testing sat1.set_connected("192.168.1.101", AudioPortInfo(2102, 2103, 2104, 2101)) sat2.set_connected("192.168.1.102", AudioPortInfo(2102, 2103, 2104, 2101)) # sat3 remains disconnected (registered status) # Test query-based access: satellite_manager["status=connected"] pprint_test("Testing query-based access...") # Test status queries connected_satellites = manager["status=connected"] assert isinstance(connected_satellites, list) assert len(connected_satellites) == 2 pprint_test(f"✓ Found {len(connected_satellites)} connected satellites") not_connected_satellites = manager["status=registered"] assert len(not_connected_satellites) == 1 pprint_test(f"✓ Found {len(not_connected_satellites)} registered (not connected) satellites") # Test room queries kitchen_satellites = manager["room=kitchen"] assert len(kitchen_satellites) == 1 assert kitchen_satellites[0].alias_name == "Kitchen Assistant" pprint_test(f"✓ Found {len(kitchen_satellites)} satellites in kitchen") # Test combined queries: satellite_manager["status=connected,room=kitchen"] connected_kitchen = manager["status=connected,room=kitchen"] assert len(connected_kitchen) == 1 pprint_test(f"✓ Found {len(connected_kitchen)} connected satellites in kitchen") # Test capability queries wakeword_satellites = manager["capability=wakeword_detection"] assert len(wakeword_satellites) == 3 # All have wakeword detection pprint_test(f"✓ Found {len(wakeword_satellites)} satellites with wakeword detection") # Test alias queries kitchen_by_alias = manager["alias contains Kitchen"] assert len(kitchen_by_alias) == 1 pprint_test(f"✓ Found {len(kitchen_by_alias)} satellites with 'Kitchen' in alias") # Test IP address queries local_satellites = manager["ip starts_with 192.168"] assert len(local_satellites) >= 2 pprint_test(f"✓ Found {len(local_satellites)} satellites on local network") pprint_test("✓ Advanced access pattern tests passed") def test_bulk_operations(manager, satellites): """Test bulk operations functionality.""" pprint_test("=== Testing Bulk Operations ===") sat1, sat2, sat3 = satellites # Test bulk TTS: say_all() pprint_test("Testing bulk TTS operations...") sent_count = manager.say_all("Hello everyone!") pprint_test(f"✓ Sent TTS to {sent_count} satellites (all connected)") # Test filtered bulk TTS sent_count = manager.say_all("Kitchen announcement", "room=kitchen") pprint_test(f"✓ Sent TTS to {sent_count} satellites in kitchen") # Test bulk capability management pprint_test("Testing bulk capability operations...") updated_count = manager.update_all_capabilities( SatelliteCapability.LED_CONTROL, add=True, query="status=connected" ) pprint_test(f"✓ Added LED control to {updated_count} connected satellites") # Verify capability was added led_satellites = manager["capability=led_control"] assert len(led_satellites) == 2 # Only connected satellites got the capability pprint_test(f"✓ Verified {len(led_satellites)} satellites now have LED control") # Test bulk disconnect: disconnect_all() pprint_test("Testing bulk disconnect operations...") disconnected_count = manager.disconnect_all("room=living_room", "bulk_test") pprint_test(f"✓ Disconnected {disconnected_count} satellites in living room") # Test bulk reconnect: reconnect_all() pprint_test("Testing bulk reconnect operations...") # First simulate that satellites can reconnect for satellite in manager["room=living_room"]: satellite._status = satellite._status # Keep current status for reconnect attempt reconnected_count = manager.reconnect_all("room=living_room") pprint_test(f"✓ Attempted to reconnect {reconnected_count} satellites in living room") pprint_test("✓ Bulk operations tests passed") def test_registration_system(): """Test satellite registration and blacklist management.""" pprint_test("=== Testing Registration System ===") # Create temporary directory temp_dir = tempfile.mkdtemp() try: reg_manager = RegistrationManager( registration_dir=os.path.join(temp_dir, "registrations"), backup_dir=os.path.join(temp_dir, "backups") ) blacklist_manager = BlacklistManager( blacklist_file=os.path.join(temp_dir, "blacklist.json") ) # Test registration creation pprint_test("Testing satellite registration...") sat_info = create_test_satellite_info( "bb:cc:dd:ee:ff:01", "office", "Office Assistant" ) registration = reg_manager.create_registration( sat_info, registered_by="test_user", notes="Test registration", auto_approve=True ) assert registration.status.value == "approved" pprint_test("✓ Created and approved registration") # Test registration retrieval retrieved = reg_manager.get_registration(sat_info.mac_address) assert retrieved is not None assert retrieved.satellite_info.alias_name == "Office Assistant" pprint_test("✓ Retrieved registration successfully") # Test blacklist management pprint_test("Testing blacklist management...") # Add to blacklist success = blacklist_manager.add_to_blacklist( "cc:dd:ee:ff:aa:01", reason="test_blacklist", blacklisted_by="test_user", notes="Testing blacklist functionality" ) assert success pprint_test("✓ Added MAC to blacklist") # Check if blacklisted is_blacklisted = blacklist_manager.is_blacklisted("cc:dd:ee:ff:aa:01") assert is_blacklisted pprint_test("✓ Confirmed MAC is blacklisted") # Get blacklist stats stats = blacklist_manager.get_blacklist_stats() assert stats["total_entries"] == 1 assert stats["active_entries"] == 1 pprint_test(f"✓ Blacklist stats: {stats['total_entries']} total, {stats['active_entries']} active") # Remove from blacklist success = blacklist_manager.remove_from_blacklist("cc:dd:ee:ff:aa:01") assert success is_blacklisted = blacklist_manager.is_blacklisted("cc:dd:ee:ff:aa:01") assert not is_blacklisted pprint_test("✓ Removed MAC from blacklist") pprint_test("✓ Registration system tests passed") finally: shutil.rmtree(temp_dir, ignore_errors=True) def test_connection_management(manager, satellites): """Test satellite connection management.""" pprint_test("=== Testing Connection Management ===") # Test registration mode pprint_test("Testing registration mode...") assert not manager.is_registration_mode_active() success = manager.enter_registration_mode(timeout=5.0) assert success assert manager.is_registration_mode_active() pprint_test("✓ Entered registration mode") success = manager.exit_registration_mode() assert success assert not manager.is_registration_mode_active() pprint_test("✓ Exited registration mode") # Test satellite connection handling pprint_test("Testing satellite connection handling...") success, message, satellite = manager.handle_satellite_connection( mac_address="dd:ee:ff:aa:bb:01", ip_address="192.168.1.200", room_id="garage", alias_name="Garage Speaker", capabilities=["wakeword_detection", "audio_output"] ) # This might fail if not in registration mode, which is expected pprint_test(f"✓ Connection attempt result: {success} - {message}") # Test with registration mode manager.enter_registration_mode(10.0) success, message, satellite = manager.handle_satellite_connection( mac_address="dd:ee:ff:aa:bb:02", ip_address="192.168.1.201", room_id="garage", alias_name="Garage Light Controller", capabilities=["led_control", "button_input"] ) if success: assert satellite is not None assert satellite.room_id == "garage" pprint_test("✓ Successfully connected new satellite") manager.exit_registration_mode() pprint_test("✓ Connection management tests passed") def test_statistics_and_monitoring(manager, satellites): """Test statistics and monitoring functionality.""" pprint_test("=== Testing Statistics and Monitoring ===") # Get comprehensive stats stats = manager.get_stats() pprint_test(f"✓ Total satellites: {stats.total_satellites}") pprint_test(f"✓ Connected satellites: {stats.connected_satellites}") pprint_test(f"✓ Registered satellites: {stats.registered_satellites}") pprint_test(f"✓ Uptime: {stats.uptime_seconds:.1f} seconds") # Get full status status = manager.get_status() pprint_test(f"✓ Statistics: {len(status['statistics'])} fields") pprint_test(f"✓ Configuration: {len(status['configuration'])} fields") pprint_test(f"✓ Satellites by room: {status['satellites_by_room']}") pprint_test(f"✓ Capabilities: {status['capabilities']}") # Test individual satellite status for i, satellite in enumerate(satellites[:2]): # Test first 2 sat_status = satellite.get_status() pprint_test(f"✓ Satellite {i+1} status: {len(sat_status)} fields") pprint_test(f" - Connection status: {sat_status['connection']['status']}") pprint_test(f" - Capabilities: {len(sat_status['capabilities'])}") pprint_test(f" - Uptime: {sat_status['connection']['uptime_seconds']:.1f}s") pprint_test("✓ Statistics and monitoring tests passed") def test_query_parser(): """Test the query parser functionality.""" pprint_test("=== Testing Query Parser ===") parser = QueryParser(case_sensitive=False) # Test query parsing test_queries = [ "status=connected", "room=kitchen", "status=connected,room=kitchen", "capability=wakeword_detection", "uptime>3600", "error_count=0", "alias contains kitchen", "ip starts_with 192.168", "status in [connected,connecting]", "capability exists", ] for query in test_queries: try: conditions = parser.parse_query(query) pprint_test(f"✓ Parsed '{query}' -> {len(conditions)} conditions") except Exception as e: pprint_test(f"✗ Failed to parse '{query}': {e}") # Test field validation supported_fields = parser.get_supported_fields() pprint_test(f"✓ Supports {len(supported_fields)} query fields") field_aliases = parser.get_field_aliases() pprint_test(f"✓ Supports {len(field_aliases)} field aliases") pprint_test("✓ Query parser tests passed") def test_file_persistence(): """Test file persistence and recovery.""" pprint_test("=== Testing File Persistence ===") temp_dir = tempfile.mkdtemp() try: # Create manager with file persistence app = MockApplication() manager = SatelliteManager( app, registration_dir=os.path.join(temp_dir, "registrations"), blacklist_file=os.path.join(temp_dir, "blacklist.json") ) # Add some satellites for i in range(3): sat_info = create_test_satellite_info( f"ff:ee:dd:cc:bb:0{i+1}", f"room_{i+1}", f"Satellite {i+1}" ) manager.add_satellite(sat_info) # Add to blacklist manager.add_to_blacklist("aa:aa:aa:aa:aa:aa", "test_persistence") pprint_test(f"✓ Created manager with {len(manager)} satellites") # Create new manager to test loading manager2 = SatelliteManager( app, registration_dir=os.path.join(temp_dir, "registrations"), blacklist_file=os.path.join(temp_dir, "blacklist.json") ) pprint_test(f"✓ Loaded manager with {len(manager2)} satellites") assert len(manager2) == len(manager) # Check blacklist was loaded assert manager2.is_blacklisted("aa:aa:aa:aa:aa:aa") pprint_test("✓ Blacklist loaded successfully") pprint_test("✓ File persistence tests passed") finally: shutil.rmtree(temp_dir, ignore_errors=True) def run_comprehensive_test(): """Run all tests in sequence.""" pprint_test("=" * 60) pprint_test("COMPREHENSIVE SATELLITE MANAGEMENT SYSTEM TEST") pprint_test("=" * 60) try: # Basic functionality manager, satellites = test_basic_satellite_manager() # Advanced patterns as specified in CLAUDE.md test_advanced_access_patterns(manager, satellites) # Bulk operations test_bulk_operations(manager, satellites) # Connection management test_connection_management(manager, satellites) # Statistics and monitoring test_statistics_and_monitoring(manager, satellites) # Registration system test_registration_system() # Query parser test_query_parser() # File persistence test_file_persistence() pprint_test("=" * 60) pprint_test("✅ ALL TESTS PASSED - SATELLITE SYSTEM READY FOR PRODUCTION") pprint_test("=" * 60) except Exception as e: pprint_test("=" * 60) pprint_test(f"❌ TEST FAILED: {e}") pprint_test("=" * 60) import traceback traceback.print_exc() return False return True if __name__ == "__main__": success = run_comprehensive_test() sys.exit(0 if success else 1)