test_fastapi_middleware.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/test_fastapi_middleware.py
 
"""
Regression Tests — fastapi_middleware
=======================================
 
Purpose
-------
Verify the ASGI contract of ``DiagnosticsMiddleware``: correct HTTP scope
handling, non-HTTP passthrough, exception logging and re-raising, header
redaction, elapsed-time reporting, and lifespan scope passthrough.
 
Responsibilities
----------------
- Confirm HTTP status codes are captured and passed through to the client.
- Confirm non-HTTP scopes (WebSocket, lifespan) are forwarded unchanged.
- Confirm application exceptions are logged at L2 and then re-raised.
- Confirm sensitive request headers are redacted in log output (FS-17).
- Confirm elapsed time in milliseconds appears in the completion log line.
- Confirm lifespan scope passes through without any HTTP logging.
 
Diagnostics
-----------
Domain: TEST
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Must not permanently modify oc_diagnostics global state.
- Must restore all patched globals and mocked internals in finally blocks.
- Must not start servers or require network access.
- Must run deterministically across repeated invocations.
- Pure ASGI contract — does not require the FastAPI package.
"""
 
import asyncio
import os
import sys
import unittest.mock as mock
 
# ---------------------------------------------------------------------------
# Ensure oc_diagnostics is importable when run from the project root or the
# tests/ subdirectory.
# ---------------------------------------------------------------------------
_REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
if _REPO_ROOT not in sys.path:
    sys.path.insert(0, _REPO_ROOT)
 
import oc_diagnostics.unified_debug as _ud
from oc_diagnostics.fastapi_middleware import DiagnosticsMiddleware
 
 
def _make_http_scope(method="GET", path="/ping", headers=None, client=("127.0.0.1", 9999)):
    """Build a minimal HTTP ASGI scope dict for testing."""
    return {
        "type": "http",
        "method": method,
        "path": path,
        "query_string": b"",
        "headers": headers or [],
        "client": client,
    }
 
 
# ============================================================
# HTTP SCOPE
# ============================================================
 
def test_http_scope_status_passthrough():
    """HTTP response status code is captured and passed through to the caller."""
    async def _run():
        received = []
 
        async def mock_app(scope, receive, send):
            await send({"type": "http.response.start", "status": 201, "headers": []})
            await send({"type": "http.response.body", "body": b""})
 
        async def mock_send(message):
            received.append(message)
 
        middleware = DiagnosticsMiddleware(mock_app, domain="TEST_HTTP")
        await middleware(_make_http_scope(), None, mock_send)
        assert any(m.get("status") == 201 for m in received), (
            f"HTTP response status must be passed through; received: {received!r}"
        )
 
    asyncio.run(_run())
 
 
# ============================================================
# NON-HTTP SCOPES
# ============================================================
 
def test_websocket_passthrough():
    """WebSocket scope is forwarded to the app without any HTTP logging."""
    async def _run():
        called = []
 
        async def mock_app(scope, receive, send):
            called.append(scope["type"])
 
        scope = {"type": "websocket", "path": "/ws"}
        middleware = DiagnosticsMiddleware(mock_app, domain="TEST_HTTP")
        await middleware(scope, None, None)
        assert called == ["websocket"], (
            f"WebSocket scope must pass through unchanged; app called with: {called!r}"
        )
 
    asyncio.run(_run())
 
 
def test_lifespan_scope_passthrough():
    """lifespan scope is forwarded to the app without any HTTP logging."""
    async def _run():
        called = []
 
        async def mock_app(scope, receive, send):
            called.append(scope["type"])
 
        scope = {"type": "lifespan", "asgi": {"version": "3.0"}}
        middleware = DiagnosticsMiddleware(mock_app, domain="TEST_HTTP")
        await middleware(scope, None, None)
        assert called == ["lifespan"], (
            f"lifespan scope must pass through unchanged; app called with: {called!r}"
        )
 
    asyncio.run(_run())
 
 
# ============================================================
# EXCEPTION HANDLING
# ============================================================
 
def test_exception_reraise():
    """Application exceptions are logged at L2 and then re-raised unchanged."""
    async def _run():
        async def error_app(scope, receive, send):
            await send({"type": "http.response.start", "status": 500, "headers": []})
            raise ValueError("deliberate-test-error")
 
        async def mock_send(message):
            pass
 
        middleware = DiagnosticsMiddleware(error_app, domain="TEST_HTTP")
        raised = None
        try:
            await middleware(_make_http_scope(method="POST", path="/err"), None, mock_send)
        except ValueError as exc:
            raised = exc
        assert raised is not None, "ValueError must be re-raised by middleware"
        assert "deliberate-test-error" in str(raised), (
            f"Expected 'deliberate-test-error' in exception, got: {raised!r}"
        )
 
    asyncio.run(_run())
 
 
# ============================================================
# HEADER REDACTION (FS-17)
# ============================================================
 
def test_http_header_redaction():
    """Sensitive request headers appear as [REDACTED] in log output; others are not redacted."""
    logged = []
    old_emit = _ud._emit
    old_redact = _ud.HTTP_REDACT_HEADERS
    old_levels = _ud.DEBUG_LEVELS.copy()
 
    def _capture(line, level, record=None):
        logged.append(line)
 
    _ud._emit = _capture
    _ud.HTTP_REDACT_HEADERS = frozenset(["authorization", "cookie"])
    _ud.DEBUG_LEVELS["_REDACT_TEST_"] = 9
 
    async def _run():
        async def mock_app(scope, receive, send):
            await send({"type": "http.response.start", "status": 200, "headers": []})
            await send({"type": "http.response.body", "body": b""})
 
        async def mock_send(message):
            pass
 
        scope = _make_http_scope(
            path="/secure",
            headers=[
                (b"authorization", b"Bearer secret-token"),
                (b"content-type", b"application/json"),
            ],
        )
        middleware = DiagnosticsMiddleware(mock_app, domain="_REDACT_TEST_")
        await middleware(scope, None, mock_send)
 
    try:
        asyncio.run(_run())
    finally:
        _ud._emit = old_emit
        _ud.HTTP_REDACT_HEADERS = old_redact
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
 
    auth_lines = [l for l in logged if "authorization" in l.lower()]
    assert auth_lines, (
        f"Expected authorization header in log output; got: {logged!r}"
    )
    for line in auth_lines:
        assert "[REDACTED]" in line, (
            f"Authorization header must be [REDACTED], got: {line!r}"
        )
    ct_lines = [l for l in logged if "content-type" in l.lower()]
    for line in ct_lines:
        assert "[REDACTED]" not in line, (
            f"content-type header must NOT be redacted, got: {line!r}"
        )
 
 
# ============================================================
# ELAPSED TIME
# ============================================================
 
def test_elapsed_time_in_log():
    """Elapsed milliseconds appear in the completion log line for HTTP requests."""
    logged = []
    old_emit = _ud._emit
    old_levels = _ud.DEBUG_LEVELS.copy()
 
    def _capture(line, level, record=None):
        logged.append(line)
 
    _ud._emit = _capture
    _ud.DEBUG_LEVELS["_ELAPSED_TEST_"] = 9
 
    async def _run():
        async def mock_app(scope, receive, send):
            await send({"type": "http.response.start", "status": 200, "headers": []})
            await send({"type": "http.response.body", "body": b""})
 
        async def mock_send(message):
            pass
 
        middleware = DiagnosticsMiddleware(mock_app, domain="_ELAPSED_TEST_")
        await middleware(_make_http_scope(path="/timed"), None, mock_send)
 
    try:
        asyncio.run(_run())
    finally:
        _ud._emit = old_emit
        _ud.DEBUG_LEVELS.clear()
        _ud.DEBUG_LEVELS.update(old_levels)
 
    ms_lines = [l for l in logged if "ms" in l]
    assert ms_lines, (
        f"Expected elapsed time in milliseconds in log output, got: {logged!r}"
    )