diff --git a/src/tests/test_logger.py b/src/tests/test_logger.py index 5bc67dd..c020055 100644 --- a/src/tests/test_logger.py +++ b/src/tests/test_logger.py @@ -1,6 +1,9 @@ from unittest.mock import patch +import pytest +from fastapi import Request from fastapi.testclient import TestClient from app.main import app +from app.middleware.logging import RequestLoggingMiddleware, request_id_var client = TestClient(app) @@ -32,6 +35,39 @@ def test_x_request_id_passed(): mock_logger.assert_called_once() +@pytest.mark.anyio +async def test_logger_middleware_unit_exception(): + # Unit test RequestLoggingMiddleware directly when call_next raises an exception. + middleware = RequestLoggingMiddleware(app) + mock_request = Request( + { + "type": "http", + "method": "GET", + "path": "/error-test", + "headers": [(b"x-request-id", b"unit-test-id")], + } + ) + + async def mock_call_next(req): + raise RuntimeError("Downstream failure") + + with patch("app.middleware.logging.logger.info") as mock_logger: + with pytest.raises(RuntimeError, match="Downstream failure"): + await middleware.dispatch(mock_request, mock_call_next) + + mock_logger.assert_called_once() + args, kwargs = mock_logger.call_args + + assert "GET /error-test - 500" in args[0] + assert kwargs["extra"]["path"] == "/error-test" + assert kwargs["extra"]["method"] == "GET" + assert kwargs["extra"]["status_code"] == 500 + assert "latency" in kwargs["extra"] + + # Verify contextvar request_id_var was properly reset + assert request_id_var.get() == "" + + def test_json_formatter_redacts_exception_pii(): import json import logging diff --git a/src/tests/test_security_headers.py b/src/tests/test_security_headers.py index b9afd55..7e3ef65 100644 --- a/src/tests/test_security_headers.py +++ b/src/tests/test_security_headers.py @@ -1,3 +1,4 @@ +from unittest.mock import patch from fastapi.testclient import TestClient from app.main import app @@ -32,3 +33,28 @@ def test_security_headers(): headers["Content-Security-Policy"] == "default-src 'self'; frame-ancestors 'none';" ) + + +def test_security_headers_on_500_exception(): + # Test that security headers are present on 500 Internal Server Error responses + err_client = TestClient(app, raise_server_exceptions=False) + with patch("app.main.get_model", side_effect=RuntimeError("Unexpected error")): + response = err_client.post( + "/v1/embeddings", json={"input": "test", "model": "cl-nagoya/ruri-v3-30m"} + ) + + assert response.status_code == 500 + headers = response.headers + + assert headers.get("X-Content-Type-Options") == "nosniff" + assert headers.get("X-Frame-Options") == "DENY" + assert headers.get("X-XSS-Protection") == "1; mode=block" + assert ( + headers.get("Strict-Transport-Security") + == "max-age=31536000; includeSubDomains" + ) + assert headers.get("Referrer-Policy") == "strict-origin-when-cross-origin" + assert ( + headers.get("Content-Security-Policy") + == "default-src 'self'; frame-ancestors 'none';" + )