164 lines
6.3 KiB
Python
164 lines
6.3 KiB
Python
"""logging_setup tests: masking, rotation handler, idempotency."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import pytest
|
|
|
|
from us2kr.config import Settings
|
|
from us2kr.logging_setup import (
|
|
_HANDLER_TAG,
|
|
SecretMaskingFilter,
|
|
SecretMaskingFormatter,
|
|
mask_secrets,
|
|
register_secret_values,
|
|
setup_logging,
|
|
)
|
|
|
|
SK = "sk-cp-abcdefghijklmnopqrstuvwxyz"
|
|
BOT = "123456789:AAH-abcdefghijklmnopqrstuvwxyz0123"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_registered_secrets():
|
|
yield
|
|
register_secret_values()
|
|
|
|
|
|
@pytest.fixture
|
|
def root_cleanup():
|
|
root = logging.getLogger()
|
|
before_level = root.level
|
|
yield root
|
|
for h in list(root.handlers):
|
|
if getattr(h, _HANDLER_TAG, False):
|
|
root.removeHandler(h)
|
|
h.close()
|
|
root.setLevel(before_level)
|
|
|
|
|
|
def our_handlers(root):
|
|
return [h for h in root.handlers if getattr(h, _HANDLER_TAG, False)]
|
|
|
|
|
|
def test_mask_secrets_patterns():
|
|
assert mask_secrets(f"key={SK} token={BOT} other=abc") == "key=*** token=*** other=abc"
|
|
assert mask_secrets("chat_id=-1001234 ok") == "chat_id=-1001234 ok" # not a token shape
|
|
assert mask_secrets("sk-ab") == "sk-ab" # too short to be a key
|
|
|
|
|
|
def test_filter_masks_msg_and_args():
|
|
rec = logging.LogRecord("x", logging.INFO, __file__, 1, f"Bearer {SK} for %s and %s", (BOT, 42), None)
|
|
assert SecretMaskingFilter().filter(rec) is True
|
|
assert rec.msg == "Bearer *** for %s and %s"
|
|
assert rec.args == ("***", 42)
|
|
assert rec.getMessage() == "Bearer *** for *** and 42"
|
|
|
|
|
|
def test_setup_logging_creates_file_handler_and_masks(tmp_path, root_cleanup, capsys):
|
|
settings = Settings(_env_file=None, US2KR_LOG_DIR=str(tmp_path / "logs"), US2KR_LOG_LEVEL="DEBUG")
|
|
setup_logging(settings)
|
|
|
|
root = root_cleanup
|
|
assert root.level == logging.DEBUG
|
|
handlers = our_handlers(root)
|
|
assert len(handlers) == 2
|
|
kinds = {type(h).__name__ for h in handlers}
|
|
assert kinds == {"StreamHandler", "TimedRotatingFileHandler"}
|
|
file_handler = next(h for h in handlers if type(h).__name__ == "TimedRotatingFileHandler")
|
|
assert file_handler.when == "MIDNIGHT" and file_handler.backupCount == 14
|
|
|
|
logging.getLogger("us2kr.test").info("using key %s", SK)
|
|
try:
|
|
raise RuntimeError(f"auth failed for {BOT}")
|
|
except RuntimeError:
|
|
logging.getLogger("us2kr.test").exception("boom")
|
|
for h in handlers:
|
|
h.flush()
|
|
|
|
text = (tmp_path / "logs" / "us2kr.log").read_text(encoding="utf-8")
|
|
assert "using key ***" in text
|
|
assert SK not in text and BOT not in text # traceback text masked by the formatter too
|
|
assert "RuntimeError: auth failed for ***" in text
|
|
assert SK not in capsys.readouterr().err
|
|
|
|
|
|
def test_setup_logging_is_idempotent_and_quiets_noisy_loggers(tmp_path, root_cleanup):
|
|
settings = Settings(_env_file=None, US2KR_LOG_DIR=str(tmp_path / "logs"), US2KR_LOG_LEVEL="warning")
|
|
setup_logging(settings)
|
|
setup_logging(settings)
|
|
assert len(our_handlers(root_cleanup)) == 2
|
|
assert root_cleanup.level == logging.WARNING
|
|
for name in ("httpx", "apscheduler.executors", "yfinance", "peewee"):
|
|
assert logging.getLogger(name).level == logging.WARNING
|
|
|
|
|
|
def test_setup_logging_unknown_level_defaults_to_info(tmp_path, root_cleanup):
|
|
settings = Settings(_env_file=None, US2KR_LOG_DIR=str(tmp_path / "logs"), US2KR_LOG_LEVEL="loud")
|
|
setup_logging(settings)
|
|
assert root_cleanup.level == logging.INFO
|
|
|
|
|
|
# ------------------------------------------------------------------ URL query secrets / exact values
|
|
FRED_KEY = "0123456789abcdef0123456789abcdef"
|
|
|
|
|
|
def test_mask_secrets_url_query_parameter():
|
|
msg = (
|
|
"400 Client Error: Bad Request for url: https://api.stlouisfed.org/fred/releases/dates"
|
|
f"?realtime_start=2026-08-24&limit=1000&api_key={FRED_KEY}&file_type=json"
|
|
)
|
|
out = mask_secrets(msg)
|
|
assert FRED_KEY not in out
|
|
assert "api_key=***&file_type=json" in out
|
|
assert "limit=1000" in out # non-secret params untouched
|
|
# urllib3 ConnectionError style (path only, key last, no trailing param) and other param names
|
|
conn = (
|
|
f"Max retries exceeded with url: /fred/releases/dates?file_type=json&api_key={FRED_KEY} (Caused by X)"
|
|
)
|
|
expected = "Max retries exceeded with url: /fred/releases/dates?file_type=json&api_key=*** (Caused by X)"
|
|
assert mask_secrets(conn) == expected
|
|
assert mask_secrets("https://x/api?token=abc123&x=1") == "https://x/api?token=***&x=1"
|
|
assert mask_secrets("https://x/api?APIKEY=abc123") == "https://x/api?APIKEY=***"
|
|
# a bare word 'key=' outside a query string is not a URL secret
|
|
assert mask_secrets("chat_id=-1001234 key=value") == "chat_id=-1001234 key=value"
|
|
|
|
|
|
def test_mask_secrets_registered_exact_values():
|
|
register_secret_values("plainvalue-without-shape-1234", None, "", "short")
|
|
out = mask_secrets("x plainvalue-without-shape-1234 y short")
|
|
assert out == "x *** y short" # too-short values are not registered (would shred normal text)
|
|
register_secret_values()
|
|
assert "plainvalue-without-shape-1234" in mask_secrets("x plainvalue-without-shape-1234 y")
|
|
|
|
|
|
def test_formatter_masks_requests_style_traceback_with_fred_key():
|
|
formatter = SecretMaskingFormatter("%(message)s")
|
|
url = f"https://api.stlouisfed.org/fred/releases/dates?realtime_start=2026-08-24&api_key={FRED_KEY}&file_type=json"
|
|
try:
|
|
raise ConnectionError(f"Max retries exceeded with url: {url}")
|
|
except ConnectionError:
|
|
rec = logging.LogRecord("x", logging.ERROR, __file__, 1, "collector %s failed", ("events",), None)
|
|
import sys
|
|
|
|
rec.exc_info = sys.exc_info()
|
|
text = formatter.format(rec)
|
|
assert FRED_KEY not in text
|
|
assert "api_key=***&file_type=json" in text
|
|
assert "collector events failed" in text
|
|
|
|
|
|
def test_setup_logging_registers_configured_secret_values(tmp_path, root_cleanup):
|
|
odd_key = "FREDKEY-with-unusual-shape-XYZ" # matches no regex shape
|
|
settings = Settings(
|
|
_env_file=None, US2KR_LOG_DIR=str(tmp_path / "logs"), FRED_API_KEY=odd_key, MINIMAX_API_KEY=SK
|
|
)
|
|
setup_logging(settings)
|
|
logging.getLogger("us2kr.test").error("failed with %s and %s", odd_key, SK)
|
|
for h in our_handlers(root_cleanup):
|
|
h.flush()
|
|
text = (tmp_path / "logs" / "us2kr.log").read_text(encoding="utf-8")
|
|
assert odd_key not in text and SK not in text
|
|
assert "failed with *** and ***" in text
|