211 lines
7.9 KiB
Python
211 lines
7.9 KiB
Python
#!/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('/')
|
||
|
||
# Authentication configuration - Optional for local testing
|
||
USERNAME = os.getenv('TEST_USERNAME', 'admin')
|
||
PASSWORD = os.getenv('TEST_PASSWORD', 'admin123')
|
||
SKIP_AUTH = os.getenv('SKIP_AUTH', 'false').lower() == 'true'
|
||
|
||
# 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"
|
||
|
||
# Global variable to store authentication token
|
||
AUTH_TOKEN = None
|
||
|
||
def authenticate():
|
||
"""Authenticate with the API and get access token"""
|
||
global AUTH_TOKEN
|
||
|
||
if SKIP_AUTH:
|
||
print("🔓 Authentication skipped (SKIP_AUTH=true)")
|
||
return True
|
||
|
||
try:
|
||
login_data = {
|
||
"username": USERNAME,
|
||
"password": PASSWORD
|
||
}
|
||
|
||
response = requests.post(f"{API_BASE_URL}/users/login", json=login_data, verify=False, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
if data.get('success') and 'data' in data and 'token' in data['data']:
|
||
AUTH_TOKEN = data['data']['token']
|
||
print(f"✅ Authentication successful")
|
||
return True
|
||
else:
|
||
print(f"❌ Invalid login response: {data}")
|
||
return False
|
||
else:
|
||
print(f"❌ Authentication failed: HTTP {response.status_code}")
|
||
print(f" Response: {response.text}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ Authentication error: {e}")
|
||
return False
|
||
|
||
def get_auth_headers():
|
||
"""Get headers with authentication token"""
|
||
headers = {
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
if not SKIP_AUTH and AUTH_TOKEN:
|
||
headers['Authorization'] = f'Bearer {AUTH_TOKEN}'
|
||
|
||
return headers
|
||
|
||
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", headers=get_auth_headers(), 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 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()
|
||
|
||
# Authenticate first
|
||
print("🔐 Authenticating...")
|
||
if not authenticate():
|
||
print("❌ Authentication failed, some tests may not work")
|
||
print()
|
||
|
||
# Run relevant tests only
|
||
test_device_health_status()
|
||
test_device_list()
|
||
test_manual_health_check()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("🏁 Device Health Monitoring Test Complete")
|
||
print("=" * 60)
|
||
print("ℹ️ Note: Alert rules are managed by the backend service automatically")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|