backend/core/scripts/test_database_switch.py

#!/usr/bin/env python3
"""
🧪 COMPREHENSIVE DATABASE SWITCHING API TEST
Tests the new /db/switch-database endpoint with multiple database configurations
"""
import os
import sys
import time
import json
import requests
from typing import Dict, Any
db_url2="mysql+pymysql://root:bwgadmin@2023@65.0.127.253:3306/bookwedgo"
db_url="postgresql://neondb_owner:npg_dfWNsn2ZGk7c@ep-cool-poetry-a1puamly-pooler.ap-southeast-1.aws.neon.tech/scv-sample?sslmode=require"
db_url3="postgresql://postgres:bSOGyvUKCkQfgNFnvVXjiTpqUtbvlOMr@centerbeam.proxy.rlwy.net:46916/railway"
# Test configurations
MYSQL_CONFIG = {
    "dialect": "mysql+pymysql",
    "host": "65.0.127.253",
    "port": "3306",
    "username": "root",
    "password": "bwgadmin@2023",
    "database": "bookwedgo"
}

POSTGRES_CONFIG = {
    "dialect": "postgresql+psycopg2",
    "host": "ep-cool-poetry-a1puamly-pooler.ap-southeast-1.aws.neon.tech",
    "port": "5432",
    "username": "neondb_owner",
    "password": "npg_dfWNsn2ZGk7c",
    "database": "scv-sample",
    "params": "sslmode=require"
}

# API Configuration
API_BASE = "http://localhost:5002"  # Direct Python service
SWITCH_ENDPOINT = f"{API_BASE}/db/switch-database"
TABLES_ENDPOINT = f"{API_BASE}/db/tables"

def test_database_switch(config: Dict[str, str], config_name: str) -> bool:
    """Test switching to a specific database configuration"""
    print(f"\n🧪 TESTING {config_name.upper()} DATABASE SWITCH")
    print("=" * 60)
    
    try:
        # Step 1: Call the switch API
        print(f"[TEST] 🔄 Switching to {config['dialect']}://{config['host']}:{config['port']}/{config['database']}")
        
        response = requests.post(
            SWITCH_ENDPOINT,
            json=config,
            timeout=30  # 30 second timeout for schema generation
        )
        
        if response.status_code != 200:
            print(f"[TEST] ❌ API call failed with status {response.status_code}")
            try:
                error_data = response.json()
                print(f"[TEST] Error details: {error_data}")
            except:
                print(f"[TEST] Raw error: {response.text}")
            return False
        
        data = response.json()
        
        # Step 2: Validate response structure
        if data.get("status") != "success":
            print(f"[TEST] ❌ Switch failed: {data.get('message', 'Unknown error')}")
            return False
        
        # Step 3: Print success details
        db_info = data.get("database_info", {})
        table_count = db_info.get("table_count", 0)
        
        print(f"[TEST] ✅ Switch successful!")
        print(f"[TEST] 📊 Database: {db_info.get('dialect', 'N/A')}://{db_info.get('host', 'N/A')}:{db_info.get('port', 'N/A')}/{db_info.get('database', 'N/A')}")
        print(f"[TEST] 📋 Tables found: {table_count}")
        print(f"[TEST] 📄 Schema file: {data.get('schema_file', 'N/A')}")
        print(f"[TEST] 🕒 Timestamp: {data.get('timestamp', 'N/A')}")
        
        # Step 4: Verify schema file exists
        schema_file = data.get('schema_file')
        if schema_file and os.path.exists(schema_file):
            print(f"[TEST] ✅ Schema file verified: {schema_file}")
            
            # Show a few table names
            tables = data.get('tables', {})
            if tables:
                table_names = list(tables.keys())[:5]  # First 5 tables
                print(f"[TEST] 📋 Sample tables: {', '.join(table_names)}")
                if len(tables) > 5:
                    print(f"[TEST] 📋 ... and {len(tables) - 5} more")
        else:
            print(f"[TEST] ⚠️ Schema file not found: {schema_file}")
        
        # Step 5: Test the /tables endpoint to ensure it works
        print(f"[TEST] 🔍 Testing /tables endpoint...")
        tables_response = requests.get(TABLES_ENDPOINT, timeout=10)
        
        if tables_response.status_code == 200:
            tables_data = tables_response.json()
            print(f"[TEST] ✅ /tables endpoint working - returned {len(tables_data)} tables")
        else:
            print(f"[TEST] ⚠️ /tables endpoint failed with status {tables_response.status_code}")
        
        return True
        
    except requests.exceptions.Timeout:
        print(f"[TEST] ❌ Request timed out - database switch might be taking too long")
        return False
    except requests.exceptions.ConnectionError:
        print(f"[TEST] ❌ Connection error - is the Python service running on {API_BASE}?")
        return False
    except Exception as e:
        print(f"[TEST] ❌ Unexpected error: {str(e)}")
        return False

def check_service_health() -> bool:
    """Check if the API service is running"""
    try:
        # Try to reach the base service
        response = requests.get(f"{API_BASE}/", timeout=5)
        if response.status_code == 200:
            print(f"[HEALTH] ✅ Python service is running at {API_BASE}")
            return True
        else:
            print(f"[HEALTH] ⚠️ Python service responded with status {response.status_code}")
            return True  # Service is running but might have different root endpoint
    except requests.exceptions.ConnectionError:
        print(f"[HEALTH] ❌ Cannot connect to Python service at {API_BASE}")
        print(f"[HEALTH] Make sure to start the service with: python backend/excel_service.py")
        return False
    except Exception as e:
        print(f"[HEALTH] ❌ Health check error: {str(e)}")
        return False

def main():
    print("🚀 STARTING DATABASE SWITCHING API TESTS")
    print("="*80)
    
    # Check service health first
    if not check_service_health():
        print("\n❌ TESTS ABORTED - Service not available")
        return False
    
    # Test switching between databases
    results = []
    
    # Test 1: Switch to MySQL
    results.append(test_database_switch(MYSQL_CONFIG, "MySQL BookWedGo"))
    time.sleep(2)  # Brief pause between tests
    
    # Test 2: Switch to PostgreSQL  
    results.append(test_database_switch(POSTGRES_CONFIG, "PostgreSQL CloseBi"))
    time.sleep(2)  # Brief pause between tests
    
    # Test 3: Switch back to MySQL
    results.append(test_database_switch(MYSQL_CONFIG, "MySQL BookWedGo (Again)"))
    
    # Results summary
    print("\n" + "="*80)
    print("📊 TEST RESULTS SUMMARY")
    print("="*80)
    
    total_tests = len(results)
    passed_tests = sum(results)
    
    for i, result in enumerate(results, 1):
        status = "✅ PASS" if result else "❌ FAIL"
        test_names = ["MySQL BookWedGo", "PostgreSQL CloseBi", "MySQL BookWedGo (Again)"]
        print(f"Test {i}: {test_names[i-1]} - {status}")
    
    print(f"\nOverall: {passed_tests}/{total_tests} tests passed")
    
    if passed_tests == total_tests:
        print("🎉 ALL TESTS PASSED! Database switching is working perfectly!")
        return True
    else:
        print("⚠️ Some tests failed. Check the logs above for details.")
        return False

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)