""" Integration tests for Visualization Agent HTTP endpoint Tests the FastAPI endpoint with various request scenarios """ import os import sys import json from unittest.mock import Mock, patch from fastapi.testclient import TestClient # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # Set test API key before importing main os.environ["GEMINI_API_KEY"] = "test_api_key_for_http_tests" from main import app from models import RiskData, HazardData, SeismicHazards, VolcanicHazards, HydroHazards from models import HazardDetail, LocationInfo, Coordinates, RiskSummary, FacilityInfo, Metadata # Create test client client = TestClient(app) def create_test_risk_data_dict(): """Create a test risk data dictionary for HTTP requests""" return { "success": True, "summary": { "overall_risk_level": "HIGH", "total_hazards_assessed": 10, "high_risk_count": 2, "moderate_risk_count": 3, "critical_hazards": ["Active Fault", "Ground Shaking"] }, "location": { "name": "Manila", "coordinates": { "latitude": 14.5995, "longitude": 120.9842 }, "administrative_area": "Metro Manila" }, "hazards": { "seismic": { "active_fault": {"status": "PRESENT", "description": "Active fault present"}, "ground_shaking": {"status": "HIGH", "description": "High ground shaking risk"}, "liquefaction": {"status": "MODERATE", "description": "Moderate liquefaction risk"}, "tsunami": {"status": "NONE", "description": "No tsunami risk"}, "earthquake_induced_landslide": {"status": "NONE", "description": "No landslide risk"}, "fissure": {"status": "NONE", "description": "No fissure risk"}, "ground_rupture": {"status": "NONE", "description": "No ground rupture risk"} }, "volcanic": { "active_volcano": {"status": "NONE", "description": "No active volcano"}, "potentially_active_volcano": {"status": "NONE", "description": "No potentially active volcano"}, "inactive_volcano": {"status": "NONE", "description": "No inactive volcano"}, "ashfall": {"status": "NONE", "description": "No ashfall risk"}, "pyroclastic_flow": {"status": "NONE", "description": "No pyroclastic flow risk"}, "lahar": {"status": "NONE", "description": "No lahar risk"}, "lava": {"status": "NONE", "description": "No lava risk"}, "ballistic_projectile": {"status": "NONE", "description": "No ballistic projectile risk"}, "base_surge": {"status": "NONE", "description": "No base surge risk"}, "volcanic_tsunami": {"status": "NONE", "description": "No volcanic tsunami risk"} }, "hydrometeorological": { "flood": {"status": "MODERATE", "description": "Moderate flood risk"}, "rain_induced_landslide": {"status": "NONE", "description": "No rain-induced landslide risk"}, "storm_surge": {"status": "NONE", "description": "No storm surge risk"}, "severe_winds": {"status": "NONE", "description": "No severe wind risk"} } }, "facilities": { "schools": [], "hospitals": [], "road_networks": [] }, "metadata": { "timestamp": "2024-01-01T00:00:00Z", "source": "test", "cache_status": "fresh", "ttl": 3600 } } def test_health_check(): """Test health check endpoint""" print("\n=== Testing Health Check Endpoint ===") try: response = client.get("/health") assert response.status_code == 200, f"Expected 200, got {response.status_code}" data = response.json() assert data["status"] == "healthy", "Health check status not healthy" assert data["agent"] == "visualization-agent", "Wrong agent name" print("✓ Health check endpoint working") return True except Exception as e: print(f"✗ Health check test failed: {e}") return False def test_valid_request_payload(): """Test endpoint with valid request payload""" print("\n=== Testing Valid Request Payload ===") try: # Create mock response mock_image = Mock() mock_image.image.data = b"test_image_data_http" mock_response = Mock() mock_response.generated_images = [mock_image] # Mock the Gemini client with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.return_value = mock_response MockClient.return_value = mock_client_instance # Create request payload payload = { "risk_data": create_test_risk_data_dict(), "building_type": "residential_single_family", "recommendations": None } # Make request response = client.post("/", json=payload) assert response.status_code == 200, f"Expected 200, got {response.status_code}" data = response.json() assert data["success"] is True, "Request should succeed" assert "visualization_data" in data, "Missing visualization_data" viz_data = data["visualization_data"] assert "image_base64" in viz_data, "Missing image_base64" assert "prompt_used" in viz_data, "Missing prompt_used" assert "model_version" in viz_data, "Missing model_version" assert "generation_timestamp" in viz_data, "Missing generation_timestamp" assert "features_included" in viz_data, "Missing features_included" print("✓ Valid request processed successfully") print(f" Status: {response.status_code}") print(f" Features: {len(viz_data['features_included'])}") return True except Exception as e: print(f"✗ Valid request test failed: {e}") return False def test_generate_endpoint(): """Test /generate endpoint (alternative path)""" print("\n=== Testing /generate Endpoint ===") try: # Create mock response mock_image = Mock() mock_image.image.data = b"test_image_generate_endpoint" mock_response = Mock() mock_response.generated_images = [mock_image] with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.return_value = mock_response MockClient.return_value = mock_client_instance payload = { "risk_data": create_test_risk_data_dict(), "building_type": "commercial_office" } response = client.post("/generate", json=payload) assert response.status_code == 200, f"Expected 200, got {response.status_code}" assert response.json()["success"] is True, "Request should succeed" print("✓ /generate endpoint working") return True except Exception as e: print(f"✗ /generate endpoint test failed: {e}") return False def test_invalid_payload_missing_fields(): """Test endpoint with invalid payload (missing required fields)""" print("\n=== Testing Invalid Payload: Missing Fields ===") try: # Missing building_type payload = { "risk_data": create_test_risk_data_dict() } response = client.post("/", json=payload) # Should return 422 for validation error assert response.status_code == 422, f"Expected 422, got {response.status_code}" print("✓ Missing fields handled correctly") print(f" Status: {response.status_code}") return True except Exception as e: print(f"✗ Invalid payload test failed: {e}") return False def test_invalid_payload_wrong_types(): """Test endpoint with invalid payload (wrong data types)""" print("\n=== Testing Invalid Payload: Wrong Types ===") try: # Wrong type for building_type (should be string) payload = { "risk_data": create_test_risk_data_dict(), "building_type": 12345 # Should be string } response = client.post("/", json=payload) # Should return 422 for validation error assert response.status_code == 422, f"Expected 422, got {response.status_code}" print("✓ Wrong types handled correctly") return True except Exception as e: print(f"✗ Wrong types test failed: {e}") return False def test_invalid_risk_data_structure(): """Test endpoint with invalid risk data structure""" print("\n=== Testing Invalid Risk Data Structure ===") try: # Invalid risk data (missing required fields) payload = { "risk_data": { "success": True, # Missing other required fields }, "building_type": "residential_single_family" } response = client.post("/", json=payload) # Should return error (either 400 or 422) assert response.status_code in [400, 422, 500], f"Expected error status, got {response.status_code}" print("✓ Invalid risk data structure handled") print(f" Status: {response.status_code}") return True except Exception as e: print(f"✗ Invalid risk data test failed: {e}") return False def test_endpoint_error_responses(): """Test endpoint error responses for API failures""" print("\n=== Testing Endpoint Error Responses ===") try: # Mock API failure with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.side_effect = Exception("API Error: Service unavailable") MockClient.return_value = mock_client_instance payload = { "risk_data": create_test_risk_data_dict(), "building_type": "institutional_school" } response = client.post("/", json=payload) # Should return 200 with success=False in body (agent pattern) assert response.status_code == 200, f"Expected 200, got {response.status_code}" data = response.json() assert data["success"] is False, "Should indicate failure" assert "error" in data, "Missing error field" print("✓ Error responses formatted correctly") print(f" Error code: {data['error']['code']}") return True except Exception as e: print(f"✗ Error response test failed: {e}") return False def test_response_format_compatibility(): """Test response format compatibility with agent pattern""" print("\n=== Testing Response Format Compatibility ===") try: # Create mock response mock_image = Mock() mock_image.image.data = b"compatibility_test_image" mock_response = Mock() mock_response.generated_images = [mock_image] with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.return_value = mock_response MockClient.return_value = mock_client_instance payload = { "risk_data": create_test_risk_data_dict(), "building_type": "mixed_use" } response = client.post("/", json=payload) data = response.json() # Check standard agent response format assert "success" in data, "Missing success field" assert isinstance(data["success"], bool), "success should be boolean" if data["success"]: assert "visualization_data" in data, "Missing visualization_data on success" assert data.get("error") is None, "error should be None on success" else: assert "error" in data, "Missing error on failure" assert data.get("visualization_data") is None, "visualization_data should be None on failure" print("✓ Response format compatible with agent pattern") return True except Exception as e: print(f"✗ Response format test failed: {e}") return False def test_with_recommendations(): """Test endpoint with recommendations included""" print("\n=== Testing With Recommendations ===") try: # Create mock response mock_image = Mock() mock_image.image.data = b"recommendations_test_image" mock_response = Mock() mock_response.generated_images = [mock_image] with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.return_value = mock_response MockClient.return_value = mock_client_instance payload = { "risk_data": create_test_risk_data_dict(), "building_type": "commercial_retail", "recommendations": { "general_guidelines": ["Use reinforced concrete"], "seismic_recommendations": [], "volcanic_recommendations": [], "hydrometeorological_recommendations": [], "priority_actions": [], "building_codes": [] } } response = client.post("/", json=payload) assert response.status_code == 200, f"Expected 200, got {response.status_code}" assert response.json()["success"] is True, "Request should succeed" print("✓ Recommendations processed correctly") return True except Exception as e: print(f"✗ Recommendations test failed: {e}") return False def test_all_building_types(): """Test endpoint with all building types""" print("\n=== Testing All Building Types ===") try: building_types = [ "residential_single_family", "residential_multi_family", "residential_high_rise", "commercial_office", "commercial_retail", "industrial_warehouse", "institutional_school", "institutional_hospital", "infrastructure_bridge", "mixed_use" ] # Create mock response mock_image = Mock() mock_image.image.data = b"building_type_test_image" mock_response = Mock() mock_response.generated_images = [mock_image] with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.return_value = mock_response MockClient.return_value = mock_client_instance for building_type in building_types: payload = { "risk_data": create_test_risk_data_dict(), "building_type": building_type } response = client.post("/", json=payload) assert response.status_code == 200, f"Failed for {building_type}" assert response.json()["success"] is True, f"Failed for {building_type}" print(f" ✓ {building_type}") print("✓ All building types processed successfully") return True except Exception as e: print(f"✗ All building types test failed: {e}") return False def test_concurrent_requests(): """Test handling of multiple concurrent requests""" print("\n=== Testing Concurrent Requests ===") try: import concurrent.futures # Create mock response mock_image = Mock() mock_image.image.data = b"concurrent_test_image" mock_response = Mock() mock_response.generated_images = [mock_image] with patch('agent.genai.Client') as MockClient: mock_client_instance = Mock() mock_client_instance.models.generate_images.return_value = mock_response MockClient.return_value = mock_client_instance def make_request(i): payload = { "risk_data": create_test_risk_data_dict(), "building_type": "residential_single_family" } response = client.post("/", json=payload) return response.status_code == 200 # Make 5 concurrent requests with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(make_request, i) for i in range(5)] results = [f.result() for f in concurrent.futures.as_completed(futures)] assert all(results), "Some concurrent requests failed" print("✓ Concurrent requests handled correctly") print(f" Processed {len(results)} concurrent requests") return True except Exception as e: print(f"✗ Concurrent requests test failed: {e}") return False if __name__ == "__main__": print("=" * 60) print("Visualization Agent HTTP Endpoint Tests") print("=" * 60) # Run all tests test_results = [] test_results.append(("Health Check", test_health_check())) test_results.append(("Valid Request", test_valid_request_payload())) test_results.append(("/generate Endpoint", test_generate_endpoint())) test_results.append(("Missing Fields", test_invalid_payload_missing_fields())) test_results.append(("Wrong Types", test_invalid_payload_wrong_types())) test_results.append(("Invalid Risk Data", test_invalid_risk_data_structure())) test_results.append(("Error Responses", test_endpoint_error_responses())) test_results.append(("Response Format", test_response_format_compatibility())) test_results.append(("With Recommendations", test_with_recommendations())) test_results.append(("All Building Types", test_all_building_types())) test_results.append(("Concurrent Requests", test_concurrent_requests())) # Summary print("\n" + "=" * 60) print("TEST SUMMARY") print("=" * 60) for test_name, result in test_results: status = "✅ PASS" if result else "❌ FAIL" print(f" {status}: {test_name}") passed = sum(1 for _, result in test_results if result) total = len(test_results) print(f"\n{'=' * 60}") print(f"Overall: {passed}/{total} tests passed") print("=" * 60) if passed == total: print("\n✅ All HTTP endpoint tests passed!") sys.exit(0) else: print(f"\n❌ {total - passed} test(s) failed") sys.exit(1)