122 lines
3.5 KiB
Python
122 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from io import StringIO
|
|
|
|
import pytest
|
|
|
|
from app.core.logging import PlanetContextFilter, PlanetFormatter, get_logger
|
|
from app.core.request_context import set_request_id
|
|
from app.services import business_logs
|
|
|
|
|
|
def _capture_output(callback):
|
|
stream = StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
handler.setFormatter(PlanetFormatter(datefmt="%Y-%m-%d %H:%M:%S"))
|
|
handler.addFilter(PlanetContextFilter())
|
|
|
|
adapter = get_logger("tests.logging")
|
|
target_logger = adapter.logger
|
|
original_handlers = list(target_logger.handlers)
|
|
original_level = target_logger.level
|
|
original_propagate = target_logger.propagate
|
|
|
|
target_logger.handlers = [handler]
|
|
target_logger.setLevel(logging.INFO)
|
|
target_logger.propagate = False
|
|
|
|
try:
|
|
callback(adapter)
|
|
finally:
|
|
handler.flush()
|
|
target_logger.handlers = original_handlers
|
|
target_logger.setLevel(original_level)
|
|
target_logger.propagate = original_propagate
|
|
|
|
return stream.getvalue()
|
|
|
|
|
|
def test_structured_logger_injects_request_id_and_event():
|
|
set_request_id("req-test-123")
|
|
try:
|
|
output = _capture_output(
|
|
lambda logger: logger.info_event(
|
|
"collector started",
|
|
event="collector.run.started",
|
|
context={"collector_name": "bgp_news"},
|
|
)
|
|
)
|
|
finally:
|
|
set_request_id(None)
|
|
|
|
assert "request_id=req-test-123" in output
|
|
assert "event=collector.run.started" in output
|
|
assert "service=backend" in output
|
|
assert '"collector_name": "bgp_news"' in output
|
|
|
|
|
|
def test_structured_logger_redacts_sensitive_text_and_context():
|
|
set_request_id("req-test-redact")
|
|
try:
|
|
output = _capture_output(
|
|
lambda logger: logger.error_event(
|
|
"Authorization: Bearer super-secret-token",
|
|
event="auth.token.failed",
|
|
context={
|
|
"token": "plain-secret",
|
|
"nested": {"password": "hunter2"},
|
|
"safe": "visible",
|
|
},
|
|
)
|
|
)
|
|
finally:
|
|
set_request_id(None)
|
|
|
|
assert "super-secret-token" not in output
|
|
assert "plain-secret" not in output
|
|
assert "hunter2" not in output
|
|
assert "[REDACTED]" in output
|
|
assert '"safe": "visible"' in output
|
|
|
|
|
|
def test_business_context_redacts_nested_sensitive_values():
|
|
context = business_logs.build_business_context(
|
|
{
|
|
"provider": "openai",
|
|
"api_key": "sk-secret",
|
|
"nested": {
|
|
"token": "plain-token",
|
|
"safe": "visible",
|
|
},
|
|
}
|
|
)
|
|
|
|
assert context["api_key"] == "[REDACTED]"
|
|
assert context["nested"]["token"] == "[REDACTED]"
|
|
assert context["nested"]["safe"] == "visible"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_emit_business_log_persists_sanitized_system_event(monkeypatch):
|
|
events = []
|
|
|
|
async def fake_record_system_log(**payload):
|
|
events.append(payload)
|
|
|
|
monkeypatch.setattr(business_logs, "record_system_log", fake_record_system_log)
|
|
|
|
await business_logs.emit_business_log(
|
|
get_logger("tests.business"),
|
|
event="ai.provider.analyze.success",
|
|
message="AI request completed",
|
|
category="ai",
|
|
context={"model": "gpt-test", "api_key": "sk-secret"},
|
|
)
|
|
|
|
assert events[0]["event"] == "ai.provider.analyze.success"
|
|
assert events[0]["category"] == "ai"
|
|
assert events[0]["context"]["model"] == "gpt-test"
|
|
assert events[0]["context"]["api_key"] == "[REDACTED]"
|