heartbeat/scripts/demo_http_api.py

#!/usr/bin/env python3
"""
Demo script for HTTP API endpoints.
Tests and demonstrates the plugin data and alert APIs.
"""

import requests
import json
import sys
from datetime import datetime
from time import sleep

BASE_URL = "http://localhost:50004"

def print_section(title):
    """Print a formatted section header."""
    print(f"\n{'=' * 70}")
    print(f"  {title}")
    print('=' * 70)

def format_timestamp(timestamp):
    """Convert Unix timestamp to readable format."""
    return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')

def format_duration(seconds):
    """Format duration in human-readable format."""
    if seconds < 60:
        return f"{int(seconds)}s"
    elif seconds < 3600:
        minutes = int(seconds / 60)
        secs = int(seconds % 60)
        return f"{minutes}m {secs}s"
    elif seconds < 86400:
        hours = int(seconds / 3600)
        minutes = int((seconds % 3600) / 60)
        return f"{hours}h {minutes}m"
    else:
        days = int(seconds / 86400)
        hours = int((seconds % 86400) / 3600)
        return f"{days}d {hours}h"

def test_hosts_api():
    """Test GET /api/0/hosts endpoint."""
    print_section("1. List All Monitored Hosts")
    
    try:
        response = requests.get(f"{BASE_URL}/api/0/hosts", timeout=5)
        response.raise_for_status()
        hosts = response.json()
        
        print(f"Found {len(hosts)} hosts:\n")
        for host in hosts:
            name = host.get('name', 'unknown')
            dyn = host.get('dyn', False)
            conn_count = len(host.get('connections', []))
            
            print(f"  • {name}")
            print(f"    - Protocol: IPv{ver}")
            print(f"    - Dynamic: {dyn}")
            print(f"    - Connections: {conn_count}")
        
        return hosts
    
    except requests.RequestException as e:
        print(f"❌ Error: {e}")
        return []

def test_host_plugins_api(hostname):
    """Test GET /api/0/hosts/{hostname}/plugins endpoint."""
    print_section(f"2. Get All Plugins for Host: {hostname}")
    
    try:
        response = requests.get(f"{BASE_URL}/api/0/hosts/{hostname}/plugins", timeout=5)
        response.raise_for_status()
        data = response.json()
        
        plugins = data.get('plugins', {})
        print(f"Found {len(plugins)} plugins:\n")
        
        for plugin_name, plugin_data in plugins.items():
            timestamp = plugin_data.get('timestamp', 0)
            sample_count = plugin_data.get('sample_count', 0)
            metrics = plugin_data.get('data', {})
            
            print(f"  📦 {plugin_name}")
            print(f"     Last update: {format_timestamp(timestamp)}")
            print(f"     Samples: {sample_count}")
            print(f"     Metrics: {len(metrics)}")
            
            # Show first few metrics
            for i, (metric, value) in enumerate(metrics.items()):
                if i < 3:  # Show only first 3 metrics
                    if isinstance(value, float):
                        print(f"       - {metric}: {value:.2f}")
                    elif isinstance(value, dict):
                        print(f"       - {metric}: [nested data, {len(value)} keys]")
                    else:
                        print(f"       - {metric}: {value}")
            
            if len(metrics) > 3:
                print(f"       ... and {len(metrics) - 3} more")
            print()
        
        return list(plugins.keys())
    
    except requests.RequestException as e:
        print(f"❌ Error: {e}")
        return []

def test_plugin_detail_api(hostname, plugin_name, limit=5):
    """Test GET /api/0/hosts/{hostname}/plugins/{plugin_name} endpoint."""
    print_section(f"3. Get Detailed Data: {hostname}/{plugin_name}")
    
    try:
        url = f"{BASE_URL}/api/0/hosts/{hostname}/plugins/{plugin_name}"
        params = {'limit': limit}
        response = requests.get(url, params=params, timeout=5)
        response.raise_for_status()
        data = response.json()
        
        samples = data.get('samples', [])
        print(f"Retrieved {len(samples)} samples (limit={limit}):\n")
        
        for i, sample in enumerate(samples):
            timestamp = sample.get('timestamp', 0)
            metrics = sample.get('data', {})
            
            print(f"  [{i+1}] {format_timestamp(timestamp)}")
            for metric, value in sorted(metrics.items())[:5]:  # Show first 5 metrics
                if isinstance(value, float):
                    print(f"      {metric}: {value:.2f}")
                elif isinstance(value, dict):
                    print(f"      {metric}: [nested: {len(value)} keys]")
                else:
                    print(f"      {metric}: {value}")
            print()
        
        return samples
    
    except requests.RequestException as e:
        print(f"❌ Error: {e}")
        return []

def test_host_alerts_api(hostname):
    """Test GET /api/0/hosts/{hostname}/alerts endpoint."""
    print_section(f"4. Get Alerts for Host: {hostname}")
    
    try:
        response = requests.get(f"{BASE_URL}/api/0/hosts/{hostname}/alerts", timeout=5)
        response.raise_for_status()
        data = response.json()
        
        alerts = data.get('alerts', [])
        summary = data.get('summary', {})
        
        print(f"Summary:")
        print(f"  ✓ OK: {summary.get('ok', 0)}")
        print(f"  ⚠️ Warning: {summary.get('warning', 0)}")
        print(f"  🔴 Critical: {summary.get('critical', 0)}")
        print(f"  ❓ Unknown: {summary.get('unknown', 0)}")
        print()
        
        # Show non-OK alerts
        active_alerts = [a for a in alerts if a.get('level') != 'OK']
        if active_alerts:
            print(f"Active Alerts ({len(active_alerts)}):")
            for alert in active_alerts:
                metric = alert.get('metric_path', 'unknown')
                level = alert.get('level', 'UNKNOWN')
                value = alert.get('last_value', 0)
                since = alert.get('since', 0)
                duration = datetime.now().timestamp() - since
                
                icon = '⚠️' if level == 'WARNING' else '🔴'
                print(f"  {icon} {metric}")
                print(f"     Level: {level}")
                print(f"     Value: {value:.2f}" if isinstance(value, float) else f"     Value: {value}")
                print(f"     Duration: {format_duration(duration)}")
                print()
        else:
            print("✓ No active alerts - all systems normal!")
        
        return data
    
    except requests.RequestException as e:
        print(f"❌ Error: {e}")
        return {}

def test_all_alerts_api():
    """Test GET /api/0/alerts endpoint."""
    print_section("5. Get All Active Alerts Across All Hosts")
    
    try:
        response = requests.get(f"{BASE_URL}/api/0/alerts", timeout=5)
        response.raise_for_status()
        data = response.json()
        
        alerts = data.get('alerts', [])
        summary = data.get('summary', {})
        host_count = data.get('host_count', 0)
        
        print(f"Monitoring {host_count} hosts")
        print(f"Active Alerts: {summary.get('total', 0)}")
        print(f"  🔴 Critical: {summary.get('critical', 0)}")
        print(f"  ⚠️ Warning: {summary.get('warning', 0)}")
        print()
        
        if alerts:
            print("Alert Details:")
            for alert in alerts:
                hostname = alert.get('hostname', 'unknown')
                metric = alert.get('metric_path', 'unknown')
                level = alert.get('level', 'UNKNOWN')
                value = alert.get('last_value', 0)
                since = alert.get('since', 0)
                duration = datetime.now().timestamp() - since
                notification_count = alert.get('notification_count', 0)
                
                icon = '⚠️' if level == 'WARNING' else '🔴'
                print(f"  {icon} {hostname} / {metric}")
                print(f"     Level: {level}")
                print(f"     Value: {value:.2f}" if isinstance(value, float) else f"     Value: {value}")
                print(f"     Duration: {format_duration(duration)}")
                print(f"     Notifications: {notification_count}")
                print()
        else:
            print("✅ All systems normal - no active alerts!")
        
        return data
    
    except requests.RequestException as e:
        print(f"❌ Error: {e}")
        return {}

def test_messages_api():
    """Test GET /api/0/messages endpoint."""
    print_section("6. Get Recent Messages")
    
    try:
        response = requests.get(f"{BASE_URL}/api/0/messages", timeout=5)
        response.raise_for_status()
        messages = response.json()
        
        print(f"Last {len(messages)} messages:\n")
        for msg in messages[-5:]:  # Show last 5
            timestamp = msg.get('time', 0)
            host = msg.get('host', 'unknown')
            text = msg.get('msg', '')
            
            print(f"  [{format_timestamp(timestamp)}] {host}: {text}")
        
        return messages
    
    except requests.RequestException as e:
        print(f"❌ Error: {e}")
        return []

def test_error_handling():
    """Test API error handling."""
    print_section("7. Error Handling Tests")
    
    # Test non-existent host
    print("Testing non-existent host...")
    try:
        response = requests.get(f"{BASE_URL}/api/0/hosts/nonexistenthost/plugins", timeout=5)
        if response.status_code == 404:
            error_data = response.json()
            print(f"  ✓ Correctly returned 404: {error_data.get('error', 'No error message')}")
        else:
            print(f"  ⚠️ Unexpected status code: {response.status_code}")
    except Exception as e:
        print(f"  ❌ Error: {e}")
    
    # Test non-existent plugin
    print("\nTesting non-existent plugin...")
    try:
        # Get first host
        hosts = requests.get(f"{BASE_URL}/api/0/hosts", timeout=5).json()
        if hosts:
            hostname = hosts[0]['name']
            response = requests.get(
                f"{BASE_URL}/api/0/hosts/{hostname}/plugins/nonexistentplugin",
                timeout=5
            )
            if response.status_code == 404:
                error_data = response.json()
                print(f"  ✓ Correctly returned 404: {error_data.get('error', 'No error message')}")
            else:
                print(f"  ⚠️ Unexpected status code: {response.status_code}")
    except Exception as e:
        print(f"  ❌ Error: {e}")

def demo_monitoring_loop():
    """Demonstrate continuous monitoring."""
    print_section("8. Continuous Monitoring Demo (5 iterations)")
    
    print("Monitoring alerts every 3 seconds (Ctrl+C to stop)...\n")
    
    try:
        for i in range(5):
            response = requests.get(f"{BASE_URL}/api/0/alerts", timeout=5)
            response.raise_for_status()
            data = response.json()
            
            summary = data.get('summary', {})
            critical = summary.get('critical', 0)
            warning = summary.get('warning', 0)
            
            timestamp = datetime.now().strftime('%H:%M:%S')
            status = "🔴 CRITICAL" if critical > 0 else "⚠️ WARNING" if warning > 0 else "✅ OK"
            
            print(f"[{timestamp}] {status} - Critical: {critical}, Warning: {warning}")
            
            if i < 4:  # Don't sleep after last iteration
                sleep(3)
    
    except KeyboardInterrupt:
        print("\n\nMonitoring stopped by user")
    except Exception as e:
        print(f"\n❌ Error: {e}")

def main():
    """Run all API tests."""
    print("""
╔══════════════════════════════════════════════════════════════╗
║         Heartbeat Daemon HTTP API Demo & Test Suite         ║
╚══════════════════════════════════════════════════════════════╝
""")
    
    print(f"Testing API at: {BASE_URL}")
    print(f"Ensure the heartbeat daemon is running!")
    
    # Test basic connectivity
    try:
        response = requests.get(f"{BASE_URL}/api/0/hosts", timeout=2)
        response.raise_for_status()
        print("✅ API is reachable\n")
    except Exception as e:
        print(f"❌ Cannot connect to API: {e}")
        print("\nPlease ensure:")
        print("  1. Heartbeat daemon is running")
        print("  2. HTTP server is enabled in configuration")
        print(f"  3. Server is listening on port {BASE_URL.split(':')[-1]}")
        sys.exit(1)
    
    # Run test suite
    hosts = test_hosts_api()
    
    if not hosts:
        print("\n⚠️ No hosts found. Ensure clients are sending heartbeats.")
        return
    
    # Pick first host for detailed testing
    hostname = hosts[0].get('name', '')
    
    if hostname:
        plugins = test_host_plugins_api(hostname)
        
        if plugins:
            # Test detailed plugin data
            test_plugin_detail_api(hostname, plugins[0], limit=3)
        
        # Test alert endpoints
        test_host_alerts_api(hostname)
    
    # Test global endpoints
    test_all_alerts_api()
    test_messages_api()
    
    # Test error handling
    test_error_handling()
    
    # Continuous monitoring demo
    demo_monitoring_loop()
    
    print_section("Test Suite Complete")
    print("""
Next Steps:
  • View the web UI at http://localhost:50004/live
  • Check plugin metrics at http://localhost:50004/plugins
  • Monitor alerts at http://localhost:50004/alerts
  • Read API documentation: docs/HTTP_API.md
""")

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print("\n\nDemo interrupted by user")
        sys.exit(0)