diff --git a/health_probe_simulator.py b/health_probe_simulator.py index ec3b920..27810a4 100644 --- a/health_probe_simulator.py +++ b/health_probe_simulator.py @@ -12,28 +12,93 @@ import json from datetime import datetime from typing import List, Dict +# Disable SSL warnings for self-signed certificates +import warnings +warnings.filterwarnings('ignore', message='Unverified HTTPS request') + class DeviceProbeSimulator: def __init__(self): - self.api_base_url = os.getenv('API_BASE_URL', 'https://selfservice.cqers.com/drones/api') + # Configuration from environment variables + # Tests default to localhost:3002 for local development + self.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 self.api_base_url.endswith('/api'): + # Extract domain from API_BASE_URL and add base path + domain = self.api_base_url.replace('/api', '').replace('/drones/api', '').replace('/uggla/api', '') + self.api_base_url = f"{domain}{base_path}/api" + self.probe_interval = int(os.getenv('PROBE_INTERVAL_SECONDS', '60')) # 1 minute default self.probe_failrate = int(os.getenv('PROBE_FAILRATE', '30')) # 30% failure rate + + # Authentication configuration - Optional for local testing + self.username = os.getenv('TEST_USERNAME', 'admin') + self.password = os.getenv('TEST_PASSWORD', 'admin123') + self.skip_auth = os.getenv('SKIP_AUTH', 'false').lower() == 'true' + self.devices = [] self.session = requests.Session() self.session.headers.update({ 'Content-Type': 'application/json', 'User-Agent': 'HealthProbeSimulator/1.0' }) + self.auth_token = None print(f"🔧 Health Probe Simulator Configuration:") print(f" API URL: {self.api_base_url}") print(f" Probe Interval: {self.probe_interval} seconds") print(f" Failure Rate: {self.probe_failrate}%") + print(f" Authentication: {'Disabled' if self.skip_auth else 'Enabled'}") print() + def authenticate(self) -> bool: + """Authenticate with the API and get a token""" + if self.skip_auth: + print("🔓 Authentication skipped (SKIP_AUTH=true)") + return True + + try: + login_data = { + "username": self.username, + "password": self.password + } + + response = self.session.post( + f"{self.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']: + self.auth_token = data['data']['token'] + self.session.headers.update({ + 'Authorization': f'Bearer {self.auth_token}' + }) + print(f"✅ Authentication successful") + return True + else: + print(f"❌ Invalid login response format: {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 fetch_devices(self) -> bool: """Fetch all devices from the API""" try: - response = self.session.get(f"{self.api_base_url}/devices") + response = self.session.get( + f"{self.api_base_url}/devices", + verify=False, + timeout=10 + ) if response.status_code == 200: data = response.json() if data.get('success') and 'data' in data: @@ -72,8 +137,9 @@ class DeviceProbeSimulator: print(f"📡 Sending heartbeat: {payload}") response = self.session.post( - f"{self.api_base_url}/heartbeat", # Changed to correct endpoint + f"{self.api_base_url}/detectors", # Updated to use detectors endpoint json=payload, + verify=False, timeout=10 ) @@ -131,6 +197,11 @@ class DeviceProbeSimulator: print(" Press Ctrl+C to stop") print() + # Authenticate first + if not self.authenticate(): + print("❌ Failed to authenticate, exiting") + return + # Initial device fetch if not self.fetch_devices(): print("❌ Failed to fetch initial device list, exiting") @@ -164,15 +235,19 @@ def main(): print("=" * 60) print() - # Load environment variables + # Load environment variables with localhost defaults probe_failrate = os.getenv('PROBE_FAILRATE', '30') probe_interval = os.getenv('PROBE_INTERVAL_SECONDS', '60') - api_url = os.getenv('API_BASE_URL', 'https://selfservice.cqers.com/drones/api') + api_url = os.getenv('API_BASE_URL', 'http://localhost:3002/api') + base_path = os.getenv('VITE_BASE_PATH', '') + skip_auth = os.getenv('SKIP_AUTH', 'false').lower() == 'true' print(f"📋 Configuration:") print(f" PROBE_FAILRATE: {probe_failrate}%") print(f" PROBE_INTERVAL_SECONDS: {probe_interval}s") print(f" API_BASE_URL: {api_url}") + print(f" VITE_BASE_PATH: {base_path}") + print(f" SKIP_AUTH: {skip_auth}") print() simulator = DeviceProbeSimulator()