Tyler
47e0f80786
Cyclone previously emitted stdlib default-formatted log lines that
operators couldn't parse with anything beyond grep. SP18 replaces
that with:
- JsonFormatter: newline-delimited JSON, ISO-8601 ms timestamps,
structured "extra" dict, exception tracebacks serialized as a
single string.
- CycloneDevFormatter: tabular format for tail -f in dev.
- PiiScrubber: logging.Filter that redacts NPIs, SSNs, DOBs,
patient names — both inline in the message ("npi 1881068062")
and via PHI-keyed extras ({"dob": "1980-04-12"}).
- setup_logging(): idempotent entry point used by the API lifespan
and CLI main; respects CYCLONE_LOG_LEVEL/FILE/JSON/NO_PII_SCRUB.
- CLI --log-format=json|dev + --log-file=… flags.
- Migrate 9 highest-value log sites in scheduler/backup_scheduler/
backup_service to use extra={...} (input_filename, claims, parser,
backup_id, db_fingerprint, etc.).
34 new tests (test_logging_formatter + test_logging_scrubber +
test_logging_setup). All 867 backend tests pass.
158 lines
4.7 KiB
Python
158 lines
4.7 KiB
Python
"""SP18 — PII scrubber tests.
|
|
|
|
Covers each PHI pattern, the false-positive guard, and the
|
|
disable toggle.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from cyclone.logging_config import (
|
|
PiiScrubber,
|
|
get_scrubber,
|
|
setup_logging,
|
|
)
|
|
|
|
|
|
def _make_record(msg: str, extras: dict | None = None) -> logging.LogRecord:
|
|
record = logging.getLogger("test.scrub").makeRecord(
|
|
name="test.scrub", level=logging.INFO, fn="t.py", lno=1,
|
|
msg=msg, args=(), exc_info=None,
|
|
)
|
|
if extras:
|
|
for k, v in extras.items():
|
|
setattr(record, k, v)
|
|
return record
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_scrubber():
|
|
"""Make sure the scrubber is enabled + on the root after each test."""
|
|
yield
|
|
get_scrubber().enable()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# NPI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_scrubs_ten_digit_npi_in_message():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("processed claim with npi 1881068062 ok")
|
|
assert scrubber.filter(rec) is True
|
|
assert rec.getMessage() == "processed claim with npi <redacted:npi> ok"
|
|
|
|
|
|
def test_scrubs_npi_in_extras():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("ok", extras={"provider_npi": "1881068062"})
|
|
scrubber.filter(rec)
|
|
assert rec.provider_npi == "<redacted:npi>"
|
|
|
|
|
|
def test_does_not_scrub_short_numbers():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("processed 5 claims in 2 batches")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "processed 5 claims in 2 batches"
|
|
|
|
|
|
def test_does_not_scrub_eleven_digit_numbers():
|
|
"""11+ digit numbers aren't NPIs — leave them alone."""
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("control number 12345678901")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "control number 12345678901"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SSN
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_scrubs_dashed_ssn_in_message():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("ssn=123-45-6789 detected")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "ssn=<redacted:ssn> detected"
|
|
|
|
|
|
def test_scrubs_undashed_ssn_at_phrase_boundary():
|
|
"""A bare 9-digit number is ambiguous — we scrub it when followed
|
|
by whitespace/punctuation/closing paren/brace/comma (so we don't
|
|
hit claim control numbers or zip codes mid-sentence)."""
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("ssn 123456789 on file")
|
|
scrubber.filter(rec)
|
|
assert "<redacted:ssn>" in rec.getMessage()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DOB
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_scrubs_dob_field():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("dob=1980-04-12 verified")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "dob=<redacted:dob> verified"
|
|
|
|
|
|
def test_scrubs_dob_in_extras():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("ok", extras={"date_of_birth": "1980-04-12"})
|
|
scrubber.filter(rec)
|
|
assert rec.date_of_birth == "<redacted:dob>"
|
|
|
|
|
|
def test_does_not_scrub_bare_iso_date():
|
|
"""A YYYY-MM-DD without a dob= prefix isn't necessarily PHI."""
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("parsed at 2026-06-21")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "parsed at 2026-06-21"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Patient name
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_scrubs_patient_name_field():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record('patient_name="John Doe" verified')
|
|
scrubber.filter(rec)
|
|
assert "John Doe" not in rec.getMessage()
|
|
assert "<redacted:patient_name>" in rec.getMessage()
|
|
|
|
|
|
def test_does_not_scrub_random_words():
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("the parser ran successfully")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "the parser ran successfully"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Disable toggle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_scrubber_disabled_leaves_message_intact():
|
|
scrubber = PiiScrubber()
|
|
scrubber.disable()
|
|
rec = _make_record("npi 1881068062 ok")
|
|
scrubber.filter(rec)
|
|
assert rec.getMessage() == "npi 1881068062 ok"
|
|
|
|
|
|
def test_scrubber_does_not_crash_on_unusual_records():
|
|
"""Even with weird attribute combinations the filter returns True."""
|
|
scrubber = PiiScrubber()
|
|
rec = _make_record("ok", extras={"weird": object()})
|
|
assert scrubber.filter(rec) is True
|