conftest.py•6.06 kB
"""Pytest configuration and shared fixtures."""
import asyncio
import json
from pathlib import Path
from typing import AsyncGenerator, Dict, Any
from unittest.mock import AsyncMock, MagicMock
import pytest
import pytest_asyncio
from httpx import Response
from src.cache.redis_cache import CacheManager
from src.clients.kyc_api_client import KYCAPIClient
from src.registry.tool_registry import ToolRegistry
from src.tools.pan_verification import PANVerificationTool
from src.tools.pan_aadhaar_link import PANAadhaarLinkTool
from src.utils.rate_limiter import RateLimiter
@pytest.fixture(scope="session")
def event_loop():
"""Create event loop for async tests."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
def test_data_dir() -> Path:
"""Get test data directory."""
return Path(__file__).parent / "fixtures"
@pytest.fixture
def test_data(test_data_dir: Path) -> Dict[str, Any]:
"""Load test data from JSON file."""
with open(test_data_dir / "test_data.json", "r") as f:
return json.load(f)
@pytest.fixture
def mock_responses(test_data_dir: Path) -> Dict[str, Any]:
"""Load mock API responses from JSON file."""
with open(test_data_dir / "mock_responses.json", "r") as f:
return json.load(f)
@pytest.fixture
def mock_api_client() -> AsyncMock:
"""Create mock KYC API client."""
client = AsyncMock(spec=KYCAPIClient)
client.base_url = "https://api.sandbox.co.in"
client.api_key = "test_api_key"
client.jwt_token = "test_jwt_token"
return client
@pytest_asyncio.fixture
async def cache_manager() -> AsyncGenerator[CacheManager, None]:
"""Create cache manager with mock Redis."""
cache = CacheManager(host="localhost", port=6379, db=1)
# Mock the Redis client
cache.client = AsyncMock()
cache.client.ping = AsyncMock()
cache.client.get = AsyncMock(return_value=None)
cache.client.setex = AsyncMock()
cache.client.delete = AsyncMock()
cache.client.exists = AsyncMock(return_value=False)
yield cache
@pytest.fixture
def rate_limiter() -> RateLimiter:
"""Create rate limiter instance."""
return RateLimiter(requests_per_minute=60, requests_per_hour=1000)
@pytest.fixture
def pan_verification_tool(mock_api_client: AsyncMock) -> PANVerificationTool:
"""Create PAN verification tool with mock client."""
return PANVerificationTool(api_client=mock_api_client)
@pytest.fixture
def pan_aadhaar_tool(mock_api_client: AsyncMock) -> PANAadhaarLinkTool:
"""Create PAN-Aadhaar link tool with mock client."""
return PANAadhaarLinkTool(api_client=mock_api_client)
@pytest.fixture
def tool_registry(tmp_path: Path) -> ToolRegistry:
"""Create tool registry with temporary metadata directory."""
metadata_dir = tmp_path / "metadata" / "tools"
metadata_dir.mkdir(parents=True, exist_ok=True)
# Create sample metadata files
pan_verify_meta = {
"name": "verify_pan",
"description": "Verify PAN card with name and DOB matching",
"input_schema": {
"type": "object",
"properties": {
"pan": {"type": "string", "pattern": "^[A-Z]{5}[0-9]{4}[A-Z]$"},
"name_as_per_pan": {"type": "string"},
"date_of_birth": {"type": "string", "pattern": "^\\d{2}/\\d{2}/\\d{4}$"},
"consent": {"type": "string", "pattern": "^[Yy]$"},
"reason": {"type": "string"}
},
"required": ["pan", "name_as_per_pan", "date_of_birth", "consent", "reason"]
}
}
with open(metadata_dir / "pan_verification.json", "w") as f:
json.dump(pan_verify_meta, f)
registry = ToolRegistry(metadata_dir=metadata_dir)
registry.initialize()
return registry
@pytest.fixture
def valid_pan_request() -> Dict[str, Any]:
"""Valid PAN verification request."""
return {
"pan": "ABCDE1234F",
"name_as_per_pan": "John Doe",
"date_of_birth": "01/01/1990",
"consent": "Y",
"reason": "KYC verification"
}
@pytest.fixture
def valid_pan_aadhaar_request() -> Dict[str, Any]:
"""Valid PAN-Aadhaar link request."""
return {
"pan": "ABCPE1234F",
"aadhaar_number": "123456789012",
"consent": "Y",
"reason": "Link status check"
}
@pytest.fixture
def mock_pan_verification_response() -> Dict[str, Any]:
"""Mock successful PAN verification response."""
return {
"data": {
"pan": "ABCDE1234F",
"category": "individual",
"status": "valid",
"remarks": None,
"name_as_per_pan_match": True,
"date_of_birth_match": True,
"aadhaar_seeding_status": "y"
}
}
@pytest.fixture
def mock_pan_aadhaar_response() -> Dict[str, Any]:
"""Mock successful PAN-Aadhaar link response."""
return {
"data": {
"aadhaar_seeding_status": "y",
"message": "PAN and Aadhaar are linked"
}
}
def create_mock_httpx_response(
status_code: int,
json_data: Dict[str, Any] = None,
text: str = None
) -> Response:
"""Create mock httpx Response object."""
response = MagicMock(spec=Response)
response.status_code = status_code
response.json.return_value = json_data or {}
response.text = text or json.dumps(json_data or {})
response.raise_for_status = MagicMock()
if status_code >= 400:
from httpx import HTTPStatusError, Request
request = MagicMock(spec=Request)
request.url = "https://api.sandbox.co.in/test"
request.method = "POST"
response.request = request
response.raise_for_status.side_effect = HTTPStatusError(
message=f"HTTP {status_code}",
request=request,
response=response
)
return response
@pytest.fixture
def mock_httpx_response():
"""Factory fixture for creating mock httpx responses."""
return create_mock_httpx_response