Fix jwt-token
This commit is contained in:
195
test_device_health.py
Normal file
195
test_device_health.py
Normal file
@@ -0,0 +1,195 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user