efi-kernel/内核/自测日志.py

#!/usr/bin/env python3
"""日志库真机自测: 真写文件, 真轮转, 真起子进程跟日志 -- 不 mock.

用法: .venv/bin/python 内核/自测日志.py

覆盖: 级别门槛 / 行解析与猜级别 / 写(门槛过滤) / 尾读(级别+关键词过滤) / 轮转(真造超限文件) /
      实时跟(真起一个子进程跟在后面, 追加两条看它吐哪条) / 选项解析 / 启动分隔头(真起进程).

隔离: 所有测试文件都在 tempfile.mkdtemp() 的临时目录里, 跑完整个删掉 --
      生产日志 (内核/logs/*.log, 驱动/*/logs/*.log) 一个字节都不碰 (开头结尾各拍一次快照核对).
"""

from __future__ import annotations

import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path

项目根 = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(项目根 / "内核"))

import 日志  # noqa: E402
import 进程  # noqa: E402

工作 = Path(tempfile.mkdtemp(prefix="_自测日志_"))
生产日志目录 = 项目根 / "内核" / "logs"
失败 = 0


def 核对(名称: str, 实际: object, 期望: object) -> None:
    """一项核对 (照 自测进程.py 的口径: 打实际值, 不对才追加期望值)."""
    global 失败
    ok = 实际 == 期望
    if not ok:
        失败 += 1
    print(f"[{'OK  ' if ok else 'FAIL'}] {名称}: {实际!r}" + ("" if ok else f"  (期望 {期望!r})"))


def 快照(目录: Path) -> dict[str, int]:
    """目录里 .log 文件的 名字->大小 (核对"自测没碰生产日志")."""
    if not 目录.exists():
        return {}
    return {项.name: 项.stat().st_size for 项 in sorted(目录.glob("*.log"))}


def 造满(路径: Path, 标记: str, 行数: int = 400) -> int:
    """造一个"全是标记行"的大文件 (轮转测试用: 之后靠标记认得出这份是哪一轮的)."""
    路径.write_text("".join(f"{标记} 第 {序} 行 填充填充填充填充\n" for 序 in range(行数)), encoding="utf-8")
    return 路径.stat().st_size


生产快照 = 快照(生产日志目录)
print(f"工作目录: {工作}")
print(f"生产日志快照: {生产快照}")

print("== 1) 级别与门槛 (纯逻辑) ==")
核对("规范化 info -> INFO", 日志.规范化级别("info"), "INFO")
核对("规范化 认不出来 -> INFO", 日志.规范化级别("瞎写的"), "INFO")
核对("规范化 空 -> INFO", 日志.规范化级别(None), "INFO")
核对("级别表顺序", 日志.级别表, ("DEBUG", "INFO", "WARN", "ERROR"))
核对("WARN 比 INFO 严重", 日志.级别序号("WARN") > 日志.级别序号("INFO"), True)
核对("ERROR 过 WARN 门槛", 日志.达标("ERROR", "WARN"), True)
核对("INFO 过不了 WARN 门槛", 日志.达标("INFO", "WARN"), False)
核对("空门槛全过", 日志.达标("DEBUG", ""), True)

print("== 2) 行解析与猜级别 ==")
样本 = 日志.格式化("2026-09-16T20:00:00+08:00", "WARN", "内核", "测试 一条")
核对("格式化", 样本, "2026-09-16T20:00:00+08:00 WARN  [内核] 测试 一条")
字段 = 日志.解析行(样本)
核对("解析 级别", (字段 or {}).get("级别"), "WARN")
核对("解析 来源", (字段 or {}).get("来源"), "内核")
核对("解析 消息", (字段 or {}).get("消息"), "测试 一条")
核对("解析 时刻", (字段 or {}).get("时刻"), "2026-09-16T20:00:00+08:00")
核对("裸行解析给 None", 日志.解析行("假内核: 心跳"), None)
核对("猜级别 裸行 Traceback", 日志.猜级别("Traceback (most recent call last):"), "ERROR")
核对("猜级别 裸行 [WARN]", 日志.猜级别("[WARN] 快没电了"), "WARN")
核对("猜级别 裸行 普通输出", 日志.猜级别("假内核: 心跳"), "INFO")
核对("猜级别 结构化行用真字段", 日志.猜级别(样本), "WARN")
核对("命中 关键词在", 日志.命中(样本, "", "测试"), True)
核对("命中 级别不够", 日志.命中(样本, "ERROR", ""), False)
核对("命中 关键词不在", 日志.命中(样本, "", "没有这段"), False)

print("== 3) 记(): 真写文件 + 门槛过滤 ==")
一个 = 工作 / "一个.log"
核对("写 INFO 返回 True", 日志.记(一个, "INFO", "内核", "第一条", 控制台=False), True)
核对("DEBUG 被门槛挡 (返回 False)", 日志.记(一个, "DEBUG", "内核", "被挡掉的", 门槛="INFO"), False)
核对("ERROR 过门槛", 日志.记(一个, "ERROR", "内核", "真错误", 门槛="INFO"), True)
核对("写到子目录也能建", 日志.记(工作 / "子" / "目录.log", "INFO", "内核", "x"), True)
行表 = 日志.读全部(一个)
核对("文件里 2 行 (DEBUG 没进去)", len(行表), 2)
核对("第一行内容", (日志.解析行(行表[0]) or {}).get("消息"), "第一条")
核对("第二行级别", (日志.解析行(行表[1]) or {}).get("来源"), "内核")
核对("门槛空 = 全记", 日志.记(一个, "DEBUG", "内核", "门槛空", 门槛=""), True)
核对("大小文本 5MB", 日志.大小文本(5 * 1024 * 1024), "5.0MB")

print("== 4) 尾读: 行数 / 级别 / 关键词 ==")
多条 = 工作 / "多条.log"
for 序 in range(12):
    级 = "ERROR" if 序 in (0, 5) else "INFO"
    日志.记(多条, 级, "内核", f"第{序}条 标记{'甲' if 序 % 2 else '乙'}", 门槛="")
核对("尾 5 行", len(日志.尾(多条, 5)), 5)
核对("尾 5 行是最后 5 条", (日志.解析行(日志.尾(多条, 5)[0]) or {}).get("消息"), "第7条 标记甲")
核对("级别 ERROR 只 2 条", len(日志.尾(多条, 100, 级别="ERROR")), 2)
核对("关键词 甲 6 条", len(日志.尾(多条, 100, 关键词="甲")), 6)
核对("级别+关键词 同时筛", len(日志.尾(多条, 100, 级别="ERROR", 关键词="乙")), 1)
核对("尾 0 行给空表", 日志.尾(多条, 0), [])
核对("尾 文件不存在给空表", 日志.尾(工作 / "没有这个.log", 5), [])
核对("行数比文件大 = 全给", len(日志.尾(多条, 999)), 12)

print("== 5) 轮转 (真造超限文件) ==")
大 = 工作 / "大.log"
尺寸 = 造满(大, "第一批")
核对("不到上限不轮转", 日志.轮转(大, 上限字节=尺寸 * 2, 保留份数=2), False)
核对("超上限就轮转", 日志.轮转(大, 上限字节=尺寸 // 2, 保留份数=2), True)
核对("原文件已改名 (.1)", 日志.带序路径(大, 1).exists(), True)
核对("原文件不在了 (被改名搬走)", 大.exists(), False)
核对(".1 里是第一批", "第一批" in 日志.带序路径(大, 1).read_text(encoding="utf-8"), True)
造满(大, "第二批")
核对("第二轮轮转", 日志.轮转(大, 上限字节=尺寸 // 2, 保留份数=2), True)
造满(大, "第三批")
核对("第三轮轮转", 日志.轮转(大, 上限字节=尺寸 // 2, 保留份数=2), True)
核对("名单 老->新 (第三次轮转后原文件也不在了)", [项.name for 项 in 日志.轮转名单(大)], ["大.log.2", "大.log.1"])
核对(".2 是第二批 (第三批把第一批挤掉了)", "第二批" in 日志.带序路径(大, 2).read_text(encoding="utf-8"), True)
核对(".1 是第三批", "第三批" in 日志.带序路径(大, 1).read_text(encoding="utf-8"), True)
核对("保留份数=0 = 不轮转", 日志.轮转(大, 上限字节=1, 保留份数=0), False)
核对("上限=0 = 不轮转", 日志.轮转(大, 上限字节=0, 保留份数=3), False)
核对("尾读含历史份 (最早那批已按保留份数挤掉, 剩 .2 + .1)", len(日志.尾(大, 9999)), 800)
核对("尾读不含历史份 -> 原文件不在给 0", len(日志.尾(大, 9999, 含轮转=False)), 0)
造满(大, "第四批")
核对("尾读不带历史份 = 只看当前文件", len(日志.尾(大, 9999, 含轮转=False)), 400)
核对("尾读带历史份 = 当前 + 两份历史", len(日志.尾(大, 9999)), 1200)

print("== 6) 实时跟 (-f): 真起子进程跟在后面 ==")
跟文件 = 工作 / "跟.log"
跟文件.write_text("开头一条\n", encoding="utf-8")
代码 = (
    "import sys; sys.path.insert(0, "
    + repr(str(项目根 / "内核"))
    + "); from pathlib import Path; import 日志; sys.exit(日志.跟(Path("
    + repr(str(跟文件))
    + "), 级别='ERROR', 初始行=10))"
)
子 = subprocess.Popen(
    [sys.executable, "-c", 代码],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
)
time.sleep(1.2)
日志.记(跟文件, "INFO", "内核", "这条不该跟出来", 门槛="")
日志.记(跟文件, "ERROR", "内核", "这条该跟出来", 门槛="")
time.sleep(1.5)
子.terminate()
try:
    跟出, _ = 子.communicate(timeout=5)
except subprocess.TimeoutExpired:
    子.kill()
    跟出, _ = 子.communicate(timeout=5)
核对("跟: 吐出了 ERROR 那条", "这条该跟出来" in 跟出, True)
核对("跟: 过滤掉了 INFO 那条", "这条不该跟出来" in 跟出, False)
核对("跟: 初始的裸行 (猜 INFO) 也被级别挡住", "开头一条" in 跟出, False)
核对("跟: 子进程已收工", 子.poll() is not None, True)

print("== 7) 选项解析 (内核与引导器共用一份) ==")
选, 问题 = 日志.解析选项(["样板常驻", "-n", "5", "--级别", "warn", "-g", "心跳", "--json"])
核对("名", 选.名, "样板常驻")
核对("行数", 选.行数, 5)
核对("级别自动大写", 选.级别, "WARN")
核对("关键词", 选.关键词, "心跳")
核对("json", 选.json输出, True)
核对("没问题", 问题, "")
核对("默认行数可传", 日志.解析选项([], 77)[0].行数, 77)
核对("--全部", 日志.解析选项(["--全部"])[0].全部, True)
核对("--内核", 日志.解析选项(["--内核"])[0].内核, True)
核对("--引导器", 日志.解析选项(["--引导器"])[0].引导器, True)
核对("--输出", 日志.解析选项(["--输出"])[0].输出, True)
核对("-f", 日志.解析选项(["-f"])[0].跟随, True)
核对("坏级别报错", 日志.解析选项(["--级别", "瞎写"])[1], "--级别 只认 DEBUG/INFO/WARN/ERROR, 收到 '瞎写'")
核对("不认识的参数报错", 日志.解析选项(["--瞎写"])[1], "不认识的参数: --瞎写")
核对("缺值报错", 日志.解析选项(["-n"])[1], "-n 后面要跟一个数字")
核对("行数写错报错", 日志.解析选项(["-n", "五"])[1], "-n 后面要跟一个数字, 收到 '五'")

print("== 8) JSON Lines 输出 ==")
多行 = 日志.转json行([样本, "假内核: 心跳"])
核对("两行 jsonl", len(多行.strip().splitlines()), 2)
核对("结构化行带字段", '"级别": "WARN"' in 多行, True)
核对("裸行给空来源", '"来源": ""' in 多行, True)
核对("裸行级别靠猜", '"级别": "INFO"' in 多行, True)

print("== 9) 启动分隔头 (真起进程, 走 进程.启动) ==")
假驱动 = 工作 / "说一句.py"
假驱动.write_text("print('我起来了')\n", encoding="utf-8")
分隔日志 = 工作 / "分隔.log"
结果 = 进程.启动(
    [sys.executable, str(假驱动)],
    cwd=工作,
    日志=分隔日志,
    env=None,
    入口=假驱动,
    探活秒=0.6,
    分隔="启动 假驱动 说一句.py",
)
文本 = 分隔日志.read_text(encoding="utf-8")
核对("分隔头写进日志了", "==== 启动 假驱动 说一句.py" in 文本, True)
核对("进程自己的输出也在", "我起来了" in 文本, True)
核对("分隔头在进程输出之前", 文本.index("====") < 文本.index("我起来了"), True)
核对("秒退仍如实报 (这驱动就是跑完就退)", 结果.ok, False)
分隔日志2 = 工作 / "不分隔.log"
_ = 进程.启动([sys.executable, "-c", "print('没分隔')"], cwd=工作, 日志=分隔日志2, 入口=工作 / "x.py")
核对("不传分隔就不写分隔头", "====" in 分隔日志2.read_text(encoding="utf-8"), False)

print("== 10) 清场与残留终检 ==")
shutil.rmtree(工作, ignore_errors=True)
核对("临时工作目录已删干净", 工作.exists(), False)
核对("生产日志一个字节都没动", 快照(生产日志目录), 生产快照)
剩: list[int] = []
for 项 in Path("/proc").glob("[0-9]*"):
    try:
        命令行 = (项 / "cmdline").read_bytes().decode("utf-8", "replace")
    except OSError:
        continue  # 进程刚没了 / 读不到, 跳过
    if "日志.跟(" in 命令行 and str(工作) in 命令行:
        剩.append(int(项.name))
核对("没有跟日志的自测残留进程", 剩, [])

print()
print(f"结论: {'全过' if 失败 == 0 else f'{失败} 项没过'}")
sys.exit(0 if 失败 == 0 else 1)