Spaces:
Running
Running
File size: 4,247 Bytes
2aa2b79 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 |
"""
Tests for middleware functionality.
"""
import pytest
from unittest.mock import Mock, patch
from fastapi import Request, Response
from app.core.middleware import request_context_middleware
class TestRequestContextMiddleware:
"""Test request_context_middleware functionality."""
@pytest.mark.asyncio
async def test_middleware_adds_request_id(self):
"""Test that middleware adds request ID to request and response."""
# Mock request and response
request = Mock(spec=Request)
request.headers = {}
request.state = Mock()
request.method = "GET"
request.url.path = "/test"
response = Mock(spec=Response)
response.headers = {}
response.status_code = 200
# Mock the call_next function
async def mock_call_next(req):
return response
# Test the middleware
result = await request_context_middleware(request, mock_call_next)
# Verify request ID was added to request state
assert hasattr(request.state, 'request_id')
assert request.state.request_id is not None
assert len(request.state.request_id) == 36 # UUID length
# Verify request ID was added to response headers
assert "X-Request-ID" in result.headers
assert result.headers["X-Request-ID"] == request.state.request_id
@pytest.mark.asyncio
async def test_middleware_preserves_existing_request_id(self):
"""Test that middleware preserves existing request ID from headers."""
# Mock request with existing request ID
request = Mock(spec=Request)
request.headers = {"X-Request-ID": "custom-id-123"}
request.state = Mock()
request.method = "POST"
request.url.path = "/api/test"
response = Mock(spec=Response)
response.headers = {}
response.status_code = 201
# Mock the call_next function
async def mock_call_next(req):
return response
# Test the middleware
result = await request_context_middleware(request, mock_call_next)
# Verify existing request ID was preserved
assert request.state.request_id == "custom-id-123"
assert result.headers["X-Request-ID"] == "custom-id-123"
@pytest.mark.asyncio
async def test_middleware_handles_exception(self):
"""Test that middleware handles exceptions properly."""
# Mock request
request = Mock(spec=Request)
request.headers = {}
request.state = Mock()
request.method = "GET"
request.url.path = "/error"
# Mock the call_next function to raise an exception
async def mock_call_next(req):
raise Exception("Test exception")
# Test that middleware doesn't suppress exceptions
with pytest.raises(Exception, match="Test exception"):
await request_context_middleware(request, mock_call_next)
# Verify request ID was still added
assert hasattr(request.state, 'request_id')
assert request.state.request_id is not None
@pytest.mark.asyncio
async def test_middleware_logging_integration(self):
"""Test that middleware integrates with logging."""
with patch('app.core.middleware.request_logger') as mock_logger:
# Mock request and response
request = Mock(spec=Request)
request.headers = {}
request.state = Mock()
request.method = "GET"
request.url.path = "/test"
response = Mock(spec=Response)
response.headers = {}
response.status_code = 200
# Mock the call_next function
async def mock_call_next(req):
return response
# Test the middleware
result = await request_context_middleware(request, mock_call_next)
# Verify logging was called
mock_logger.log_request.assert_called_once_with("GET", "/test", request.state.request_id)
mock_logger.log_response.assert_called_once()
|