backend_test.py

#!/usr/bin/env python3
"""
NeuralVault Backend API Testing
Tests all API endpoints with real Gemini integration
"""
import requests
import sys
import json
import time
from datetime import datetime

class NeuralVaultAPITester:
    def __init__(self, base_url="http://localhost:8000"):
        self.base_url = base_url
        self.tests_run = 0
        self.tests_passed = 0
        self.failed_tests = []

    def run_test(self, name, method, endpoint, expected_status, data=None, timeout=30):
        """Run a single API test"""
        url = f"{self.base_url}/api/{endpoint}"
        headers = {'Content-Type': 'application/json'}

        self.tests_run += 1
        print(f"\n🔍 Testing {name}...")
        print(f"   URL: {url}")
        
        try:
            start_time = time.time()
            if method == 'GET':
                response = requests.get(url, headers=headers, timeout=timeout)
            elif method == 'POST':
                response = requests.post(url, json=data, headers=headers, timeout=timeout)
            
            elapsed = time.time() - start_time
            
            success = response.status_code == expected_status
            if success:
                self.tests_passed += 1
                print(f"✅ Passed - Status: {response.status_code} ({elapsed:.2f}s)")
                try:
                    response_data = response.json()
                    if 'ok' in response_data:
                        print(f"   Response OK: {response_data['ok']}")
                    return True, response_data
                except:
                    return True, {}
            else:
                print(f"❌ Failed - Expected {expected_status}, got {response.status_code}")
                print(f"   Response: {response.text[:200]}...")
                self.failed_tests.append({
                    "test": name,
                    "endpoint": endpoint,
                    "expected": expected_status,
                    "actual": response.status_code,
                    "response": response.text[:200]
                })
                return False, {}

        except requests.exceptions.Timeout:
            print(f"❌ Failed - Timeout after {timeout}s")
            self.failed_tests.append({
                "test": name,
                "endpoint": endpoint,
                "error": f"Timeout after {timeout}s"
            })
            return False, {}
        except Exception as e:
            print(f"❌ Failed - Error: {str(e)}")
            self.failed_tests.append({
                "test": name,
                "endpoint": endpoint,
                "error": str(e)
            })
            return False, {}

    def test_basic_endpoints(self):
        """Test basic API endpoints"""
        print("\n=== Testing Basic Endpoints ===")
        
        # Test root endpoint
        self.run_test("API Root", "GET", "", 200)
        
        # Test health endpoint
        self.run_test("Health Check", "GET", "health", 200)

    def test_nl2sql_endpoint(self):
        """Test Natural Language to SQL endpoint"""
        print("\n=== Testing NL2SQL Endpoint ===")
        
        test_queries = [
            "Show me all customers with high lifetime value",
            "Find reviews with negative sentiment",
            "Get top 10 products by category"
        ]
        
        for query in test_queries:
            success, response = self.run_test(
                f"NL2SQL: {query[:30]}...",
                "POST",
                "demo/nl2sql",
                200,
                data={"query": query},
                timeout=20  # Gemini calls can take time
            )
            if success and response.get('ok'):
                print(f"   SQL Generated: {response.get('sql', 'N/A')[:50]}...")
                print(f"   Explanation: {response.get('explanation', 'N/A')[:50]}...")

    def test_trigger_simulator(self):
        """Test AI Trigger Simulator endpoint"""
        print("\n=== Testing Trigger Simulator ===")
        
        # Test review sentiment analysis
        success, response = self.run_test(
            "Trigger: Review Sentiment",
            "POST",
            "demo/trigger",
            200,
            data={
                "entity_type": "review",
                "text": "This product is absolutely terrible and broke after one day"
            },
            timeout=20
        )
        if success and response.get('ok'):
            print(f"   Analysis: {response.get('data', {})}")

        # Test product classification
        success, response = self.run_test(
            "Trigger: Product Classification",
            "POST",
            "demo/trigger",
            200,
            data={
                "entity_type": "product",
                "product_name": "iPhone 15 Pro",
                "product_description": "Latest smartphone with advanced camera"
            },
            timeout=20
        )
        if success and response.get('ok'):
            print(f"   Classification: {response.get('data', {})}")

        # Test transaction fraud analysis
        success, response = self.run_test(
            "Trigger: Transaction Fraud",
            "POST",
            "demo/trigger",
            200,
            data={
                "entity_type": "transaction",
                "amount": 9999.99,
                "customer_id": "cust_suspicious",
                "merchant": "Unknown Store",
                "location": "Foreign Country"
            },
            timeout=20
        )
        if success and response.get('ok'):
            print(f"   Fraud Analysis: {response.get('data', {})}")

    def test_hybrid_search(self):
        """Test Hybrid Search endpoint"""
        print("\n=== Testing Hybrid Search ===")
        
        search_queries = [
            {"query": "wireless headphones", "search_mode": "hybrid"},
            {"query": "laptop computer", "search_mode": "vector"},
            {"query": "smartphone", "search_mode": "fulltext"}
        ]
        
        for search_data in search_queries:
            success, response = self.run_test(
                f"Hybrid Search: {search_data['query']} ({search_data['search_mode']})",
                "POST",
                "demo/hybrid-search",
                200,
                data=search_data,
                timeout=20
            )
            if success and response.get('ok'):
                results = response.get('results', [])
                print(f"   Found {len(results)} results")
                if results:
                    print(f"   Top result: {results[0].get('name', 'N/A')}")

    def test_error_handling(self):
        """Test error handling with invalid requests"""
        print("\n=== Testing Error Handling ===")
        
        # Test empty query for NL2SQL
        self.run_test(
            "NL2SQL: Empty Query",
            "POST",
            "demo/nl2sql",
            422,  # Validation error
            data={"query": ""}
        )
        
        # Test invalid entity type for trigger
        self.run_test(
            "Trigger: Invalid Entity",
            "POST",
            "demo/trigger",
            422,  # Validation error
            data={"entity_type": "invalid_type"}
        )
        
        # Test empty search query
        self.run_test(
            "Hybrid Search: Empty Query",
            "POST",
            "demo/hybrid-search",
            422,  # Validation error
            data={"query": ""}
        )

def main():
    print("🧠 NeuralVault Backend API Testing")
    print("=" * 50)
    
    tester = NeuralVaultAPITester()
    
    # Run all tests
    tester.test_basic_endpoints()
    tester.test_nl2sql_endpoint()
    tester.test_trigger_simulator()
    tester.test_hybrid_search()
    tester.test_error_handling()
    
    # Print summary
    print(f"\n📊 Test Summary")
    print("=" * 30)
    print(f"Tests run: {tester.tests_run}")
    print(f"Tests passed: {tester.tests_passed}")
    print(f"Tests failed: {len(tester.failed_tests)}")
    print(f"Success rate: {(tester.tests_passed/tester.tests_run)*100:.1f}%")
    
    if tester.failed_tests:
        print(f"\n❌ Failed Tests:")
        for failure in tester.failed_tests:
            error_msg = failure.get('error', f"Status {failure.get('actual')} != {failure.get('expected')}")
            print(f"  - {failure['test']}: {error_msg}")
    
    return 0 if len(tester.failed_tests) == 0 else 1

if __name__ == "__main__":
    sys.exit(main())