logging: 全链路日志与完整异常堆栈
- 新增 logutil:DEBUG 全量滚动日志文件 + 控制台 INFO 简洁输出 + GUI 队列处理器 - providers:每次尝试失败记录 HTTP 状态/响应体,网络异常 logger.exception 留完整堆栈 - pipeline:单任务异常不拖垮整批,缓存命中/阶段耗时/停止过程全程留痕 - CLI/GUI 顶层兜底:未预期异常打印完整 traceback 并指引日志文件;GUI 新增「打开日志」按钮 - 新增 test_logutil(文件落盘/级别过滤/异常堆栈/队列),共 33 个测试全部通过
This commit is contained in:
@@ -0,0 +1,54 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""logutil 模块测试:文件落盘、级别过滤、异常堆栈、队列处理器。"""
|
||||
import logging
|
||||
import queue
|
||||
|
||||
from violation_detector.logutil import LOGGER_NAME, QueueHandler, setup_logging
|
||||
|
||||
|
||||
def get_logger():
|
||||
return logging.getLogger(LOGGER_NAME)
|
||||
|
||||
|
||||
def test_setup_creates_file_and_logs(tmp_path):
|
||||
path = setup_logging(console=False, log_dir=tmp_path)
|
||||
assert path.exists()
|
||||
get_logger().info("你好日志")
|
||||
for h in get_logger().handlers:
|
||||
h.flush()
|
||||
text = path.read_text(encoding="utf-8")
|
||||
assert "你好日志" in text
|
||||
assert "[violation_detector]" in text # 详细格式含模块名
|
||||
|
||||
|
||||
def test_exception_logs_full_traceback(tmp_path):
|
||||
path = setup_logging(console=False, log_dir=tmp_path)
|
||||
try:
|
||||
raise ValueError("爆炸点")
|
||||
except ValueError:
|
||||
get_logger().exception("捕获到异常")
|
||||
for h in get_logger().handlers:
|
||||
h.flush()
|
||||
text = path.read_text(encoding="utf-8")
|
||||
assert "Traceback (most recent call last)" in text
|
||||
assert "ValueError: 爆炸点" in text
|
||||
assert "test_exception_logs_full_traceback" in text # 堆栈含调用位置
|
||||
|
||||
|
||||
def test_debug_goes_to_file_not_queue(tmp_path):
|
||||
"""文件记录 DEBUG 全量;队列处理器(GUI)只收 INFO 以上。"""
|
||||
path = setup_logging(console=False, log_dir=tmp_path)
|
||||
q = queue.Queue()
|
||||
get_logger().addHandler(QueueHandler(q))
|
||||
get_logger().debug("调试细节")
|
||||
get_logger().warning("告警信息")
|
||||
msgs = []
|
||||
while not q.empty():
|
||||
msgs.append(q.get_nowait()[1])
|
||||
# 队列里只有 WARNING,DEBUG 只进文件
|
||||
assert any("告警信息" in m for m in msgs)
|
||||
assert not any("调试细节" in m for m in msgs)
|
||||
for h in get_logger().handlers:
|
||||
if isinstance(h, logging.FileHandler):
|
||||
h.stream.flush()
|
||||
assert "调试细节" in path.read_text(encoding="utf-8")
|
||||
Reference in New Issue
Block a user