kairee
e2cdf997af
Implement P0 of the reaction dashboard: config loading with hot reload, SQLite migrations with backups, single-writer queue, read-only Bilibili fav client, folder import/diff sync, invalid-resource handling, blacklist seed, scheduler, and JSON redacted logging. Includes offline unit and transport-level contract tests; all existing and new tests pass.
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
# -*- 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()
|