WorkBuddy
f0c1cc1491
- LOG_FILE 配置: 空(默认)保持 stdout + Admin 内存日志页(重启清零); 填路径后额外写 RotatingFileHandler(单文件 10MB × 5 份,utf-8) - setup_logging 幂等挂载(按 abspath 判重,相对/绝对同文件算一个); 目录自动创建;文件基础设施失败只 warning 绝不拖垮启动 - docker-compose: api 挂 applogs 卷,默认 LOG_FILE=/app/logs/app.log, 容器重建日志不丢;.env.example/.gitignore/docs 同步 - tests/test_log_persistence.py: 落盘/幂等/空值禁用/自动建目录/失败降级 - test_p0_standings_cutoff: 匿名约束 name=None 使 any() 子串匹配 TypeError (集合顺序不定 → flaky),先判真值再匹配(存量缺陷顺手加固)
128 lines
4.4 KiB
Python
128 lines
4.4 KiB
Python
"""日志持久化(LOG_FILE)测试:setup_logging 启用滚动文件后日志必须落盘。
|
|
|
|
背景: 此前应用日志只进 stdout(docker json-file 收集,不可控) + Admin 内存
|
|
日志页(环形缓冲 2000 条,进程重启清零),没有任何应用层持久化 —— 排查
|
|
「昨晚采集为什么失败」这类问题时无据可查。本测试守护:
|
|
1. 传 log_file → root logger 挂 RotatingFileHandler,日志写入文件
|
|
2. 幂等:重复调用不重复挂 handler
|
|
3. log_file 为空 → 不挂文件 handler(保持旧行为)
|
|
4. 目录不存在 → 自动创建
|
|
5. 文件打开失败(如路径是已存在的目录) → 只降级不炸,stdout/内存日志仍在
|
|
|
|
范式: 直接操作 root logger + tmp_path;fixture 保存/恢复 root 状态,
|
|
并关闭新增 handler 的文件句柄(Windows 上句柄不关会锁住 tmp 目录)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
import pytest
|
|
|
|
from src.core.log_buffer import MemoryLogHandler, setup_logging
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _restore_root_logger():
|
|
"""保存/恢复 root logger;关闭测试期间新挂 handler 的句柄。"""
|
|
root = logging.getLogger()
|
|
saved_handlers = list(root.handlers)
|
|
saved_level = root.level
|
|
yield
|
|
for h in root.handlers:
|
|
if h not in saved_handlers and hasattr(h, "close"):
|
|
try:
|
|
h.close()
|
|
except Exception:
|
|
pass
|
|
root.handlers[:] = saved_handlers
|
|
root.setLevel(saved_level)
|
|
|
|
|
|
def _file_handlers():
|
|
return [h for h in logging.getLogger().handlers if isinstance(h, RotatingFileHandler)]
|
|
|
|
|
|
def _memory_handlers():
|
|
return [h for h in logging.getLogger().handlers if isinstance(h, MemoryLogHandler)]
|
|
|
|
|
|
class TestFileHandlerAttached:
|
|
def test_attaches_rotating_file_handler(self, tmp_path):
|
|
log_file = tmp_path / "app.log"
|
|
|
|
setup_logging("INFO", str(log_file))
|
|
|
|
fhs = _file_handlers()
|
|
assert len(fhs) == 1
|
|
assert fhs[0].baseFilename == str(log_file)
|
|
# 内存日志页照常工作,两者并存
|
|
assert len(_memory_handlers()) == 1
|
|
# 滚动参数与文档口径一致: 单文件 10MB × 5 份
|
|
assert fhs[0].maxBytes == 10 * 1024 * 1024
|
|
assert fhs[0].backupCount == 5
|
|
|
|
def test_log_written_to_file(self, tmp_path):
|
|
log_file = tmp_path / "app.log"
|
|
setup_logging("INFO", str(log_file))
|
|
|
|
logging.getLogger("persist-test").info("hello-persist-12345")
|
|
|
|
content = log_file.read_text(encoding="utf-8")
|
|
assert "hello-persist-12345" in content
|
|
assert "persist-test" in content # logger 名可追溯
|
|
assert "INFO" in content # 级别在行首可过滤
|
|
|
|
|
|
class TestIdempotent:
|
|
def test_repeated_call_does_not_duplicate_handlers(self, tmp_path):
|
|
log_file = tmp_path / "app.log"
|
|
|
|
setup_logging("INFO", str(log_file))
|
|
setup_logging("INFO", str(log_file))
|
|
setup_logging("INFO", str(log_file))
|
|
|
|
assert len(_file_handlers()) == 1
|
|
assert len(_memory_handlers()) == 1
|
|
|
|
def test_same_file_via_relative_and_absolute_path_counts_as_one(self, tmp_path, monkeypatch):
|
|
"""相对/绝对路径指向同一文件时不得重复挂(幂等按 abspath 判重)。"""
|
|
monkeypatch.chdir(tmp_path)
|
|
|
|
setup_logging("INFO", "app.log")
|
|
setup_logging("INFO", str(tmp_path / "app.log"))
|
|
|
|
assert len(_file_handlers()) == 1
|
|
|
|
|
|
class TestDisabled:
|
|
def test_empty_log_file_keeps_old_behavior(self):
|
|
setup_logging("INFO", "")
|
|
|
|
assert _file_handlers() == []
|
|
assert len(_memory_handlers()) == 1
|
|
|
|
|
|
class TestAutoMkdir:
|
|
def test_creates_missing_directories(self, tmp_path):
|
|
log_file = tmp_path / "deep" / "nested" / "app.log"
|
|
|
|
setup_logging("INFO", str(log_file))
|
|
|
|
assert len(_file_handlers()) == 1
|
|
assert log_file.parent.is_dir()
|
|
|
|
|
|
class TestGracefulDegradation:
|
|
def test_unopenable_path_degrades_without_raising(self, tmp_path):
|
|
"""路径是已存在的目录 → 打开必然失败;只降级,不炸启动。"""
|
|
dir_as_file = tmp_path / "occupied"
|
|
dir_as_file.mkdir()
|
|
|
|
# 不应抛异常(文件日志是可观测性基础设施,失败只 warning)
|
|
setup_logging("INFO", str(dir_as_file))
|
|
|
|
assert _file_handlers() == []
|
|
# 降级后 stdout/内存日志路径仍在
|
|
assert len(_memory_handlers()) == 1
|