""" Module: tests.integration.test_basic_api Description: Basic integration tests for core API endpoints Author: Anderson H. Silva Date: 2025-01-24 License: Proprietary - All rights reserved """ import pytest from fastapi.testclient import TestClient from httpx import AsyncClient from unittest.mock import patch # Import just the FastAPI app without triggering full agent imports import sys from unittest.mock import MagicMock # Mock heavy dependencies before importing sys.modules['numpy'] = MagicMock() sys.modules['scikit-learn'] = MagicMock() sys.modules['torch'] = MagicMock() sys.modules['transformers'] = MagicMock() from src.api.app import app # Test client for synchronous tests client = TestClient(app) @pytest.fixture async def async_client(): """Async test client for async endpoints.""" async with AsyncClient(app=app, base_url="http://test") as ac: yield ac class TestBasicAPIFunctionality: """Test basic API functionality without heavy dependencies.""" def test_root_endpoint(self): """Test root API endpoint.""" response = client.get("/") assert response.status_code == 200 data = response.json() assert "message" in data assert "version" in data assert "description" in data assert data["status"] == "operational" def test_api_info_endpoint(self): """Test API information endpoint.""" response = client.get("/api/v1/info") assert response.status_code == 200 data = response.json() assert "api" in data assert "agents" in data assert "data_sources" in data assert "formats" in data # Verify API info structure api_info = data["api"] assert api_info["name"] == "Cidadão.AI API" assert api_info["version"] == "1.0.0" # Verify agent information agents = data["agents"] assert "investigator" in agents assert "analyst" in agents assert "reporter" in agents # Check agent capabilities investigator = agents["investigator"] assert "description" in investigator assert "capabilities" in investigator assert isinstance(investigator["capabilities"], list) def test_openapi_schema_generation(self): """Test OpenAPI schema generation.""" response = client.get("/openapi.json") assert response.status_code == 200 schema = response.json() assert "openapi" in schema assert "info" in schema assert "paths" in schema assert "components" in schema # Verify API metadata info = schema["info"] assert info["title"] == "Cidadão.AI API" assert info["version"] == "1.0.0" # Verify some expected paths exist paths = schema["paths"] assert "/" in paths assert "/api/v1/info" in paths def test_docs_endpoint(self): """Test API documentation endpoint.""" response = client.get("/docs") assert response.status_code == 200 # Should return HTML content assert "text/html" in response.headers.get("content-type", "") def test_health_endpoint_basic(self): """Test basic health check endpoint.""" response = client.get("/health/") assert response.status_code == 200 data = response.json() assert "status" in data assert data["status"] in ["healthy", "degraded", "unhealthy"] assert "timestamp" in data assert "version" in data assert "uptime" in data assert "services" in data def test_health_liveness_probe(self): """Test Kubernetes liveness probe.""" response = client.get("/health/live") assert response.status_code == 200 data = response.json() assert data["status"] == "alive" assert "timestamp" in data def test_health_readiness_probe(self): """Test Kubernetes readiness probe.""" response = client.get("/health/ready") # Should return 200 or 503 depending on services assert response.status_code in [200, 503] data = response.json() assert "status" in data assert data["status"] in ["ready", "not_ready"] class TestAPIErrorHandling: """Test API error handling and edge cases.""" def test_404_handling(self): """Test 404 error handling for non-existent endpoints.""" response = client.get("/api/v1/nonexistent-endpoint") assert response.status_code == 404 data = response.json() assert "detail" in data or "message" in data def test_method_not_allowed(self): """Test 405 method not allowed.""" response = client.patch("/") # Wrong method for root endpoint assert response.status_code == 405 def test_large_payload_handling(self): """Test handling of very large payloads.""" large_data = {"data": "x" * 1000} # 1KB payload # Try posting to an endpoint that should exist response = client.post("/api/v1/info", json=large_data) # Should handle gracefully (405 for wrong method, or other appropriate error) assert response.status_code in [405, 422, 413] def test_invalid_json_payload(self): """Test handling of invalid JSON payloads.""" response = client.post( "/api/v1/info", data="invalid json content", headers={"Content-Type": "application/json"} ) # Should return 422 for invalid JSON assert response.status_code == 422 class TestAPICORSAndSecurity: """Test CORS and security configurations.""" def test_cors_preflight_handling(self): """Test CORS preflight request handling.""" headers = { "Origin": "http://localhost:3000", "Access-Control-Request-Method": "POST", "Access-Control-Request-Headers": "Content-Type" } response = client.options("/api/v1/info", headers=headers) # Should handle CORS preflight (200 or 204) assert response.status_code in [200, 204] def test_cors_headers_in_response(self): """Test that CORS headers are included in responses.""" headers = {"Origin": "http://localhost:3000"} response = client.get("/", headers=headers) assert response.status_code == 200 # CORS headers might be added by middleware # In test environment, they might not be present def test_trusted_host_validation(self): """Test trusted host middleware behavior.""" # Test with potentially malicious host header headers = {"Host": "malicious-site.example.com"} response = client.get("/", headers=headers) # Should either accept (if not configured) or reject assert response.status_code in [200, 400, 403] def test_security_headers_present(self): """Test that basic security headers are present.""" response = client.get("/") headers = response.headers # Basic security check - should have content-type assert "content-type" in headers # Additional security headers might be added by middleware # In production: X-Content-Type-Options, X-Frame-Options, etc. class TestAPIPerformance: """Test basic API performance characteristics.""" def test_response_time_reasonable(self): """Test that basic endpoints respond within reasonable time.""" import time start_time = time.time() response = client.get("/") end_time = time.time() response_time = end_time - start_time assert response.status_code == 200 assert response_time < 2.0 # Should respond within 2 seconds def test_concurrent_requests_handling(self): """Test basic concurrent request handling.""" import threading import time results = [] def make_request(): response = client.get("/") results.append(response.status_code) # Create 3 concurrent requests threads = [] for _ in range(3): thread = threading.Thread(target=make_request) threads.append(thread) thread.start() # Wait for all to complete for thread in threads: thread.join() # All should succeed assert len(results) == 3 assert all(status == 200 for status in results) def test_health_check_performance(self): """Test that health checks are fast.""" import time start_time = time.time() response = client.get("/health/live") end_time = time.time() response_time = end_time - start_time assert response.status_code == 200 assert response_time < 1.0 # Health checks should be very fast @pytest.mark.asyncio class TestAsyncEndpoints: """Test async endpoint functionality.""" async def test_async_client_basic_functionality(self, async_client): """Test that async client works with basic endpoints.""" response = await async_client.get("/") assert response.status_code == 200 data = response.json() assert "message" in data assert "version" in data async def test_async_health_check(self, async_client): """Test health check via async client.""" response = await async_client.get("/health/") assert response.status_code == 200 data = response.json() assert "status" in data assert "services" in data async def test_async_api_info(self, async_client): """Test API info via async client.""" response = await async_client.get("/api/v1/info") assert response.status_code == 200 data = response.json() assert "api" in data assert "agents" in data class TestAPIValidation: """Test API request validation.""" def test_content_type_validation(self): """Test content type validation for POST requests.""" # Try posting without proper content-type response = client.post("/api/v1/info", data="test data") # Should handle appropriately assert response.status_code in [405, 415, 422] def test_accept_header_handling(self): """Test Accept header handling.""" # Request JSON specifically headers = {"Accept": "application/json"} response = client.get("/", headers=headers) assert response.status_code == 200 assert "application/json" in response.headers.get("content-type", "") def test_user_agent_handling(self): """Test User-Agent header handling.""" headers = {"User-Agent": "Test Client / Integration Test"} response = client.get("/", headers=headers) assert response.status_code == 200 # Should handle any user agent gracefully if __name__ == "__main__": pytest.main([__file__, "-v"])