Spaces:
Running
Running
File size: 5,470 Bytes
2aa2b79 7ab470d 2aa2b79 2ed2bd7 7ab470d 2ed2bd7 2aa2b79 7ab470d 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 7ab470d 2ed2bd7 2aa2b79 7ab470d 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 2ed2bd7 2aa2b79 7ab470d 2ed2bd7 7ab470d |
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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 |
"""
Tests for middleware functionality.
Tests verify core middleware behavior (request ID handling, exception handling).
Logging integration tests are in test_loguru_integration.py.
"""
from pathlib import Path
from unittest.mock import Mock
import pytest
from fastapi import Request, Response
from loguru import logger
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 ValueError("Test exception")
# Test that middleware doesn't suppress exceptions
with pytest.raises(ValueError, 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_sets_context_var(self, tmp_path: Path):
"""Test that middleware sets request ID in context variable."""
from app.core.logging import request_id_var
log_file = tmp_path / "middleware_context.log"
logger.remove()
logger.add(log_file, format="{message}", level="INFO")
# Mock request
request = Mock(spec=Request)
request.headers = {"X-Request-ID": "context-test-456"}
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):
# Log inside request handling (context var should be set)
current_request_id = request_id_var.get()
logger.info(f"Processing request {current_request_id}")
return response
# Test the middleware
await request_context_middleware(request, mock_call_next)
# Verify context var was set
content = log_file.read_text()
assert "context-test-456" in content
@pytest.mark.asyncio
async def test_middleware_logging_works(self, tmp_path: Path):
"""Test that middleware logs requests and responses."""
log_file = tmp_path / "middleware_log.log"
logger.remove()
logger.add(log_file, format="{message}", level="INFO")
# Mock request and response
request = Mock(spec=Request)
request.headers = {}
request.state = Mock()
request.method = "POST"
request.url.path = "/api/data"
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
await request_context_middleware(request, mock_call_next)
# Verify logs contain request and response
content = log_file.read_text()
assert "POST /api/data" in content
assert "201" in content
|