remote-sensing/test_new_features.py

#!/usr/bin/env python3
"""
Demo script để test các chức năng mới của API
"""

import requests
import json
import time
from pathlib import Path

BASE_URL = "http://localhost:8000"

def print_section(title):
    print("\n" + "=" * 70)
    print(f"  {title}")
    print("=" * 70)

def test_dashboard_statistics():
    print_section("📊 Test Dashboard Statistics")
    try:
        response = requests.get(f"{BASE_URL}/api/dashboard/statistics")
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Success!")
            print(f"   Models: {data['models']['total']}")
            print(f"   Predictions: {data['predictions']['total']}")
            print(f"   Reports: {data['reports']['total']}")
        else:
            print(f"❌ Error: {response.status_code}")
    except Exception as e:
        print(f"❌ Exception: {e}")

def test_accuracy_trends():
    print_section("📈 Test Accuracy Trends")
    try:
        response = requests.get(f"{BASE_URL}/api/dashboard/accuracy-trends")
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Success!")
            print(f"   Trends: {len(data['trends'])} records")
            print(f"   Models: {data['models']}")
        else:
            print(f"❌ Error: {response.status_code}")
    except Exception as e:
        print(f"❌ Exception: {e}")

def test_class_distribution():
    print_section("📊 Test Class Distribution")
    try:
        # First, get list of models
        response = requests.get(f"{BASE_URL}/api/models/list")
        if response.status_code == 200:
            models = response.json()['models']
            if models:
                model_filename = models[0]['filename']
                print(f"   Using model: {model_filename}")
                
                # Get class distribution
                response = requests.get(f"{BASE_URL}/api/dashboard/class-distribution/{model_filename}")
                if response.status_code == 200:
                    data = response.json()
                    print(f"✅ Success!")
                    print(f"   Total samples: {data['total_samples']}")
                    print(f"   Classes: {list(data['class_distribution'].keys())}")
                else:
                    print(f"❌ Error: {response.status_code}")
            else:
                print("⚠️  No models found")
        else:
            print(f"❌ Error getting models: {response.status_code}")
    except Exception as e:
        print(f"❌ Exception: {e}")

def test_batch_status():
    print_section("🔄 Test Batch Status")
    try:
        response = requests.get(f"{BASE_URL}/api/batch/status")
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Success!")
            print(f"   Queued: {data['queue']['queued']}")
            print(f"   Running: {data['queue']['running']}")
            print(f"   Completed: {data['queue']['completed']}")
            print(f"   Failed: {data['queue']['failed']}")
        else:
            print(f"❌ Error: {response.status_code}")
    except Exception as e:
        print(f"❌ Exception: {e}")

def test_batch_prediction_demo():
    print_section("🚀 Test Batch Prediction (Demo)")
    try:
        # Get a model
        response = requests.get(f"{BASE_URL}/api/models/list")
        if response.status_code != 200:
            print("❌ Cannot get models list")
            return
        
        models = response.json()['models']
        if not models:
            print("⚠️  No models available for testing")
            return
        
        model_filename = models[0]['filename']
        print(f"   Using model: {model_filename}")
        
        # Create test batch
        batch_config = {
            "model_filename": model_filename,
            "items": [
                {
                    "name": "Test_Region_1",
                    "min_lon": 105.6,
                    "min_lat": 9.3,
                    "max_lon": 105.7,
                    "max_lat": 9.4,
                    "start_date": "2023-03-01",
                    "end_date": "2023-03-31",
                    "max_scenes": 5,
                    "cloud_cover": 30,
                    "resolution": 20
                }
            ],
            "auto_retry": True,
            "max_retries": 2
        }
        
        print("   Creating batch job...")
        response = requests.post(
            f"{BASE_URL}/api/batch/start",
            json=batch_config
        )
        
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Success!")
            print(f"   {data['message']}")
            print(f"   Batch ID: {data['batch_id']}")
            
            # Check status after a moment
            time.sleep(2)
            response = requests.get(f"{BASE_URL}/api/batch/status")
            if response.status_code == 200:
                status = response.json()
                print(f"   Current queue: {status['queue']}")
        else:
            print(f"❌ Error: {response.status_code} - {response.text}")
    except Exception as e:
        print(f"❌ Exception: {e}")

def test_reports_list():
    print_section("📄 Test Reports List")
    try:
        response = requests.get(f"{BASE_URL}/api/reports/list")
        if response.status_code == 200:
            data = response.json()
            print(f"✅ Success!")
            print(f"   Total reports: {data['count']}")
            if data['reports']:
                print(f"   Latest report: {data['reports'][0]['filename']}")
        else:
            print(f"❌ Error: {response.status_code}")
    except Exception as e:
        print(f"❌ Exception: {e}")

def main():
    print("=" * 70)
    print("  🧪 API Testing Suite - New Features")
    print("=" * 70)
    print(f"\n  Base URL: {BASE_URL}")
    print(f"  Đảm bảo server đang chạy: python api_server.py")
    
    input("\n  Press ENTER to start testing...")
    
    # Run all tests
    test_dashboard_statistics()
    test_accuracy_trends()
    test_class_distribution()
    test_reports_list()
    test_batch_status()
    test_batch_prediction_demo()
    
    print("\n" + "=" * 70)
    print("  ✅ Testing completed!")
    print("=" * 70)
    print(f"\n  Dashboard: {BASE_URL}/dashboard")
    print(f"  API Docs: {BASE_URL}/docs")
    print("=" * 70 + "\n")

if __name__ == "__main__":
    main()