efi-kernel/试跑引导器.py

#!/usr/bin/env python3
"""试跑引导器: 把 UEFI.boot.py 的命令真跑一遍, 原样贴输出 + 逐条判定.

用法:
    python3 试跑引导器.py             # 只读验收 (不碰任何文件)
    python3 试跑引导器.py --假内核    # 额外演示"引导器管内核进程" (临时换入假内核, 跑完自动还原)

为什么要有它:
    引导器能不能立得住, 得真跑一遍才算数.老板亲自跑: 每步都打**命令原文 + 原样输出**,
    不加工不摘要; 末了给一张汇总表, 退出码 0 = 没有 FAIL.

判定口径 (三种, 别混):
    PASS  退出码和输出都符合预期
    预期  按设计就该这样 (例: 内核入口还是 0 字节, 空内核进来秒退是对的, 不算错)
    FAIL  真出问题了 (退出码不对 / 输出缺关键内容 / 环境状态文件对不上)

只读原则:
    默认模式不改任何文件 (体检 --check 只读; 环境 --json 会写环境状态快照, 那是它的正常产出).
    --假内核 会临时改 内核/内核.py, 但 finally 里一定按**原样字节**还原, 并补一次 内核 停止
    兜底, 绝不留孤儿进程.
"""

from __future__ import annotations

import json
import subprocess
import sys
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, cast

项目根 = Path(__file__).resolve().parent
# 和 UEFI.boot.py 一个套路: 先把 内核/ 塞进模块搜索路径, 再 import 兄弟模块
# (共用 内核/文本.py 的表格实现 -- 中文是双宽, len() 排出来是歪的, 只此一份实现别复制)
sys.path.insert(0, str(项目根 / "内核"))

import 文本  # noqa: E402  -- 必须等 sys.path 改完才能 import

引导器 = 项目根 / "UEFI.boot.py"
内核入口 = 项目根 / "内核" / "内核.py"
状态文件 = 项目根 / "环境状态.efi.json"

# 假内核: 只为验证"引导器能不能拉起 / 看住 / 停掉一个常驻进程", 不含任何真逻辑.
# 铁律: 收到 SIGTERM 就正常退出 -- 别装死, 否则引导器只能 SIGKILL (那就不算优雅停止).
假内核源码 = '''#!/usr/bin/env python3
# 假内核 (临时占位): 主人是 试跑引导器.py, 跑完就被还原, 不会留在这里.
import os
import signal
import sys
import time


def 收工(号: int, 帧: object) -> None:
    # 引导器停止时给整个进程组发 SIGTERM, 这里干净退出
    print("假内核: 收到信号", 号, "收工", flush=True)
    sys.exit(0)


signal.signal(signal.SIGTERM, 收工)
print("假内核: 起来了 pid=", os.getpid(), flush=True)
while True:
    print("假内核: 心跳", flush=True)
    time.sleep(2)
'''


def _文本(值: object) -> str:
    """把 subprocess 的输出片段统一成 str (text=True 时本来就是 str, 这里兜住 bytes / None)."""
    if isinstance(值, str):
        return 值
    if isinstance(值, (bytes, bytearray)):
        return bytes(值).decode("utf-8", "replace")
    return ""


@dataclass
class 结果:
    """一条步骤的记录, 最后用来渲染汇总表."""

    名称: str
    命令: str
    退出码: int
    判定: str
    备注: str = ""


记录: list[结果] = []


def 跑(参数: list[str], 超时: int = 180) -> tuple[int, str]:
    """跑一条引导器命令, 回收输出.

    参数:
        参数: 追加在 UEFI.boot.py 后面的参数 (列表透传, 不过 shell, 不怕空格/中文).
        超时: 秒.守护启动要等"秒退判定"落地 (约 1 秒), 给足 180 不会误杀.
    返回:
        (退出码, 输出); 输出 = stdout 原样 + 若 stderr 有内容再加一段带 [stderr] 标记的尾巴.
        不截断, 不加工 -- 老板要看的就是原生输出.
        超时返回 -1 (常驻内核前台挂着不返回, 这不是错: 见第 8 项的判定).
    """
    try:
        完成 = subprocess.run(
            [sys.executable, str(引导器), *参数],
            cwd=str(项目根),
            capture_output=True,
            text=True,
            timeout=超时,
        )
    except subprocess.TimeoutExpired as 超时了:
        # 超时不是失败: 裸跑那种"前台起常驻内核"的命令本来就该一直挂着.
        # 注意: subprocess.run 超时只 kill 直接子进程, 孙进程 (引导器前台起的常驻内核)
        # 会活下来 -- 调用方还得自己收 (第 8 项就是干这个的).
        片段 = _文本(超时了.stdout)
        尾巴 = _文本(超时了.stderr)
        输出 = 片段 + ("[stderr]\n" + 尾巴 if 尾巴.strip() else "")
        return -1, 输出 + f"\n[超时 {超时}s] 命令还挂着 (常驻内核就是这个样子, 已按超时收掉)"
    输出 = 完成.stdout
    if 完成.stderr.strip():
        输出 += "[stderr]\n" + 完成.stderr
    return 完成.returncode, 输出


def 记(名称: str, 参数: list[str], 退出码: int, 输出: str, 判定: str, 备注: str = "") -> None:
    """打一块"命令原文 + 原样输出 + 判定", 同时记进汇总表.

    参数:
        名称: 这一步在汇总表里的名字 (对着汇总表看能找回来).
        参数: 命令参数 (空列表 = 这一步不是引导器命令, 例如"看快照文件").
        退出码: 引导器返回的退出码; 非命令步骤传 0 或 -1.
        输出: 原样输出 (命令步骤就是它吐的原文).
        判定: PASS / 预期 / FAIL.
        备注: 一句话说明判据或失败原因; 走旁路打印, 方便老板直接看出"凭什么这么判".
    """
    print(f"\n[第 {len(记录) + 1} 项] {名称}")
    if 参数:
        print(f"$ {Path(sys.executable).name} UEFI.boot.py {' '.join(参数)}")
    if 输出.strip():
        print(输出.rstrip())
    print(f"  退出码 {退出码} | 判定 {判定}" + (f" | {备注}" if 备注 else ""))
    记录.append(结果(名称=名称, 命令=" ".join(参数), 退出码=退出码, 判定=判定, 备注=备注))


def 残留内核进程() -> list[int]:
    """扫 /proc 找还活着的内核进程.

    为什么不用 ps:
        软路由上踩过 -- busybox 的 ps 会截断长命令行, 而且会把**我自己的排查命令**也算进去
        (假计数).直读 /proc/<pid>/cmdline 最稳, 还能排除自己.
    """
    出: list[int] = []
    目标 = str(内核入口)
    for 目录 in Path("/proc").iterdir():
        if not 目录.name.isdigit():
            continue
        try:
            命令行 = (目录 / "cmdline").read_bytes().decode("utf-8", "replace")
        except OSError:
            continue
        if 目标 in 命令行:
            出.append(int(目录.name))
    return 出


def 演示假内核() -> None:
    """临时把 内核/内核.py 换成假内核, 看引导器能不能真的管住它.

    四步一条链: 启动 --守护 -> 状态(运行中) -> 日志(看得到它自己打的话) -> 停止(收干净).
    兜底: finally 里先补一次 内核 停止 (万一中途崩了别留孤儿), 再按原样字节还原真内核文件.
    """
    原字节 = 内核入口.read_bytes()
    print(f"\n[假内核] 备份 {内核入口.name} ({len(原字节)} 字节) -> 换入临时假内核")
    try:
        内核入口.write_text(假内核源码, encoding="utf-8")

        码, 出 = 跑(["内核", "启动", "--守护"])
        状态码, 状态出 = 跑(["内核", "状态"])
        好 = 码 == 0 and 状态码 == 0 and "运行中" in 状态出
        记(
            "假内核: 启动 --守护",
            ["内核", "启动", "--守护"],
            码,
            出,
            "PASS" if 好 else "FAIL",
            "守护起来了, 且 内核 状态 显示 运行中" if 好 else "期望守护启动成功 + 状态显示 运行中",
        )
        记(
            "假内核: 内核 状态",
            ["内核", "状态"],
            状态码,
            状态出,
            "PASS" if 状态码 == 0 else "FAIL",
            "能看到 pid / 启动时刻 / 今日运行次数" if 状态码 == 0 else "",
        )

        码, 出 = 跑(["内核", "日志", "--输出", "-n", "20"])
        看到 = "假内核" in 出
        记(
            "假内核: 内核 日志 --输出 (它的话在 out.log 里)",
            ["内核", "日志", "--输出", "-n", "20"],
            码,
            出,
            "PASS" if 码 == 0 and 看到 else "FAIL",
            "日志里看得到假内核自己打的话" if 看到 else "期望日志里出现 假内核 字样",
        )

        码, 出 = 跑(["内核", "停止"])
        记("假内核: 内核 停止", ["内核", "停止"], 码, 出, "PASS" if 码 == 0 else "FAIL",
           "SIGTERM 给进程组, 连子孙一起收" if 码 == 0 else "")

        剩 = 残留内核进程()
        记(
            "假内核: 残留检查",
            [],
            0 if not 剩 else -1,
            f"残留进程: {剩}" if 剩 else "进程组收干净了 (无残留)",
            "PASS" if not 剩 else "FAIL",
        )
    finally:
        跑(["内核", "停止"])  # 兜底: 上面哪步炸了也别留孤儿
        内核入口.write_bytes(原字节)  # 还原真内核文件 (哪怕它是 0 字节)
        print(f"[假内核] 已还原 {内核入口.name} ({len(原字节)} 字节)")


def main(argv: list[str]) -> int:
    """把引导器的每条命令跑一遍, 返回 0 = 没有 FAIL."""
    print("=" * 78)
    print(f"试跑引导器 -- 端到端验收   {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"  项目根: {项目根}")
    print(f"  解释器: {sys.executable}")
    print(f"  模式  : {'含假内核进程演示' if '--假内核' in argv else '只读验收 (加 --假内核 演示管进程)'}")
    print("=" * 78)

    # 1) 用法: 纯 stdlib 就能打出命令一览, 任何环境都不该崩
    码, 出 = 跑(["-h"])
    判 = "PASS" if 码 == 0 and "用法" in 出 else "FAIL"
    记("用法 (-h)", ["-h"], 码, 出, 判, "" if 判 == "PASS" else "期望退出码 0 且输出含 用法")

    # 2) 体检: 全绿才回 0 (阻塞项会回 1), 所以退出码本身就是判据
    码, 出 = 跑(["--check"])
    判 = "PASS" if 码 == 0 else "FAIL"
    记("体检 (--check)", ["--check"], 码, 出, 判,
       "6 项全过, 退出码 0" if 判 == "PASS" else "有阻塞项: 按上面的修复命令修完再跑")

    # 3) 机器读的体检: --json 必须是纯 JSON (混了人读的表格就没法直接喂给程序)
    码, 出 = 跑(["环境", "--json"])
    快照: Any = None
    备注 = ""
    try:
        快照 = json.loads(出)
    except json.JSONDecodeError as 错:
        备注 = f"stdout 不是纯 JSON: {错}"
    if 快照 is None:
        判 = "FAIL"
    else:
        体检段 = 快照.get("checks")
        # isinstance 窄化出来的只是 list[Unknown], 取 len 会被判"类型部分未知"; cast 明确一次
        项数 = len(cast(list[Any], 体检段)) if isinstance(体检段, list) else 0
        判 = "PASS" if 码 == 0 and 快照.get("blocking_ok") is True and 项数 == 6 else "FAIL"
        备注 = f"纯 JSON 可解析, checks {项数} 项, blocking_ok={快照.get('blocking_ok')}"
    记("体检 (机器读 环境 --json)", ["环境", "--json"], 码, 出, 判, 备注)

    # 4) 包台账: 显式命令才有台账看, 不自动装不自动升
    码, 出 = 跑(["包"])
    判 = "PASS" if 码 == 0 and "psycopg2-binary" in 出 else "FAIL"
    记("包 (台账)", ["包"], 码, 出, 判,
       "台账里有 psycopg2-binary 那一行" if 判 == "PASS" else "期望看到 psycopg2-binary")

    # 5) 快照文件: 体检结论要落盘留痕, 而且得是"刚刚"写的
    if not 状态文件.exists():
        记("环境状态快照", [], -1, f"{状态文件.name} 不存在", "FAIL", "跑一次 环境 --json 应该会写它")
    else:
        数据: Any = json.loads(状态文件.read_text(encoding="utf-8"))
        try:
            时刻 = datetime.fromisoformat(str(数据.get("ts", "")))
            新鲜 = datetime.now(时刻.tzinfo) - 时刻 < timedelta(minutes=5)
        except ValueError:
            新鲜 = False
        项数 = len(数据.get("checks") or [])
        好 = 数据.get("efi") == 1 and 新鲜 and 项数 == 6
        记(
            "环境状态快照",
            [],
            0 if 好 else -1,
            f"{状态文件.name}: efi={数据.get('efi')} ts={数据.get('ts')} "
            f"checks={项数} 新鲜={新鲜} blocking_ok={数据.get('blocking_ok')}",
            "PASS" if 好 else "FAIL",
            "刚落盘 (5 分钟内) 且 6 项齐全" if 好 else "快照对不上: 看 ts / checks 数",
        )

    # 6) 内核状态: 还没跑内核时, 它应该老实说"已停止", 而不是编一个假状态
    码, 出 = 跑(["内核", "状态"])
    记("内核 状态 (没跑时)", ["内核", "状态"], 码, 出, "PASS" if 码 == 0 else "FAIL",
       "状态 + 入口 + 今日运行次数 + 台账来源" if 码 == 0 else "")

    # 7) 透传: 引导器不认识的话一律交给内核, 退出码照内核的回传
    码, 出 = 跑(["列表"])
    记("透传 (列表)", ["列表"], 码, 出, "PASS" if 码 == 0 else "FAIL",
       "退出码按内核的退出码回传" if 码 == 0 else "")

    # 8) 裸跑 (无参数): 体检 -> 移交内核, 内核的默认动作 = 常驻调度 (甲).
    #    三种都算过: 1 内核不在跑 -> 前台常驻 (会一直挂着, 用短超时收掉);
    #                2 已经有内核 -> 被独一份调度锁挡下 (该挡就得挡);
    #                3 内核自己退了 (退出码 0).
    码, 出 = 跑([], 超时=8)
    if 码 == -1:
        判, 备注 = "PASS", "裸跑 = 前台起常驻内核 (设计如此, 会一直挂着); 超时收掉前台壳, 常驻内核留给第 11 项收"
    elif 码 == 0:
        判, 备注 = "PASS", "裸跑: 体检 -> 移交内核 -> 内核自己退出 (退出码 0)"
    elif "已经有一个内核在跑" in 出:
        判, 备注 = "PASS", "被独一份调度锁挡下 (已经有一个内核在跑) -- 挡得对"
    else:
        判, 备注 = "FAIL", "既没正常退出, 也不是'已经有一个内核在跑'"
    记("裸跑 (无参数)", [], 码, 出, 判, 备注)

    # 9) 内核启停: 内核入口空着时, 空内核秒退是**对的**, 引导器该如实报"不像常驻进程"
    空内核 = 内核入口.stat().st_size == 0
    码, 出 = 跑(["内核", "启动", "--守护"])
    if 空内核:
        判, 备注 = "预期", "内核入口 0 字节: 空内核进来就退是对的 (退出码 1 = 我没把它当常驻)"
    elif 码 == 0:
        判, 备注 = "PASS", "守护起来了"
    else:
        判, 备注 = "FAIL", "内核有内容但守护没起来: 看上面的日志尾巴"
    记("内核 启动 --守护", ["内核", "启动", "--守护"], 码, 出, 判, 备注)

    码, 出 = 跑(["内核", "日志", "-n", "10"])
    记("内核 日志 -n 10", ["内核", "日志", "-n", "10"], 码, 出, "PASS" if 码 == 0 else "FAIL")

    码, 出 = 跑(["内核", "停止"])
    记("内核 停止 (幂等)", ["内核", "停止"], 码, 出, "PASS" if 码 == 0 else "FAIL",
       "没在跑也回 0 (幂等), 顺手给没写 finished_at 的运行记录收尸" if 码 == 0 else "")

    # 10) 可选: 假内核演示 (只在 --假内核 且内核入口还空着时做, 绝不覆盖老板写的代码)
    if "--假内核" in argv:
        if 空内核:
            演示假内核()
        else:
            记("假内核演示", [], -1, f"跳过: {内核入口.name} 已有内容, 不覆盖", "预期")

    # 收尾: 全跑完再看一眼有没有漏网的进程
    剩 = 残留内核进程()

    print("\n" + "=" * 78)
    print("汇总")
    print("=" * 78)
    行表 = [[str(i), r.名称, str(r.退出码), r.判定, r.备注] for i, r in enumerate(记录, 1)]
    文本.打印(文本.表格(["#", "步骤", "退出码", "判定", "备注"], 行表,
                    [文本.右, 文本.左, 文本.右, 文本.左, 文本.左]))

    过 = sum(1 for r in 记录 if r.判定 == "PASS")
    预期数 = sum(1 for r in 记录 if r.判定 == "预期")
    败 = sum(1 for r in 记录 if r.判定 == "FAIL")
    print(f"\n  合计: PASS {过} | 预期 {预期数} | FAIL {败}")
    print(f"  残留: {'无' if not 剩 else f'还有 {剩} 没收 -- 收: python3 UEFI.boot.py 内核 停止'}")
    print(f"  结论: {'全过' if 败 == 0 else '有 FAIL, 往上翻对应那块的原样输出'}")
    if 空内核:
        print("  提醒: 内核入口还是 0 字节, '常驻'那两项要等 内核/内核.py 写出来才作数")
    return 0 if 败 == 0 else 1


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))