drone-detector/test_device_health.py

#!/usr/bin/env python3
"""
Device Health Monitoring Test Script
Tests the device health monitoring system by checking current device status
and simulating offline/recovery scenarios.
"""

import requests
import time
import os
from datetime import datetime

# Disable SSL warnings for self-signed certificates
import warnings
warnings.filterwarnings('ignore', message='Unverified HTTPS request')

# Configuration from environment variables
API_BASE_URL = os.getenv('API_BASE_URL', 'http://localhost:3002/api')
BASE_PATH = os.getenv('VITE_BASE_PATH', '').rstrip('/')

# If BASE_PATH is set, construct the full URL
if BASE_PATH and not API_BASE_URL.endswith('/api'):
    domain = API_BASE_URL.replace('/api', '').replace('/drones/api', '').replace('/uggla/api', '')
    API_BASE_URL = f"{domain}{BASE_PATH}/api"

def test_device_health_status():
    """Test device health monitoring status"""
    print("🔍 Testing Device Health Monitoring System")
    print("=" * 60)
    
    try:
        # Check device health service status
        print("📊 Checking device health service status...")
        response = requests.get(f"{API_BASE_URL}/device-health/status", verify=False, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            if data.get('success'):
                status = data['data']
                print(f"✅ Device Health Service Status:")
                print(f"   Running: {status['isRunning']}")
                print(f"   Check Interval: {status['checkIntervalMinutes']} minutes")
                print(f"   Offline Threshold: {status['offlineThresholdMinutes']} minutes")
                print(f"   Currently Offline Devices: {status['offlineDevicesCount']}")
                
                if status['offlineDevices']:
                    print(f"   Offline Devices:")
                    for device in status['offlineDevices']:
                        print(f"     - Device {device['deviceId']}: {device['deviceName']}")
                        print(f"       Offline since: {device['offlineSince']}")
                        print(f"       Alert sent: {device['alertSent']}")
                else:
                    print(f"   ✅ All devices are online")
            else:
                print(f"❌ Failed to get status: {data.get('message', 'Unknown error')}")
        else:
            print(f"❌ HTTP Error: {response.status_code}")
            print(f"   Response: {response.text}")
            
    except requests.exceptions.RequestException as e:
        print(f"❌ Network error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")

def test_manual_health_check():
    """Trigger a manual health check"""
    print("\n🔧 Triggering Manual Health Check")
    print("-" * 40)
    
    try:
        response = requests.post(f"{API_BASE_URL}/device-health/check", verify=False, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            if data.get('success'):
                print(f"✅ {data['message']}")
            else:
                print(f"❌ Failed: {data.get('message', 'Unknown error')}")
        else:
            print(f"❌ HTTP Error: {response.status_code}")
            print(f"   Response: {response.text}")
            
    except requests.exceptions.RequestException as e:
        print(f"❌ Network error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")

def test_device_list():
    """Check current device list and their status"""
    print("\n📱 Checking Device List and Status")
    print("-" * 40)
    
    try:
        response = requests.get(f"{API_BASE_URL}/devices", verify=False, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            if data.get('success') and data.get('data'):
                devices = data['data']
                print(f"📊 Found {len(devices)} devices:")
                
                for device in devices:
                    status = device.get('stats', {}).get('status', 'unknown')
                    last_heartbeat = device.get('last_heartbeat')
                    time_since = device.get('stats', {}).get('time_since_last_heartbeat')
                    
                    status_icon = "✅" if status == "online" else "❌" if status == "offline" else "⚠️"
                    
                    print(f"   {status_icon} Device {device['id']}: {device.get('name', 'Unnamed')}")
                    print(f"      Status: {status}")
                    print(f"      Location: {device.get('location_description', 'No location')}")
                    print(f"      Active: {device.get('is_active', False)}")
                    print(f"      Approved: {device.get('is_approved', False)}")
                    
                    if last_heartbeat:
                        print(f"      Last Heartbeat: {last_heartbeat}")
                    else:
                        print(f"      Last Heartbeat: Never")
                    
                    if time_since is not None:
                        print(f"      Time Since Last Heartbeat: {time_since:.0f} seconds")
                    print()
            else:
                print(f"❌ No devices found or invalid response")
        else:
            print(f"❌ HTTP Error: {response.status_code}")
            print(f"   Response: {response.text}")
            
    except requests.exceptions.RequestException as e:
        print(f"❌ Network error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")

def test_alert_rules():
    """Check if there are any device offline alert rules configured"""
    print("\n⚠️  Checking Device Offline Alert Rules")
    print("-" * 40)
    
    try:
        response = requests.get(f"{API_BASE_URL}/alerts/rules", verify=False, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            if data.get('success') and data.get('data'):
                rules = data['data']
                offline_rules = []
                
                for rule in rules:
                    conditions = rule.get('conditions', {})
                    if conditions.get('device_offline'):
                        offline_rules.append(rule)
                
                if offline_rules:
                    print(f"📋 Found {len(offline_rules)} device offline alert rules:")
                    for rule in offline_rules:
                        print(f"   📌 Rule: {rule['name']}")
                        print(f"      Description: {rule.get('description', 'No description')}")
                        print(f"      Active: {rule.get('is_active', False)}")
                        print(f"      Channels: {rule.get('alert_channels', [])}")
                        if rule.get('sms_phone_number'):
                            print(f"      SMS: {rule['sms_phone_number']}")
                        print()
                else:
                    print("⚠️  No device offline alert rules found!")
                    print("   You may want to create alert rules for device offline monitoring.")
            else:
                print(f"❌ No alert rules found")
        else:
            print(f"❌ HTTP Error: {response.status_code}")
            print(f"   Response: {response.text}")
            
    except requests.exceptions.RequestException as e:
        print(f"❌ Network error: {e}")
    except Exception as e:
        print(f"❌ Unexpected error: {e}")

def main():
    print("🏥 DEVICE HEALTH MONITORING TEST")
    print("=" * 60)
    print(f"📡 API URL: {API_BASE_URL}")
    print(f"⏰ Test Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print()
    
    # Run all tests
    test_device_health_status()
    test_device_list()
    test_alert_rules()
    test_manual_health_check()
    
    print("\n" + "=" * 60)
    print("🏁 Device Health Monitoring Test Complete")
    print("=" * 60)

if __name__ == "__main__":
    main()