test_admin_monitoring.py

#!/usr/bin/env python3
"""
Test script to verify admin monitoring functionality
"""

import requests
import json
import time

def test_admin_monitoring():
    """Test the admin monitoring system"""
    base_url = "http://127.0.0.1:8000"
    
    print("Testing UnForkRAG Admin Monitoring System")
    print("=" * 50)
    
    # Test 1: Check if server is running
    try:
        response = requests.get(f"{base_url}/health", timeout=5)
        if response.status_code == 200:
            print("✅ Server is running")
        else:
            print(f"❌ Server health check failed: {response.status_code}")
            return
    except Exception as e:
        print(f"❌ Cannot connect to server: {e}")
        return
    
    # Test 2: Check observability API
    try:
        response = requests.get(f"{base_url}/api/observe", timeout=5)
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Observability API working - {len(data.get('observations', []))} observations")
        else:
            print(f"❌ Observability API failed: {response.status_code}")
    except Exception as e:
        print(f"❌ Observability API error: {e}")
    
    # Test 3: Check admin login page
    try:
        response = requests.get(f"{base_url}/admin/login", timeout=5)
        if response.status_code == 200:
            print("✅ Admin login page accessible")
        else:
            print(f"❌ Admin login page failed: {response.status_code}")
    except Exception as e:
        print(f"❌ Admin login page error: {e}")
    
    # Test 4: Make some API calls to generate observations
    print("\nGenerating test API calls...")
    try:
        # Test API endpoints
        requests.get(f"{base_url}/api/collections", timeout=5)
        requests.get(f"{base_url}/api/version", timeout=5)
        requests.get(f"{base_url}/qdrant/healthz", timeout=5)
        
        # Wait a moment for observations to be logged
        time.sleep(1)
        
        # Check observations again
        response = requests.get(f"{base_url}/api/observe", timeout=5)
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Generated {len(data.get('observations', []))} observations from test calls")
            
            # Show recent observations
            if data.get('observations'):
                print("\nRecent observations:")
                for obs in data['observations'][:3]:
                    print(f"  - {obs['timestamp']}: {obs['method']} {obs['path']} ({obs['duration_ms']}ms)")
        else:
            print("❌ Cannot retrieve observations after test calls")
    except Exception as e:
        print(f"❌ Error during test API calls: {e}")
    
    print("\n" + "=" * 50)
    print("Admin Monitoring Test Complete")
    print("\nTo access the admin panel:")
    print("1. Open http://127.0.0.1:8000/admin in your browser")
    print("2. You will be redirected to the login page")
    print("3. Login with:")
    print("   Username: admin")
    print("   Password: admin123")
    print("4. The admin dashboard will show real-time API monitoring")
    print("5. Click 'Start Viewer' to enable live updates")

if __name__ == "__main__":
    test_admin_monitoring()