Samir Boulahtit
238c1ec9b8
- Replace black, isort, and flake8 with Ruff (all-in-one linter and formatter) - Add comprehensive pyproject.toml configuration - Simplify Makefile code quality targets - Configure exclusions for venv/.venv in pyproject.toml - Auto-fix 1,359 linting issues across codebase Benefits: - Much faster builds (Ruff is written in Rust) - Single tool replaces multiple tools - More comprehensive rule set (UP, B, C4, SIM, PIE, RET, Q) - All configuration centralized in pyproject.toml - Better import sorting and formatting consistency 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
245 lines
8.1 KiB
Python
245 lines
8.1 KiB
Python
# tests/unit/middleware/test_logging.py
|
|
"""
|
|
Comprehensive unit tests for LoggingMiddleware.
|
|
|
|
Tests cover:
|
|
- Request/response logging
|
|
- Performance monitoring with X-Process-Time header
|
|
- Client IP address handling
|
|
- Exception logging
|
|
- Timing accuracy
|
|
- Edge cases (missing client info, etc.)
|
|
"""
|
|
|
|
import asyncio
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
from fastapi import Request
|
|
|
|
from middleware.logging import LoggingMiddleware
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLoggingMiddleware:
|
|
"""Test suite for LoggingMiddleware."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_logs_request(self):
|
|
"""Test middleware logs incoming request."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/api/vendors")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
# Create mock response with actual dict for headers
|
|
response = Mock()
|
|
response.status_code = 200
|
|
response.headers = {} # Use actual dict
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger") as mock_logger:
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# Verify request was logged
|
|
assert mock_logger.info.call_count >= 1
|
|
first_call = mock_logger.info.call_args_list[0]
|
|
assert "GET" in str(first_call)
|
|
assert "/api/vendors" in str(first_call)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_logs_response(self):
|
|
"""Test middleware logs response with status code and duration."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "POST"
|
|
request.url = Mock(path="/api/products")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock()
|
|
response.status_code = 201
|
|
response.headers = {}
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger") as mock_logger:
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
# Verify response was logged
|
|
assert mock_logger.info.call_count >= 2 # Request + Response
|
|
last_call = mock_logger.info.call_args_list[-1]
|
|
assert "201" in str(last_call)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_adds_process_time_header(self):
|
|
"""Test middleware adds X-Process-Time header."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/test")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock()
|
|
response.status_code = 200
|
|
response.headers = {}
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger"):
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
assert "X-Process-Time" in response.headers
|
|
# Should be a numeric string
|
|
process_time = float(response.headers["X-Process-Time"])
|
|
assert process_time >= 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_handles_no_client(self):
|
|
"""Test middleware handles requests with no client info."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/test")
|
|
request.client = None # No client info
|
|
|
|
response = Mock()
|
|
response.status_code = 200
|
|
response.headers = {}
|
|
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger") as mock_logger:
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# Should log "unknown" for client
|
|
assert any(
|
|
"unknown" in str(call) for call in mock_logger.info.call_args_list
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_logs_exceptions(self):
|
|
"""Test middleware logs exceptions."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/error")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
call_next = AsyncMock(side_effect=Exception("Test error"))
|
|
|
|
with (
|
|
patch("middleware.logging.logger") as mock_logger,
|
|
pytest.raises(Exception),
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# Verify error was logged
|
|
mock_logger.error.assert_called_once()
|
|
assert "Test error" in str(mock_logger.error.call_args)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_timing_accuracy(self):
|
|
"""Test middleware timing is reasonably accurate."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/slow")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
async def slow_call_next(req):
|
|
await asyncio.sleep(0.1) # 100ms delay
|
|
response = Mock(status_code=200, headers={})
|
|
return response
|
|
|
|
call_next = slow_call_next
|
|
|
|
with patch("middleware.logging.logger"):
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
process_time = float(result.headers["X-Process-Time"])
|
|
# Should be at least 0.1 seconds
|
|
assert process_time >= 0.1
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLoggingEdgeCases:
|
|
"""Test suite for edge cases and special scenarios."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_handles_very_fast_requests(self):
|
|
"""Test middleware handles requests that complete very quickly."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/fast")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger"):
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
# Should still have process time, even if very small
|
|
assert "X-Process-Time" in response.headers
|
|
process_time = float(response.headers["X-Process-Time"])
|
|
assert process_time >= 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_logs_different_methods(self):
|
|
"""Test middleware correctly logs different HTTP methods."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
methods = ["GET", "POST", "PUT", "PATCH", "DELETE"]
|
|
|
|
for method in methods:
|
|
request = Mock(spec=Request)
|
|
request.method = method
|
|
request.url = Mock(path="/test")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger") as mock_logger:
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# Verify method was logged
|
|
assert any(
|
|
method in str(call) for call in mock_logger.info.call_args_list
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_logs_different_status_codes(self):
|
|
"""Test middleware logs various HTTP status codes."""
|
|
middleware = LoggingMiddleware(app=None)
|
|
|
|
status_codes = [200, 201, 400, 404, 500]
|
|
|
|
for status_code in status_codes:
|
|
request = Mock(spec=Request)
|
|
request.method = "GET"
|
|
request.url = Mock(path="/test")
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=status_code, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.logging.logger") as mock_logger:
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# Verify status code was logged
|
|
assert any(
|
|
str(status_code) in str(call)
|
|
for call in mock_logger.info.call_args_list
|
|
)
|