"""Unit tests for shared HTTP payload logging (core/http_logging.py). Covers header redaction, JSON secret masking, truncation, binary summarization, and that the httpx event hooks emit payload-bearing http_request/http_response events at DEBUG. """ from __future__ import annotations import json import httpx import pytest import structlog from structlog.testing import capture_logs from contract_check.core import http_logging as hhttp from contract_check.core.http_logging import ( http_log_event_hooks, redact_headers, redact_json, safe_body, ) def test_redact_headers_masks_secret_bearers() -> None: headers = { "Authorization": "Basic abcdefgh", "X-API-Key": "short", "Content-Type": "application/json", } redacted = redact_headers(headers) assert redacted["authorization"] == "Basi***" assert redacted["x-api-key"] == "shor***" assert redacted["content-type"] == "application/json" def test_redact_json_masks_secret_keys_recursively() -> None: payload = { "username": "san", "password": "hunter2", "nested": {"access_token": "jwt", "items": [{"client_secret": "s"}]}, "keep": "value", } redacted = redact_json(payload) assert redacted["password"] == "***" assert redacted["nested"]["access_token"] == "***" assert redacted["nested"]["items"][0]["client_secret"] == "***" assert redacted["username"] == "san" assert redacted["keep"] == "value" def test_safe_body_redacts_json_bodies() -> None: body = json.dumps({"grant_type": "password", "password": "hunter2"}).encode() rendered = safe_body(body, "application/json; charset=utf-8") assert rendered is not None assert "hunter2" not in rendered assert json.loads(rendered)["password"] == "***" def test_safe_body_redacts_urlencoded_bodies() -> None: rendered = safe_body(b"username=san&password=hunter2", "application/x-www-form-urlencoded") assert rendered == "username=san&password=***" def test_safe_body_summarizes_binary_payloads() -> None: assert safe_body(b"%PDF-1.7 fake", "application/pdf") == "" assert safe_body(b"--boundary\r\n...", "multipart/form-data; boundary=boundary") == ( "" ) def test_safe_body_truncates_long_text() -> None: rendered = safe_body("x" * 500, "text/plain", max_chars=100) assert rendered is not None assert rendered.startswith("x" * 100) assert "+400 chars" in rendered def test_safe_body_handles_none_and_plain_text() -> None: assert safe_body(None, "application/json") is None assert safe_body("hello", "text/plain") == "hello" async def test_event_hooks_log_request_and_response_payloads( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(hhttp, "is_debug_enabled", lambda: True) # Other tests may configure structlog with a filtering wrapper at INFO, # which causes ``capture_logs`` to drop DEBUG records. Reconfigure to a # non-filtering wrapper so the DEBUG http_request/http_response events are # always captured. _config = structlog.get_config() monkeypatch.setattr( hhttp, "get_logger", lambda name: structlog.get_logger(name), ) structlog.configure( processors=_config["processors"], wrapper_class=structlog.stdlib.BoundLogger, logger_factory=_config["logger_factory"], cache_logger_on_first_use=False, ) with capture_logs() as logs: transport = httpx.MockTransport( lambda request: httpx.Response(200, json={"ok": True, "token": "jwt-secret"}) # noqa: S106 ) async with httpx.AsyncClient( transport=transport, headers={"Authorization": "Bearer supersecretjwt"}, event_hooks=http_log_event_hooks(service="test-svc"), ) as client: response = await client.post( "https://api.test/v1/things", json={"name": "thing", "password": "hunter2"}, ) assert response.status_code == 200 events = {entry.get("event"): entry for entry in logs} request_entry = events["http_request"] assert request_entry["service"] == "test-svc" assert request_entry["method"] == "POST" assert request_entry["url"] == "https://api.test/v1/things" assert request_entry["headers"]["authorization"] == "Bear***" assert request_entry["headers"]["content-type"] == "application/json" assert "hunter2" not in request_entry["body"] assert json.loads(request_entry["body"])["password"] == "***" response_entry = events["http_response"] assert response_entry["status_code"] == 200 assert response_entry["duration_ms"] is not None assert "jwt-secret" not in response_entry["body"] assert json.loads(response_entry["body"])["token"] == "***" async def test_event_hooks_silent_without_debug(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(hhttp, "is_debug_enabled", lambda: False) with capture_logs() as logs: transport = httpx.MockTransport(lambda request: httpx.Response(204)) async with httpx.AsyncClient( transport=transport, event_hooks=http_log_event_hooks(service="test-svc"), ) as client: await client.get("https://api.test/v1/things") assert logs == [] async def test_event_hooks_handle_streaming_multipart_body( monkeypatch: pytest.MonkeyPatch, ) -> None: """Multipart uploads stream their body; the request hook must not crash.""" monkeypatch.setattr(hhttp, "is_debug_enabled", lambda: True) _config = structlog.get_config() monkeypatch.setattr( hhttp, "get_logger", lambda name: structlog.get_logger(name), ) structlog.configure( processors=_config["processors"], wrapper_class=structlog.stdlib.BoundLogger, logger_factory=_config["logger_factory"], cache_logger_on_first_use=False, ) with capture_logs() as logs: transport = httpx.MockTransport(lambda request: httpx.Response(202, json={"ok": True})) async with httpx.AsyncClient( transport=transport, event_hooks=http_log_event_hooks(service="test-svc"), ) as client: response = await client.post( "https://api.test/v1/documents", files={"file": ("contract.pdf", b"%PDF-1.7 fake", "application/pdf")}, ) assert response.status_code == 202 request_entry = next(entry for entry in logs if entry.get("event") == "http_request") assert request_entry["method"] == "POST" assert request_entry["body"] == ""