#!/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())