78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from app.core.logging import SAFE_EXTRA_KEYS, JsonFormatter, SafeTextFormatter, request_id_var
|
|
|
|
|
|
def reserved_logrecord_attributes() -> set[str]:
|
|
"""Attribute names ``logging`` sets on every record and refuses to let ``extra`` overwrite."""
|
|
record = logging.LogRecord("n", logging.INFO, "p", 1, "m", None, None)
|
|
return set(record.__dict__) | {"message", "asctime"}
|
|
|
|
|
|
def test_no_safe_key_collides_with_a_reserved_logrecord_attribute() -> None:
|
|
"""``extra={"filename": ...}`` raises KeyError and would read back the source file."""
|
|
assert not SAFE_EXTRA_KEYS & reserved_logrecord_attributes()
|
|
|
|
|
|
@pytest.mark.parametrize("key", sorted(SAFE_EXTRA_KEYS))
|
|
def test_every_safe_key_can_actually_be_logged(key: str, caplog: pytest.LogCaptureFixture) -> None:
|
|
logger = logging.getLogger("app.test")
|
|
with caplog.at_level(logging.INFO):
|
|
logger.info("event", extra={key: "value"})
|
|
assert caplog.records
|
|
|
|
|
|
class TestRedaction:
|
|
def _record(self, **extra: object) -> logging.LogRecord:
|
|
record = logging.LogRecord("app.test", logging.INFO, "p", 1, "event", None, None)
|
|
for key, value in extra.items():
|
|
setattr(record, key, value)
|
|
return record
|
|
|
|
def test_json_formatter_emits_only_allowlisted_keys(self) -> None:
|
|
record = self._record(file_name="cv.pdf", resume_text="SENSITIVE PERSONAL DATA")
|
|
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
|
|
assert payload["file_name"] == "cv.pdf"
|
|
assert "resume_text" not in payload
|
|
assert "SENSITIVE" not in json.dumps(payload)
|
|
|
|
def test_text_formatter_emits_only_allowlisted_keys(self) -> None:
|
|
record = self._record(file_name="cv.pdf", job_description="SENSITIVE JD TEXT")
|
|
|
|
line = SafeTextFormatter().format(record)
|
|
|
|
assert "cv.pdf" in line
|
|
assert "SENSITIVE" not in line
|
|
|
|
def test_json_formatter_includes_the_request_id(self) -> None:
|
|
token = request_id_var.set("req-abc")
|
|
try:
|
|
payload = json.loads(JsonFormatter().format(self._record()))
|
|
finally:
|
|
request_id_var.reset(token)
|
|
|
|
assert payload["request_id"] == "req-abc"
|
|
|
|
def test_exception_is_logged_as_type_and_frames_not_message(self) -> None:
|
|
"""Provider errors can echo request content, so the message itself is dropped."""
|
|
try:
|
|
raise ValueError("resume text leaked into the exception message")
|
|
except ValueError:
|
|
import sys
|
|
|
|
record = self._record()
|
|
record.exc_info = sys.exc_info()
|
|
|
|
payload = json.loads(JsonFormatter().format(record))
|
|
|
|
assert payload["exc_type"] == "ValueError"
|
|
assert payload["exc_frames"]
|
|
assert "leaked" not in json.dumps(payload)
|