130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""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
|
|
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") == "<binary 13 bytes>"
|
|
assert safe_body(b"--boundary\r\n...", "multipart/form-data; boundary=boundary") == (
|
|
"<binary 15 bytes>"
|
|
)
|
|
|
|
|
|
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)
|
|
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 == []
|