78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
from io import StringIO
|
||
|
||
from exporter.config import GlobalConfig
|
||
from exporter.logging_utils import DeviceLoggerAdapter, DeviceFieldFilter, init_logging
|
||
|
||
|
||
def _capture_log_output(logger: logging.Logger) -> tuple[StringIO, logging.Handler]:
|
||
stream = StringIO()
|
||
handler = logging.StreamHandler(stream)
|
||
formatter = logging.Formatter(
|
||
fmt="%(asctime)s %(levelname)s [%(name)s] device=%(device)s %(message)s",
|
||
datefmt="%Y-%m-%dT%H:%M:%S%z",
|
||
)
|
||
handler.setFormatter(formatter)
|
||
logger.addHandler(handler)
|
||
logger.setLevel(logging.INFO)
|
||
return stream, handler
|
||
|
||
|
||
def test_device_logger_adapter_injects_device() -> None:
|
||
logger = logging.getLogger("test_logger_adapter_injects_device")
|
||
stream, handler = _capture_log_output(logger)
|
||
try:
|
||
adapter = DeviceLoggerAdapter(logger, {"device": "dev1"})
|
||
adapter.info("hello world")
|
||
output = stream.getvalue()
|
||
finally:
|
||
logger.removeHandler(handler)
|
||
|
||
assert "device=dev1" in output
|
||
# 简单校验时间格式中包含 "T"（ISO8601 风格）
|
||
assert "T" in output.split()[0]
|
||
|
||
|
||
def test_device_logger_fallback_device_dash() -> None:
|
||
logger = logging.getLogger("test_logger_adapter_fallback")
|
||
stream, handler = _capture_log_output(logger)
|
||
try:
|
||
# 不在 extra 中显式提供 device，Adapter 应该使用 "-" 作为默认值
|
||
adapter = DeviceLoggerAdapter(logger, {})
|
||
adapter.info("no device context")
|
||
output = stream.getvalue()
|
||
finally:
|
||
logger.removeHandler(handler)
|
||
|
||
assert "device=-" in output
|
||
|
||
|
||
def test_init_logging_configures_root_logger_handlers() -> None:
|
||
"""验证 init_logging 根据 GlobalConfig 初始化根 logger。"""
|
||
gc = GlobalConfig(
|
||
log_level="INFO",
|
||
log_to_stdout=True,
|
||
log_file="",
|
||
)
|
||
# 调用被测函数
|
||
init_logging(gc)
|
||
|
||
root = logging.getLogger()
|
||
# 至少应该有一个 handler（stdout）
|
||
assert root.handlers, "root logger should have at least one handler"
|
||
|
||
for handler in root.handlers:
|
||
fmt = handler.formatter._fmt # type: ignore[attr-defined]
|
||
datefmt = handler.formatter.datefmt # type: ignore[attr-defined]
|
||
assert "device=%(device)s" in fmt
|
||
assert "%Y-%m-%dT%H:%M:%S%z" == datefmt
|
||
|
||
# 至少有一个 handler 安装了 DeviceFieldFilter，保证没有 device 字段的记录也能格式化
|
||
assert any(
|
||
isinstance(flt, DeviceFieldFilter)
|
||
for handler in root.handlers
|
||
for flt in handler.filters
|
||
)
|