#!/usr/bin/env python3 """ Comprehensive Backend API Testing for Aera Risk Detection System Tests all endpoints with proper authentication and error handling """ import requests import sys import json from datetime import datetime class AeraAPITester: def __init__(self, base_url="https://teams-risk.preview.emergentagent.com"): self.base_url = base_url self.token = None self.tests_run = 0 self.tests_passed = 0 self.failed_tests = [] def log(self, message, level="INFO"): """Log test messages with timestamp""" timestamp = datetime.now().strftime("%H:%M:%S") print(f"[{timestamp}] {level}: {message}") def run_test(self, name, method, endpoint, expected_status, data=None, headers=None, auth_required=True, timeout=30): """Run a single API test""" url = f"{self.base_url}/api/{endpoint}" test_headers = {'Content-Type': 'application/json'} if headers: test_headers.update(headers) if auth_required and self.token: test_headers['Authorization'] = f'Bearer {self.token}' self.tests_run += 1 self.log(f"Testing {name}...") try: if method == 'GET': response = requests.get(url, headers=test_headers, timeout=timeout) elif method == 'POST': response = requests.post(url, json=data, headers=test_headers, timeout=timeout) elif method == 'PUT': response = requests.put(url, json=data, headers=test_headers, timeout=timeout) elif method == 'DELETE': response = requests.delete(url, headers=test_headers, timeout=timeout) success = response.status_code == expected_status if success: self.tests_passed += 1 self.log(f"✅ {name} - Status: {response.status_code}", "PASS") try: return True, response.json() if response.content else {} except: return True, {} else: self.failed_tests.append({ "test": name, "expected": expected_status, "actual": response.status_code, "response": response.text[:200] if response.text else "No response" }) self.log(f"❌ {name} - Expected {expected_status}, got {response.status_code}", "FAIL") self.log(f" Response: {response.text[:200]}", "ERROR") return False, {} except requests.exceptions.RequestException as e: self.failed_tests.append({ "test": name, "expected": expected_status, "actual": "REQUEST_ERROR", "response": str(e) }) self.log(f"❌ {name} - Request Error: {str(e)}", "ERROR") return False, {} def test_health_check(self): """Test health endpoint""" return self.run_test( "Health Check", "GET", "health", 200, auth_required=False ) def test_root_endpoint(self): """Test root API endpoint""" return self.run_test( "Root API", "GET", "", 200, auth_required=False ) def test_login_admin(self): """Test admin login""" success, response = self.run_test( "Admin Login", "POST", "auth/login/json", 200, data={"email": "admin@aera.com", "password": "Admin123!"}, auth_required=False ) if success and 'access_token' in response: self.token = response['access_token'] self.log("✅ Admin token acquired", "SUCCESS") return True return False def test_login_reviewer(self): """Test reviewer login""" success, response = self.run_test( "Reviewer Login", "POST", "auth/login/json", 200, data={"email": "reviewer@aera.com", "password": "Reviewer123!"}, auth_required=False ) if success and 'access_token' in response: # Store reviewer token for later use self.reviewer_token = response['access_token'] self.log("✅ Reviewer token acquired", "SUCCESS") return True return False def test_invalid_login(self): """Test invalid login credentials""" return self.run_test( "Invalid Login", "POST", "auth/login/json", 401, data={"email": "invalid@test.com", "password": "wrongpass"}, auth_required=False ) def test_get_current_user(self): """Test get current user info""" return self.run_test( "Get Current User", "GET", "auth/me", 200 ) def test_dashboard_stats(self): """Test dashboard statistics""" return self.run_test( "Dashboard Stats", "GET", "dashboard/stats", 200 ) def test_dashboard_trends(self): """Test dashboard trends""" return self.run_test( "Dashboard Trends", "GET", "dashboard/trends", 200 ) def test_recent_alerts(self): """Test recent alerts""" return self.run_test( "Recent Alerts", "GET", "dashboard/recent-alerts?limit=5", 200 ) def test_sync_teams(self): """Test Teams sync functionality""" success, response = self.run_test( "Sync Teams Data", "POST", "sync", 200, timeout=60 # Sync can take longer due to AI processing ) if success: self.log(f" Synced: {response.get('conversations_synced', 0)} conversations", "INFO") self.log(f" Risks detected: {response.get('risks_detected', 0)}", "INFO") return success def test_get_risks(self): """Test get flagged risks""" return self.run_test( "Get Flagged Risks", "GET", "risks?limit=10", 200, timeout=30 # Risk queries can be slow ) def test_get_risks_with_filters(self): """Test get risks with filters""" success1, _ = self.run_test( "Get High Risk Messages", "GET", "risks?risk_level=HIGH&limit=5", 200, timeout=30 ) success2, _ = self.run_test( "Get Pending Reviews", "GET", "risks?status=pending&limit=5", 200, timeout=30 ) return success1 and success2 def test_risk_detail(self): """Test get specific risk detail""" # First get risks to find a valid risk ID success, risks_response = self.run_test( "Get Risks for Detail Test", "GET", "risks?limit=1", 200 ) if success and risks_response and len(risks_response) > 0: risk_id = risks_response[0].get('id') if risk_id: return self.run_test( "Get Risk Detail", "GET", f"risks/{risk_id}", 200 ) self.log("⚠️ No risks available for detail test", "WARN") return True # Not a failure if no risks exist def test_submit_review(self): """Test submitting a review action""" # First get a risk to review success, risks_response = self.run_test( "Get Risks for Review Test", "GET", "risks?status=pending&limit=1", 200 ) if success and risks_response and len(risks_response) > 0: risk_id = risks_response[0].get('id') if risk_id: return self.run_test( "Submit Review Action", "POST", f"risks/{risk_id}/review", 200, data={ "risk_analysis_id": risk_id, # Include risk_analysis_id in body as expected by model "action": "Monitor", "notes": "Test review action from automated testing" } ) self.log("⚠️ No pending risks available for review test", "WARN") return True # Not a failure if no pending risks def test_chatbot_query(self): """Test chatbot query functionality""" return self.run_test( "Chatbot Query", "POST", "chatbot/query", 200, data={"query": "How many high-risk conversations have been detected?"} ) def test_audit_logs_admin_only(self): """Test audit logs (admin only)""" if not hasattr(self, 'reviewer_token'): self.log("⚠️ Reviewer token not available for audit test", "WARN") return True # Test with reviewer token (should fail) old_token = self.token self.token = self.reviewer_token success_reviewer, _ = self.run_test( "Audit Logs (Reviewer - Should Fail)", "GET", "audit/logs?limit=5", 403 # Should be forbidden for reviewer ) # Test with admin token (should succeed) self.token = old_token success_admin, _ = self.run_test( "Audit Logs (Admin)", "GET", "audit/logs?limit=5", 200 ) return success_reviewer and success_admin def test_register_new_user(self): """Test user registration""" test_email = f"test_user_{datetime.now().strftime('%H%M%S')}@test.com" return self.run_test( "Register New User", "POST", "auth/register", 200, data={ "email": test_email, "password": "TestPass123!", "full_name": "Test User", "role": "Reviewer" }, auth_required=False ) def run_all_tests(self): """Run all API tests in sequence""" self.log("🚀 Starting Aera Risk Detection API Tests", "START") self.log(f"Testing against: {self.base_url}", "INFO") # Basic connectivity tests self.test_health_check() self.test_root_endpoint() # Authentication tests if not self.test_login_admin(): self.log("❌ Admin login failed - stopping tests", "CRITICAL") return False self.test_login_reviewer() self.test_invalid_login() self.test_get_current_user() # Dashboard tests self.test_dashboard_stats() self.test_dashboard_trends() self.test_recent_alerts() # Sync functionality self.test_sync_teams() # Risk management tests self.test_get_risks() self.test_get_risks_with_filters() self.test_risk_detail() self.test_submit_review() # Chatbot tests self.test_chatbot_query() # Admin-only features self.test_audit_logs_admin_only() # User management self.test_register_new_user() return True def print_summary(self): """Print test summary""" self.log("=" * 60, "SUMMARY") self.log(f"Tests Run: {self.tests_run}", "SUMMARY") self.log(f"Tests Passed: {self.tests_passed}", "SUMMARY") self.log(f"Tests Failed: {len(self.failed_tests)}", "SUMMARY") self.log(f"Success Rate: {(self.tests_passed/self.tests_run*100):.1f}%", "SUMMARY") if self.failed_tests: self.log("\n❌ Failed Tests:", "SUMMARY") for test in self.failed_tests: self.log(f" • {test['test']}: Expected {test['expected']}, got {test['actual']}", "FAIL") if test['response']: self.log(f" Response: {test['response']}", "FAIL") return len(self.failed_tests) == 0 def main(): """Main test execution""" tester = AeraAPITester() try: success = tester.run_all_tests() all_passed = tester.print_summary() if all_passed: tester.log("🎉 All tests passed!", "SUCCESS") return 0 else: tester.log("💥 Some tests failed!", "ERROR") return 1 except KeyboardInterrupt: tester.log("Tests interrupted by user", "WARN") return 1 except Exception as e: tester.log(f"Unexpected error: {str(e)}", "ERROR") return 1 if __name__ == "__main__": sys.exit(main())