qzl
141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from collections.abc import Iterator
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import pytest
|
|
import structlog
|
|
|
|
from core.config.settings import Settings
|
|
from core.logging.config import build_logging_config, configure_logging
|
|
|
|
|
|
def _get_handlers(config: dict[str, object]) -> dict[str, dict[str, object]]:
|
|
return cast(dict[str, dict[str, object]], config["handlers"])
|
|
|
|
|
|
def test_build_logging_config_time_rotation(tmp_path: Path) -> None:
|
|
settings = Settings()
|
|
runtime = settings.runtime.model_copy(
|
|
update={
|
|
"log_dir": str(tmp_path),
|
|
"log_error_dir": str(tmp_path / "errors"),
|
|
"log_rotation": "time",
|
|
}
|
|
)
|
|
|
|
config = build_logging_config(runtime)
|
|
handlers = _get_handlers(config)
|
|
|
|
assert handlers["file"]["class"] == "logging.handlers.TimedRotatingFileHandler"
|
|
assert handlers["error"]["class"] == "logging.handlers.TimedRotatingFileHandler"
|
|
assert handlers["error"]["level"] == "ERROR"
|
|
|
|
|
|
def test_build_logging_config_size_rotation(tmp_path: Path) -> None:
|
|
settings = Settings()
|
|
runtime = settings.runtime.model_copy(
|
|
update={
|
|
"log_dir": str(tmp_path),
|
|
"log_error_dir": str(tmp_path / "errors"),
|
|
"log_rotation": "size",
|
|
"log_rotation_max_bytes": 2048,
|
|
}
|
|
)
|
|
|
|
config = build_logging_config(runtime)
|
|
handlers = _get_handlers(config)
|
|
|
|
assert handlers["file"]["class"] == "logging.handlers.RotatingFileHandler"
|
|
assert handlers["error"]["class"] == "logging.handlers.RotatingFileHandler"
|
|
assert handlers["file"]["maxBytes"] == 2048
|
|
|
|
|
|
def test_build_logging_config_plain_formatter_when_disabled(tmp_path: Path) -> None:
|
|
settings = Settings()
|
|
runtime = settings.runtime.model_copy(
|
|
update={
|
|
"log_dir": str(tmp_path),
|
|
"log_error_dir": str(tmp_path / "errors"),
|
|
"log_json": False,
|
|
}
|
|
)
|
|
|
|
config = build_logging_config(runtime)
|
|
handlers = _get_handlers(config)
|
|
|
|
assert handlers["file"]["formatter"] == "plain"
|
|
assert handlers["error"]["formatter"] == "plain"
|
|
|
|
|
|
def _read_last_log_entry(log_path: Path) -> dict[str, object]:
|
|
assert log_path.exists(), f"Expected log file at {log_path}"
|
|
entries = [
|
|
json.loads(line) for line in log_path.read_text().splitlines() if line.strip()
|
|
]
|
|
assert entries, "Expected at least one log entry in app.log"
|
|
return entries[-1]
|
|
|
|
|
|
def _flush_root_handlers() -> None:
|
|
root_logger = logging.getLogger()
|
|
for handler in root_logger.handlers:
|
|
if hasattr(handler, "flush"):
|
|
handler.flush()
|
|
|
|
|
|
@pytest.fixture
|
|
def configured_logging(tmp_path: Path) -> Iterator[Path]:
|
|
settings = Settings()
|
|
runtime = settings.runtime.model_copy(
|
|
update={
|
|
"log_dir": str(tmp_path),
|
|
"log_error_dir": str(tmp_path / "errors"),
|
|
"log_rotation": "size",
|
|
"log_rotation_max_bytes": 2048,
|
|
"log_json": True,
|
|
}
|
|
)
|
|
root_logger = logging.getLogger()
|
|
original_handlers = root_logger.handlers[:]
|
|
original_level = root_logger.level
|
|
|
|
configure_logging(settings.model_copy(update={"runtime": runtime}))
|
|
|
|
yield tmp_path
|
|
|
|
for handler in root_logger.handlers:
|
|
handler.close()
|
|
root_logger.handlers = original_handlers
|
|
root_logger.setLevel(original_level)
|
|
structlog.reset_defaults()
|
|
|
|
|
|
def test_stdlib_logging_redacts_sensitive_fields(configured_logging: Path) -> None:
|
|
logger = logging.getLogger("tests.stdlib")
|
|
logger.info("login", extra={"password": "secret", "token": "abc"})
|
|
|
|
_flush_root_handlers()
|
|
|
|
log_path = configured_logging / "app.log"
|
|
entry = _read_last_log_entry(log_path)
|
|
|
|
assert entry["password"] == "[REDACTED]"
|
|
assert entry["token"] == "[REDACTED]"
|
|
|
|
|
|
def test_structlog_redacts_sensitive_fields(configured_logging: Path) -> None:
|
|
logger = structlog.get_logger("tests.structlog")
|
|
logger.info("login", password="secret", token="abc")
|
|
|
|
_flush_root_handlers()
|
|
|
|
log_path = configured_logging / "app.log"
|
|
entry = _read_last_log_entry(log_path)
|
|
|
|
assert entry["password"] == "[REDACTED]"
|
|
assert entry["token"] == "[REDACTED]"
|