364 lines
16 KiB
Python
364 lines
16 KiB
Python
|
#!/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:]))
|