"""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