91 lines
2.6 KiB
Python
91 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
|
|
from quartermaster.logging_config import LOG_CONFIG
|
|
|
|
|
|
def _build_formatter():
|
|
"""Instantiate the JSON formatter from LOG_CONFIG for direct testing."""
|
|
from pydoc import locate
|
|
|
|
cfg = dict(LOG_CONFIG["formatters"]["json"])
|
|
factory_path = cfg.pop("()")
|
|
factory = locate(factory_path)
|
|
assert factory is not None, f"formatter factory not importable: {factory_path}"
|
|
return factory(**cfg)
|
|
|
|
|
|
def test_log_config_emits_json_with_required_fields():
|
|
formatter = _build_formatter()
|
|
stream = io.StringIO()
|
|
handler = logging.StreamHandler(stream)
|
|
handler.setFormatter(formatter)
|
|
|
|
logger = logging.getLogger("tests.logging_smoke")
|
|
logger.handlers = [handler]
|
|
logger.propagate = False
|
|
logger.setLevel(logging.INFO)
|
|
|
|
logger.info("smoke message", extra={"event": "smoke"})
|
|
|
|
line = stream.getvalue().strip()
|
|
assert line, "formatter produced no output"
|
|
payload = json.loads(line)
|
|
|
|
assert payload["event"] == "smoke"
|
|
assert payload["level"] == "INFO"
|
|
assert payload["logger"] == "tests.logging_smoke"
|
|
assert payload["message"] == "smoke message"
|
|
assert "timestamp" in payload
|
|
|
|
|
|
def test_access_log_filter_enriches_record():
|
|
from quartermaster.logging_config import AccessLogFilter
|
|
|
|
record = logging.LogRecord(
|
|
name="uvicorn.access",
|
|
level=logging.INFO,
|
|
pathname=__file__,
|
|
lineno=0,
|
|
msg='%s - "%s %s HTTP/%s" %d',
|
|
args=("127.0.0.1:1234", "GET", "/healthz", "1.1", 200),
|
|
exc_info=None,
|
|
)
|
|
|
|
assert AccessLogFilter().filter(record) is True
|
|
assert record.event == "http_request"
|
|
assert record.method == "GET"
|
|
assert record.path == "/healthz"
|
|
assert record.status == 200
|
|
assert record.client_ip == "127.0.0.1:1234"
|
|
|
|
|
|
def test_access_log_filter_passes_through_non_access_records():
|
|
"""Non-access records (args tuple shape mismatch) are kept, unmodified."""
|
|
from quartermaster.logging_config import AccessLogFilter
|
|
|
|
record = logging.LogRecord(
|
|
name="quartermaster",
|
|
level=logging.INFO,
|
|
pathname=__file__,
|
|
lineno=0,
|
|
msg="regular app log",
|
|
args=None,
|
|
exc_info=None,
|
|
)
|
|
|
|
assert AccessLogFilter().filter(record) is True
|
|
assert not hasattr(record, "event") or record.event is None
|
|
assert not hasattr(record, "method") or record.method is None
|
|
|
|
|
|
def test_log_config_loads_via_dictconfig():
|
|
"""Verify the whole LOG_CONFIG passes logging.config.dictConfig without error."""
|
|
import logging.config
|
|
|
|
from quartermaster.logging_config import LOG_CONFIG
|
|
|
|
logging.config.dictConfig(LOG_CONFIG)
|