Trixy3/trixy_core/satellites/test_satellite_system.py

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