# -*- coding: utf-8 -*- """JSON 日志脱敏测试(不联网): uv run python tests/test_logging.py""" import io import json import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") from reaction_board.logging_json import JsonLogger, redact # noqa: E402 SECRET_COOKIE = "SESSDATA=abc%2C1800000000%2Cdef; bili_jct=csrf123; buvid3=device123" SECRET_TOKEN = "123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghi" def test_redact_keys_and_nested(): data = { "cookie": SECRET_COOKIE, "nested": { "csrf": "csrf123", "bot_token": SECRET_TOKEN, "authorization": "Bearer abc.def.ghi", "safe": "visible", }, "items": [{"sessdata": "x"}, {"note": "ok"}], } out = redact(data) assert out["cookie"] == "***" assert out["nested"]["csrf"] == "***" assert out["nested"]["bot_token"] == "***" assert out["nested"]["authorization"] == "***" assert out["nested"]["safe"] == "visible" assert out["items"][0]["sessdata"] == "***" assert out["items"][1]["note"] == "ok" def test_redact_secrets_inside_ordinary_strings(): text = ( f"request failed cookie={SECRET_COOKIE} token={SECRET_TOKEN} " "Authorization: Bearer abc.def" ) out = redact({"error": text}) assert "abc%2C1800000000" not in out["error"] assert "csrf123" not in out["error"] assert SECRET_TOKEN not in out["error"] assert "***" in out["error"] def test_logger_emits_json_lines_without_secrets(): stream = io.StringIO() logger = JsonLogger(stream) logger.info( "test.event", cookie=SECRET_COOKIE, message=f"bad cookie {SECRET_COOKIE}", nested={"bili_jct": "csrf123", "count": 2}, list_value=[SECRET_TOKEN], ) lines = [line for line in stream.getvalue().splitlines() if line] assert len(lines) == 1 payload = json.loads(lines[0]) assert payload["level"] == "info" assert payload["event"] == "test.event" assert payload["cookie"] == "***" assert payload["nested"]["bili_jct"] == "***" assert payload["list_value"] == ["***"] assert "SESSDATA=***" in lines[0] assert SECRET_TOKEN not in lines[0] assert "1800000000" not in lines[0] def test_logger_levels(): stream = io.StringIO() logger = JsonLogger(stream) logger.debug("d", x=1) logger.warning("w", x=2) logger.error("e", x=3) rows = [json.loads(line) for line in stream.getvalue().splitlines()] assert [r["level"] for r in rows] == ["debug", "warning", "error"] assert [r["event"] for r in rows] == ["d", "w", "e"] def main(): test_redact_keys_and_nested() test_redact_secrets_inside_ordinary_strings() test_logger_emits_json_lines_without_secrets() test_logger_levels() print("OK: JSON 日志脱敏测试全部通过") if __name__ == "__main__": main()