KYC MCP Server

"""Integration tests for external API interactions.""" import pytest from unittest.mock import AsyncMock, patch import httpx from src.clients.kyc_api_client import KYCAPIClient, ValidationError, ServiceUnavailableError from src.tools.pan_verification import PANVerificationTool from src.tools.pan_aadhaar_link import PANAadhaarLinkTool class TestExternalAPIIntegration: """Integration tests for external KYC API.""" @pytest.fixture def api_client(self): """Create API client for testing.""" return KYCAPIClient( base_url="https://api.sandbox.co.in", api_key="test_api_key", jwt_token="Bearer test_token", timeout=30 ) @pytest.mark.asyncio @pytest.mark.integration async def test_pan_verification_end_to_end(self, api_client, mock_responses): """Test complete PAN verification flow.""" # Mock the HTTP client mock_response = mock_responses["pan_verification"]["success_all_match"] api_client.client.post = AsyncMock(return_value=type('Response', (), { 'status_code': 200, 'json': lambda: mock_response, 'raise_for_status': lambda: None })()) # Create tool and execute tool = PANVerificationTool(api_client=api_client) result = await tool.execute({ "pan": "ABCDE1234F", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "KYC verification" }) assert result["pan"] == "ABCDE1234F" assert result["status"] == "valid" assert result["name_match"] is True assert result["dob_match"] is True @pytest.mark.asyncio @pytest.mark.integration async def test_pan_aadhaar_link_end_to_end(self, api_client, mock_responses): """Test complete PAN-Aadhaar link check flow.""" # Mock the HTTP client mock_response = mock_responses["pan_aadhaar_link"]["linked"] api_client.client.post = AsyncMock(return_value=type('Response', (), { 'status_code': 200, 'json': lambda: mock_response, 'raise_for_status': lambda: None })()) # Create tool and execute tool = PANAadhaarLinkTool(api_client=api_client) result = await tool.execute({ "pan": "ABCPE1234F", "aadhaar_number": "123456789012", "consent": "Y", "reason": "Link status check" }) assert result["linked"] is True assert result["status"] == "y" @pytest.mark.asyncio @pytest.mark.integration async def test_api_authentication_headers(self, api_client): """Test that authentication headers are properly set.""" # Verify headers are set in client assert "Authorization" in api_client.client.headers assert "x-api-key" in api_client.client.headers assert api_client.client.headers["Authorization"] == "Bearer test_token" assert api_client.client.headers["x-api-key"] == "test_api_key" @pytest.mark.asyncio @pytest.mark.integration async def test_api_timeout_handling(self, api_client): """Test handling of API timeouts.""" # Mock timeout api_client.client.post = AsyncMock( side_effect=httpx.TimeoutException("Request timed out") ) tool = PANVerificationTool(api_client=api_client) with pytest.raises(Exception): await tool.execute({ "pan": "ABCDE1234F", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) @pytest.mark.asyncio @pytest.mark.integration async def test_api_validation_error_422(self, api_client, mock_responses): """Test handling of 422 validation errors from API.""" # Mock validation error response mock_response = type('Response', (), { 'status_code': 422, 'json': lambda: mock_responses["pan_verification"]["invalid_pan"], 'request': type('Request', (), {'url': 'test', 'method': 'POST'})() })() def raise_error(): raise httpx.HTTPStatusError( message="422", request=mock_response.request, response=mock_response ) mock_response.raise_for_status = raise_error api_client.client.post = AsyncMock(return_value=mock_response) tool = PANVerificationTool(api_client=api_client) with pytest.raises(ValidationError): await tool.execute({ "pan": "INVALID", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) @pytest.mark.asyncio @pytest.mark.integration async def test_api_service_unavailable_503(self, api_client, mock_responses): """Test handling of 503 service unavailable errors.""" # Mock service unavailable response mock_response = type('Response', (), { 'status_code': 503, 'json': lambda: mock_responses["pan_verification"]["service_unavailable"], 'request': type('Request', (), {'url': 'test', 'method': 'POST'})() })() def raise_error(): raise httpx.HTTPStatusError( message="503", request=mock_response.request, response=mock_response ) mock_response.raise_for_status = raise_error api_client.client.post = AsyncMock(return_value=mock_response) tool = PANVerificationTool(api_client=api_client) with pytest.raises(ServiceUnavailableError): await tool.execute({ "pan": "ABCDE1234F", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) @pytest.mark.asyncio @pytest.mark.integration async def test_api_retry_on_failure(self, api_client, mock_responses): """Test that API requests are retried on failure.""" call_count = 0 async def mock_post(*args, **kwargs): nonlocal call_count call_count += 1 if call_count < 2: # Fail first attempt raise httpx.RequestError("Temporary failure") # Succeed on second attempt return type('Response', (), { 'status_code': 200, 'json': lambda: mock_responses["pan_verification"]["success_all_match"], 'raise_for_status': lambda: None })() api_client.client.post = mock_post tool = PANVerificationTool(api_client=api_client) result = await tool.execute({ "pan": "ABCDE1234F", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) assert result["status"] == "valid" assert call_count == 2 # Should have retried once @pytest.mark.asyncio @pytest.mark.integration async def test_concurrent_api_requests(self, api_client, mock_responses): """Test handling of concurrent API requests.""" import asyncio # Mock successful responses mock_response = mock_responses["pan_verification"]["success_all_match"] api_client.client.post = AsyncMock(return_value=type('Response', (), { 'status_code': 200, 'json': lambda: mock_response, 'raise_for_status': lambda: None })()) tool = PANVerificationTool(api_client=api_client) # Execute multiple requests concurrently requests = [ { "pan": f"ABCD{i}1234F", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" } for i in range(5) ] results = await asyncio.gather(*[ tool.execute(req) for req in requests ]) assert len(results) == 5 assert all(r["status"] == "valid" for r in results) @pytest.mark.asyncio @pytest.mark.integration async def test_api_response_parsing(self, api_client, mock_responses): """Test correct parsing of API responses.""" # Test with deceased holder response mock_response = mock_responses["pan_verification"]["success_deceased"] api_client.client.post = AsyncMock(return_value=type('Response', (), { 'status_code': 200, 'json': lambda: mock_response, 'raise_for_status': lambda: None })()) tool = PANVerificationTool(api_client=api_client) result = await tool.execute({ "pan": "DEFGH3456K", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) assert result["remarks"] == "Holder is Deceased" assert result["aadhaar_seeding_status"] == "na" @pytest.mark.asyncio @pytest.mark.integration async def test_api_connection_pooling(self, api_client): """Test that connection pooling is properly configured.""" # Verify limits are set assert api_client.client.limits.max_connections == 100 assert api_client.client.limits.max_keepalive_connections == 20 @pytest.mark.asyncio @pytest.mark.integration async def test_api_base_url_configuration(self, api_client): """Test that base URL is properly configured.""" assert str(api_client.client.base_url) == "https://api.sandbox.co.in" @pytest.mark.asyncio @pytest.mark.integration async def test_api_client_cleanup(self, api_client): """Test proper cleanup of API client resources.""" api_client.client.aclose = AsyncMock() await api_client.close() api_client.client.aclose.assert_called_once() @pytest.mark.asyncio @pytest.mark.integration async def test_multiple_tools_same_client(self, api_client, mock_responses): """Test using same client for multiple tools.""" # Mock responses for both tools pan_response = mock_responses["pan_verification"]["success_all_match"] aadhaar_response = mock_responses["pan_aadhaar_link"]["linked"] call_count = 0 async def mock_post(*args, **kwargs): nonlocal call_count call_count += 1 if call_count == 1: response_data = pan_response else: response_data = aadhaar_response return type('Response', (), { 'status_code': 200, 'json': lambda: response_data, 'raise_for_status': lambda: None })() api_client.client.post = mock_post # Execute both tools pan_tool = PANVerificationTool(api_client=api_client) aadhaar_tool = PANAadhaarLinkTool(api_client=api_client) pan_result = await pan_tool.execute({ "pan": "ABCDE1234F", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) aadhaar_result = await aadhaar_tool.execute({ "pan": "ABCPE1234F", "aadhaar_number": "123456789012", "consent": "Y", "reason": "Test" }) assert pan_result["status"] == "valid" assert aadhaar_result["linked"] is True assert call_count == 2 @pytest.mark.asyncio @pytest.mark.integration async def test_api_error_message_extraction(self, api_client): """Test extraction of error messages from API responses.""" error_response = { "code": "422", "message": "Invalid PAN format", "errors": [{"field": "pan", "message": "Must be 10 characters"}] } mock_response = type('Response', (), { 'status_code': 422, 'json': lambda: error_response, 'request': type('Request', (), {'url': 'test', 'method': 'POST'})() })() def raise_error(): raise httpx.HTTPStatusError( message="422", request=mock_response.request, response=mock_response ) mock_response.raise_for_status = raise_error api_client.client.post = AsyncMock(return_value=mock_response) tool = PANVerificationTool(api_client=api_client) try: await tool.execute({ "pan": "INVALID", "name_as_per_pan": "John Doe", "date_of_birth": "01/01/1990", "consent": "Y", "reason": "Test" }) except ValidationError as e: assert "Invalid PAN format" in str(e)