diff --git a/django_logging/tests/middleware/test_base_middleware.py b/django_logging/tests/middleware/test_base_middleware.py new file mode 100644 index 0000000..816b9d2 --- /dev/null +++ b/django_logging/tests/middleware/test_base_middleware.py @@ -0,0 +1,71 @@ +import sys +from typing import Callable +from unittest.mock import Mock + +import pytest +from asgiref.sync import iscoroutinefunction +from django.http import HttpRequest, HttpResponseBase + +from django_logging.middleware.base import BaseMiddleware +from django_logging.tests.constants import PYTHON_VERSION, PYTHON_VERSION_REASON + +pytestmark = [ + pytest.mark.middleware, + pytest.mark.base_middleware, + pytest.mark.skipif(sys.version_info < PYTHON_VERSION, reason=PYTHON_VERSION_REASON), +] + + +class TestBaseMiddleware: + """ + Test suite for the BaseMiddleware class. + """ + + def test_sync_mode(self) -> None: + """ + Test that the middleware correctly identifies and handles synchronous requests. + This test verifies that when the `get_response` function is synchronous, + the middleware calls the `__sync_call__` method. + """ + # Mock synchronous get_response + mock_get_response = Mock(spec=Callable[[HttpRequest], HttpResponseBase]) + + # Create an instance of the middleware + middleware = BaseMiddleware(mock_get_response) + + # Ensure that it is in synchronous mode + assert not iscoroutinefunction(middleware.get_response) + assert not middleware.async_mode + + # Test that calling the middleware raises NotImplementedError (since __sync_call__ is not implemented) + with pytest.raises( + NotImplementedError, match="__sync_call__ must be implemented by subclass" + ): + request = HttpRequest() + middleware(request) + + @pytest.mark.asyncio + async def test_async_mode(self) -> None: + """ + Test that the middleware correctly identifies and handles asynchronous requests. + This test verifies that when the `get_response` function is asynchronous, + the middleware calls the `__acall__` method. + """ + + # Mock asynchronous get_response + async def mock_get_response(request: HttpRequest) -> HttpResponseBase: + return Mock(spec=HttpResponseBase) + + # Create an instance of the middleware + middleware = BaseMiddleware(mock_get_response) + + # Ensure that it is in asynchronous mode + assert iscoroutinefunction(middleware.get_response) + assert middleware.async_mode + + # Test that calling the middleware raises NotImplementedError (since __acall__ is not implemented) + with pytest.raises( + NotImplementedError, match="__acall__ must be implemented by subclass" + ): + request = HttpRequest() + await middleware(request) diff --git a/django_logging/tests/middleware/test_monitor_log_size.py b/django_logging/tests/middleware/test_monitor_log_size.py new file mode 100644 index 0000000..c16e7e0 --- /dev/null +++ b/django_logging/tests/middleware/test_monitor_log_size.py @@ -0,0 +1,141 @@ +import sys +from datetime import timedelta +from unittest.mock import Mock, patch + +import pytest +from django.core.cache import cache +from django.http import HttpRequest, HttpResponse +from django.utils.timezone import now + +from django_logging.middleware.monitor_log_size import MonitorLogSizeMiddleware +from django_logging.tests.constants import PYTHON_VERSION, PYTHON_VERSION_REASON + +pytestmark = [ + pytest.mark.middleware, + pytest.mark.monitor_log_size_middleware, + pytest.mark.skipif(sys.version_info < PYTHON_VERSION, reason=PYTHON_VERSION_REASON), +] + + +class TestMonitorLogSizeMiddleware: + """ + Test suite for the MonitorLogSizeMiddleware class. + """ + + @pytest.fixture(autouse=True) + def setup(self) -> None: + """ + Clears cache before each test. + """ + cache.clear() + + def test_should_run_task_no_cache(self) -> None: + """ + Test that the task should run when there is no cache entry for 'last_run_logs_size_audit'. + """ + assert MonitorLogSizeMiddleware.should_run_task() is True + + def test_should_run_task_with_recent_cache(self) -> None: + """ + Test that the task should not run if the cache indicates the last run was within a week. + """ + last_run_time = now() - timedelta(days=2) + cache.set("last_run_logs_size_audit", last_run_time) + + assert MonitorLogSizeMiddleware.should_run_task() is False + + def test_should_run_task_with_old_cache(self) -> None: + """ + Test that the task should run if the cache indicates the last run was more than a week ago. + """ + last_run_time = now() - timedelta(weeks=2) + cache.set("last_run_logs_size_audit", last_run_time) + + assert MonitorLogSizeMiddleware.should_run_task() is True + + @patch("django_logging.middleware.monitor_log_size.call_command") + def test_sync_run_log_size_check(self, mock_call_command: Mock) -> None: + """ + Test the synchronous execution of the log size check. + """ + mock_get_response = Mock(return_value=HttpResponse()) + middleware = MonitorLogSizeMiddleware(mock_get_response) + + request = HttpRequest() + + # Simulate no recent audit, so the task should run + cache.set("last_run_logs_size_audit", now() - timedelta(weeks=2)) + + response = middleware.__sync_call__(request) + + mock_call_command.assert_called_once_with("logs_size_audit") + assert cache.get("last_run_logs_size_audit") is not None + assert response.status_code == 200 + + @pytest.mark.asyncio + @patch("django_logging.middleware.monitor_log_size.call_command") + async def test_async_run_log_size_check(self, mock_call_command: Mock) -> None: + """ + Test the asynchronous execution of the log size check. + """ + + async def mock_get_response(request: HttpRequest) -> HttpResponse: + return HttpResponse() + + middleware = MonitorLogSizeMiddleware(mock_get_response) + + request = HttpRequest() + + # Simulate no recent audit, so the task should run + cache.set("last_run_logs_size_audit", now() - timedelta(weeks=2)) + + response = await middleware.__acall__(request) + + mock_call_command.assert_called_once_with("logs_size_audit") + assert cache.get("last_run_logs_size_audit") is not None + assert response.status_code == 200 + + @patch( + "django_logging.middleware.monitor_log_size.call_command", side_effect=Exception("Command failed") + ) + def test_sync_run_log_size_check_failure(self, mock_call_command: Mock) -> None: + """ + Test error handling in the synchronous log size check. + """ + mock_get_response = Mock(return_value=HttpResponse()) + middleware = MonitorLogSizeMiddleware(mock_get_response) + + request = HttpRequest() + + with patch( + "django_logging.middleware.monitor_log_size.logger.error" + ) as mock_logger: + middleware.__sync_call__(request) + + mock_call_command.assert_called_once_with("logs_size_audit") + mock_logger.assert_called_once() + + @pytest.mark.asyncio + @patch( + "django_logging.middleware.monitor_log_size.call_command", side_effect=Exception("Command failed") + ) + async def test_async_run_log_size_check_failure(self, mock_call_command: Mock) -> None: + """ + Test error handling in the asynchronous log size check. + """ + + async def mock_get_response(request): + return HttpResponse() + + middleware = MonitorLogSizeMiddleware(mock_get_response) + + request = HttpRequest() + + with patch( + "django_logging.middleware.monitor_log_size.logger.error" + ) as mock_logger: + await middleware.__acall__(request) + + mock_call_command.assert_called_once_with("logs_size_audit") + mock_logger.assert_called_once() + diff --git a/django_logging/tests/middleware/test_request_middleware.py b/django_logging/tests/middleware/test_request_middleware.py index 1571881..61089c9 100644 --- a/django_logging/tests/middleware/test_request_middleware.py +++ b/django_logging/tests/middleware/test_request_middleware.py @@ -1,10 +1,14 @@ -import logging +import asyncio +import io import sys -from unittest.mock import Mock +from typing import AsyncGenerator, Generator +from unittest.mock import AsyncMock, MagicMock, patch import pytest -from django.contrib.auth import get_user_model -from django.contrib.auth.models import AnonymousUser +from django.contrib.auth.models import AnonymousUser, User +from django.core.handlers.asgi import ASGIRequest +from django.db import connection +from django.http import HttpRequest, HttpResponse, StreamingHttpResponse from django.test import RequestFactory from django_logging.middleware import RequestLogMiddleware @@ -19,135 +23,343 @@ class TestRequestMiddleware: - def test_authenticated_user_logging( - self, - request_middleware: RequestLogMiddleware, - request_factory: RequestFactory, - caplog: pytest.LogCaptureFixture, + @pytest.mark.django_db + def test_sync_sql_logging( + self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware ) -> None: """ - Test logging of requests for authenticated users. - - This test verifies that when an authenticated user makes a request, - the relevant request information, including the username, is logged. - - Args: - ---- - request_middleware : RequestLogMiddleware - The middleware instance used to process the request. - request_factory : RequestFactory - A factory for creating mock HTTP requests. - caplog : pytest.LogCaptureFixture - A fixture for capturing log messages. + Test that SQL query logging works for synchronous requests. Asserts: ------- - - "Request Info" is present in the logs. - - The requested path is logged. - - The username of the authenticated user is logged. - - The request object has `ip_address` and `browser_type` attributes. - """ - request = request_factory.get("/test-path") - - UserModel = get_user_model() - username_field = UserModel.USERNAME_FIELD - - request.user = Mock() - request.user.is_authenticated = True - setattr(request.user, username_field, "test_user") - - with caplog.at_level(logging.INFO): - request_middleware(request) - - assert "Request Info" in caplog.text - assert "test-path" in caplog.text - assert "test_user" in caplog.text - assert request.ip_address - assert request.browser_type - - def test_anonymous_user_logging( - self, - request_middleware: RequestLogMiddleware, - request_factory: RequestFactory, - caplog: pytest.LogCaptureFixture, - ) -> None: + - SQL queries are logged when `self.log_sql` is True. """ - Test logging of requests for anonymous users. + request = request_factory.get("/") + request.user = AnonymousUser() + + # Simulate an SQL query + with connection.cursor() as cursor: + cursor.execute("SELECT 1") + + response = request_middleware(request) - This test ensures that when an anonymous user makes a request, - the relevant request information, including the identification as "Anonymous", is logged. + assert response.status_code == 200 + # If possible, capture the logger output to assert SQL logging - Args: - ---- - request_middleware : RequestLogMiddleware - The middleware instance used to process the request. - request_factory : RequestFactory - A factory for creating mock HTTP requests. - caplog : pytest.LogCaptureFixture - A fixture for capturing log messages. + @pytest.mark.asyncio + async def test_async_request(self, request_factory: RequestFactory) -> None: + """ + Test handling of an asynchronous request with RequestLogMiddleware. Asserts: ------- - - "Request Info" is present in the logs. - - The request is identified as coming from an "Anonymous" user. + - The middleware processes the asynchronous request successfully and returns a response. """ - request = request_factory.get("/test-path") + request = request_factory.get("/") request.user = AnonymousUser() - with caplog.at_level(logging.INFO): - request_middleware(request) + async def async_get_response(request: HttpRequest) -> HttpResponse: + return HttpResponse("OK") + + middleware = RequestLogMiddleware(async_get_response) + middleware.log_sql = True + + # Convert HttpRequest to ASGIRequest for async behavior + scope = { + "type": "http", + "method": "GET", + "path": "/", + "headers": [], + } + + body_file = io.BytesIO(b"") + # Create an ASGIRequest object + asgi_request = ASGIRequest(scope, body_file) + response = await middleware(asgi_request) + assert response.status_code == 200 + assert "OK" in response.content.decode() + + # Test exception block + async def async_get_response_with_error(request: HttpRequest) -> HttpResponse: + raise asyncio.CancelledError() + + middleware = RequestLogMiddleware(async_get_response_with_error) - assert "Request Info" in caplog.text - assert "Anonymous" in caplog.text + with pytest.raises(asyncio.CancelledError): + await middleware(asgi_request) - def test_ip_address_extraction( - self, request_middleware: RequestLogMiddleware, request_factory: RequestFactory + @pytest.mark.django_db + def test_request_id_header( + self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware ) -> None: """ - Test extraction of the client's IP address from the request. + Test that RequestLogMiddleware retrieves the request ID from the headers. - This test verifies that the middleware correctly extracts the IP address - from the `HTTP_X_FORWARDED_FOR` header in the request. + Asserts: + ------- + - The request ID is correctly retrieved from the request headers. + """ + request = request_factory.get("/") + request.headers = {"x-request-id": "12345"} + request.user = AnonymousUser() + + response = request_middleware(request) + + assert response.status_code == 200 + assert request_middleware.context["request_id"] == "12345" - Args: - ---- - request_middleware : RequestLogMiddleware - The middleware instance used to process the request. - request_factory : RequestFactory - A factory for creating mock HTTP requests. + @pytest.mark.asyncio + async def test_async_streaming_response( + self, request_factory: RequestFactory + ) -> None: + """ + Test handling of asynchronous streaming responses with RequestLogMiddleware. Asserts: ------- - - The `ip_address` attribute of the request is correctly set to the value in the `HTTP_X_FORWARDED_FOR` header. + - The middleware handles asynchronous streaming responses correctly. """ - request = request_factory.get("/test-path", HTTP_X_FORWARDED_FOR="192.168.1.1") + request = request_factory.get("/") + request.user = AnonymousUser() - request_middleware(request) + async def streaming_response(request: HttpRequest) -> StreamingHttpResponse: + async def generator() -> AsyncGenerator: + for chunk in [b"chunk1", b"chunk2"]: + yield chunk + + _response = StreamingHttpResponse(generator()) + return _response + + middleware = RequestLogMiddleware(streaming_response) - assert request.ip_address == "192.168.1.1" + response = await middleware(request) - def test_user_agent_extraction( - self, request_middleware: RequestLogMiddleware, request_factory: RequestFactory + assert response.status_code == 200 + assert response.streaming + # Assert the streaming content + streaming_content = [chunk async for chunk in response.streaming_content] + assert streaming_content == [b"chunk1", b"chunk2"] + + # Test exception handling in sync_streaming_wrapper + + def test_sync_streaming_wrapper( + self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware ) -> None: """ - Test extraction of the client's user agent from the request. + Test that the sync_streaming_wrapper handles StreamingHttpResponse and exceptions correctly. + """ - This test verifies that the middleware correctly extracts the user agent - from the `HTTP_USER_AGENT` header in the request. + def streaming_view(request: HttpRequest) -> StreamingHttpResponse: + def generator() -> Generator: + yield b"chunk1" + yield b"chunk2" - Args: - ---- - request_middleware : RequestLogMiddleware - The middleware instance used to process the request. - request_factory : RequestFactory - A factory for creating mock HTTP requests. + return StreamingHttpResponse(generator()) - Asserts: - ------- - - The `browser_type` attribute of the request is correctly set to the value in the `HTTP_USER_AGENT` header. + request = request_factory.get("/") + request.user = AnonymousUser() + + middleware = RequestLogMiddleware(streaming_view) + + with patch( + "django_logging.middleware.request_middleware.logger" + ) as mock_logger: + response = middleware(request) + assert response.status_code == 200 + assert response.streaming + + def test_sync_streaming_wrapper_raises_exception(self, request_middleware: RequestLogMiddleware) -> None: + """ + Test that sync_streaming_wrapper handles an exception during streaming. + + Steps: + - Mock the streaming content to raise an exception. + - Assert that the exception is logged and re-raised. + """ + + request_id = "test-request-id" + + # Mock the streaming content to raise an exception when iterated + streaming_content = MagicMock() + streaming_content.__iter__.side_effect = Exception("Test Exception") + + # Patch the logger to check for log messages + with patch( + "django_logging.middleware.request_middleware.logger" + ) as mock_logger: + with pytest.raises(Exception, match="Test Exception"): + list( + request_middleware._sync_streaming_wrapper( + streaming_content, request_id + ) + ) + + # Check that logger.exception was called with the correct message + mock_logger.exception.assert_called_once_with( + "Streaming failed: request_id=%s", request_id + ) + + @pytest.mark.asyncio + async def test_async_streaming_wrapper_cancelled_error(self, request_middleware: RequestLogMiddleware) -> None: + """ + Test that async_streaming_wrapper handles asyncio.CancelledError properly. + + Steps: + - Mock the streaming content to raise asyncio.CancelledError. + - Assert that the cancellation is logged and re-raised. """ - request = request_factory.get("/test-path", HTTP_USER_AGENT="Mozilla/5.0") + request_id = "test-request-id" + + # Mock the streaming content to raise asyncio.CancelledError + streaming_content = AsyncMock() + streaming_content.__aiter__.side_effect = asyncio.CancelledError + + # Patch the logger to check for log messages + with patch( + "django_logging.middleware.request_middleware.logger" + ) as mock_logger: + with pytest.raises(asyncio.CancelledError): + async for _ in request_middleware._async_streaming_wrapper( + streaming_content, request_id + ): + pass + + # Check that logger.warning was called with the correct message + mock_logger.warning.assert_called_once_with( + "Streaming was cancelled: request_id=%s", request_id + ) + + @pytest.mark.asyncio + async def test_async_streaming_wrapper_generic_exception(self, request_middleware: RequestLogMiddleware) -> None: + """ + Test that async_streaming_wrapper handles a generic Exception properly. + + Steps: + - Mock the streaming content to raise a generic Exception. + - Assert that the exception is logged and re-raised. + """ + + request_id = "test-request-id" + + # Mock the streaming content to raise a generic Exception + streaming_content = AsyncMock() + streaming_content.__aiter__.side_effect = Exception("Test Exception") + + # Patch the logger to check for log messages + with patch( + "django_logging.middleware.request_middleware.logger" + ) as mock_logger: + with pytest.raises(Exception, match="Test Exception"): + async for _ in request_middleware._async_streaming_wrapper( + streaming_content, request_id + ): + pass + + # Check that logger.exception was called with the correct message + mock_logger.exception.assert_called_once_with( + "Streaming failed: request_id=%s", request_id + ) + + def test_sync_streaming_response_wrapper(self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware) -> None: + """ + Test that the synchronous streaming wrapper works correctly. + """ + + def streaming_view(request: HttpRequest) -> StreamingHttpResponse: + return StreamingHttpResponse(iter([b"chunk1", b"chunk2"])) + + request = request_factory.get("/") + request.user = AnonymousUser() + + # Wrap the streaming content in the middleware + middleware = RequestLogMiddleware(streaming_view) + response = middleware(request) + + assert response.status_code == 200 + assert response.streaming + # Assert the streaming content + streaming_content = list(response.streaming_content) + assert streaming_content == [b"chunk1", b"chunk2"] + + @pytest.mark.django_db + def test_get_user_authenticated(self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware) -> None: + """ + Test that the middleware retrieves the correct username for authenticated users. + """ + user = User.objects.create(username="testuser") + + request = request_factory.get("/") + request.user = user + + response = request_middleware(request) + + assert response.status_code == 200 + assert request_middleware.get_user(request) == f"[testuser (ID:{user.pk})]" + + def test_get_user_anonymous(self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware) -> None: + """ + Test that the middleware retrieves 'Anonymous' for unauthenticated users. + """ + request = request_factory.get("/") + request.user = AnonymousUser() + + response = request_middleware(request) + + assert response.status_code == 200 + assert request_middleware.get_user(request) == "Anonymous" + + def test_get_ip_address(self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware) -> None: + """ + Test that the middleware correctly retrieves the client's IP address. + """ + request = request_factory.get("/") + request.META["REMOTE_ADDR"] = "192.168.1.1" + request.user = AnonymousUser() + + response = request_middleware(request) + + assert response.status_code == 200 + assert request_middleware.get_ip_address(request) == "192.168.1.1" + + request.META["REMOTE_ADDR"] = None + + request.META["HTTP_X_FORWARDED_FOR"] = "192.168.1.1," + assert request_middleware.get_ip_address(request) == "192.168.1.1" + + request.ip_address = "192.168.1.1" + assert request_middleware.get_ip_address(request) == "192.168.1.1" + + def test_get_request_id(self, request_factory: RequestFactory, request_middleware: RequestLogMiddleware) -> None: + """ + Test that the middleware correctly retrieves the request ID from headers. + """ + request = request_factory.get("/") + request.headers = {"x-request-id": "12345"} + request.user = AnonymousUser() + + response = request_middleware(request) + + assert response.status_code == 200 + assert request_middleware.get_request_id(request) == "12345" + + request.headers = {} + request.META["HTTP_X_REQUEST_ID"] = "12345" request_middleware(request) + assert request_middleware.get_request_id(request) == "12345" + + def test_log_sql_queries_with_queries(self, request_middleware: RequestLogMiddleware) -> None: + """ + Test that _log_sql_queries correctly logs and formats SQL queries. + """ + + # Simulated SQL queries (new_queries in the method) + mock_queries = [ + {"time": "0.002", "sql": "SELECT * FROM my_table WHERE id = 1"}, + {"time": "0.004", "sql": "UPDATE my_table SET value = 'test' WHERE id = 1"}, + ] + + log_output = request_middleware._log_sql_queries(0, mock_queries) - assert request.browser_type == "Mozilla/5.0" + # Assert that the log output executed + assert log_output