drone-detector/test_device_api.py

#!/usr/bin/env python3
"""
Test script for device API authentication and detection submission
Simulates Stockholm Castle device sending drone detection data
"""

import requests
import json
import sys
import os
from datetime import datetime, timezone

# Configuration
API_BASE_URL = os.getenv('API_BASE_URL', 'http://localhost:3001/api')
TENANT_ID = os.getenv('TENANT_ID', 'uamils-ab')
DEVICE_ID = os.getenv('DEVICE_ID', '1')  # Stockholm Castle device
USERNAME = os.getenv('TEST_USERNAME', 'admin')
PASSWORD = os.getenv('TEST_PASSWORD', 'admin123')

# Headers for multi-tenant API
HEADERS = {
    'Content-Type': 'application/json',
    'x-tenant-id': TENANT_ID,
    'User-Agent': 'Stockholm-Castle-Device/1.0'
}

class DeviceAPITester:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update(HEADERS)
        self.jwt_token = None
        
    def authenticate(self):
        """Authenticate with the backend and get JWT token"""
        print(f"🔐 Authenticating as user: {USERNAME} for tenant: {TENANT_ID}")
        
        auth_data = {
            'username': USERNAME,
            'password': PASSWORD
        }
        
        try:
            response = self.session.post(f"{API_BASE_URL}/auth/local", json=auth_data)
            
            if response.status_code == 200:
                data = response.json()
                if data.get('success') and data.get('token'):
                    self.jwt_token = data['token']
                    self.session.headers['Authorization'] = f'Bearer {self.jwt_token}'
                    print(f"✅ Authentication successful")
                    print(f"   User: {data.get('user', {}).get('username', 'N/A')}")
                    print(f"   Role: {data.get('user', {}).get('role', 'N/A')}")
                    print(f"   Tenant: {data.get('user', {}).get('tenant_id', 'N/A')}")
                    return True
                else:
                    print(f"❌ Authentication failed: Invalid response format")
                    print(f"Response: {response.text}")
                    return False
            else:
                print(f"❌ Authentication failed: {response.status_code}")
                print(f"Response: {response.text}")
                return False
                
        except requests.exceptions.RequestException as e:
            print(f"❌ Authentication error: {e}")
            return False
    
    def check_health(self):
        """Check if API is healthy"""
        try:
            response = self.session.get(f"{API_BASE_URL}/health")
            if response.status_code == 200:
                print("✅ API health check passed")
                return True
            else:
                print(f"❌ API health check failed: {response.status_code}")
                return False
        except requests.exceptions.RequestException as e:
            print(f"❌ API health check error: {e}")
            return False
    
    def send_detection(self, drone_type="Orlan", distance=1000, confidence=0.85):
        """Send a drone detection to the API"""
        detection_data = {
            'device_id': DEVICE_ID,
            'drone_type': drone_type,
            'confidence': confidence,
            'distance': distance,
            'signal_strength': -45.2,
            'frequency': 2400,
            'coordinates': {
                'latitude': 59.3251,  # Stockholm Castle
                'longitude': 18.0719
            },
            'raw_data': {
                'rssi': -45.2,
                'snr': 12.5,
                'frequency_mhz': 2400,
                'bandwidth': '20MHz',
                'detected_at': datetime.now(timezone.utc).isoformat()
            },
            'metadata': {
                'device_location': 'Stockholm Castle',
                'weather': 'Clear',
                'temperature': 15.2
            }
        }
        
        print(f"📡 Sending detection: {drone_type} at {distance}m distance")
        
        try:
            response = self.session.post(f"{API_BASE_URL}/detections", json=detection_data)
            
            if response.status_code == 201:
                data = response.json()
                print(f"✅ Detection sent successfully")
                print(f"   Detection ID: {data.get('id', 'N/A')}")
                print(f"   Timestamp: {data.get('timestamp', 'N/A')}")
                return True
            else:
                print(f"❌ Detection failed: {response.status_code}")
                print(f"Response: {response.text}")
                return False
                
        except requests.exceptions.RequestException as e:
            print(f"❌ Detection error: {e}")
            return False
    
    def send_heartbeat(self):
        """Send a heartbeat to show device is online"""
        heartbeat_data = {
            'device_id': DEVICE_ID,
            'status': 'online',
            'battery_level': 95,
            'signal_quality': 85,
            'last_detection': datetime.now(timezone.utc).isoformat(),
            'location': {
                'latitude': 59.3251,
                'longitude': 18.0719
            },
            'system_info': {
                'firmware_version': '1.2.3',
                'uptime': 3600,
                'cpu_usage': 25.5,
                'memory_usage': 45.2
            }
        }
        
        print(f"💓 Sending heartbeat for device {DEVICE_ID}")
        
        try:
            response = self.session.post(f"{API_BASE_URL}/heartbeat", json=heartbeat_data)
            
            if response.status_code in [200, 201]:
                print(f"✅ Heartbeat sent successfully")
                return True
            else:
                print(f"❌ Heartbeat failed: {response.status_code}")
                print(f"Response: {response.text}")
                return False
                
        except requests.exceptions.RequestException as e:
            print(f"❌ Heartbeat error: {e}")
            return False

def main():
    print("=" * 70)
    print("🎯 STOCKHOLM CASTLE DEVICE API TEST")
    print("=" * 70)
    print(f"API URL: {API_BASE_URL}")
    print(f"Tenant: {TENANT_ID}")
    print(f"Device ID: {DEVICE_ID}")
    print(f"Username: {USERNAME}")
    print("=" * 70)
    
    tester = DeviceAPITester()
    
    # Step 1: Health check
    if not tester.check_health():
        print("❌ API not healthy. Please check if server is running.")
        sys.exit(1)
    
    # Step 2: Authenticate
    if not tester.authenticate():
        print("❌ Authentication failed. Please check credentials and tenant.")
        print("\n💡 Troubleshooting:")
        print("1. Ensure backend is running on correct port")
        print("2. Check username/password are correct")
        print("3. Verify tenant 'uamils-ab' exists")
        print("4. Check database seeding completed successfully")
        sys.exit(1)
    
    # Step 3: Send heartbeat
    print("\n" + "=" * 50)
    print("Testing Device Heartbeat")
    print("=" * 50)
    tester.send_heartbeat()
    
    # Step 4: Send detection
    print("\n" + "=" * 50)
    print("Testing Drone Detection")
    print("=" * 50)
    tester.send_detection(drone_type="Orlan", distance=1500, confidence=0.92)
    
    # Step 5: Send another detection (closer)
    print("\n" + "=" * 50)
    print("Testing Critical Alert (Close Drone)")
    print("=" * 50)
    tester.send_detection(drone_type="Orlan", distance=500, confidence=0.98)
    
    print("\n" + "=" * 70)
    print("✅ TEST COMPLETED")
    print("=" * 70)
    print("Check the web interface to see the detections and device status")

if __name__ == "__main__":
    main()