254 lines
9.7 KiB
Python
254 lines
9.7 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 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", headers=get_auth_headers(), 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()
|
|
|
|
# Authenticate first
|
|
print("🔐 Authenticating...")
|
|
if not authenticate():
|
|
print("❌ Authentication failed, some tests may not work")
|
|
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()
|