lou
0827b2399c
引导器 UEFI.boot.py (纯 stdlib, 内核的管家) 体检 6 项 (解释器/venv/包/驱动目录/PG) / venv 重建 / 包台账 / 内核进程启停记账 / 参数原样透传 内核日志 / 引导器日志 / 内核 stdio 三条道的查看入口都归它 内核 内核/ (常驻调度 = 甲) 命令消费 (PG commands + LISTEN/NOTIFY) / 驱动调用仲裁六条 (越权·成环·无人提供·按需拉起·同锁排队·超时收权) 依赖链巡检 10s / 心跳 300s / 独一份调度锁 (pg_try_advisory_lock 0x65666901) 扫描: 9 条校验 + 契约匹配 + 拓扑排序; 状态: 快照原子写 + 断电收尸判定 db.py 是唯一碰 SQL 的文件 (drivers/driver_state/events/scans/commands/calls/kernel_env/kernel_runs) 日志系统 (2026-09-16 完整化, 设计/04-日志系统.md) 三条道一个文件一种内容: 内核.log (结构化行) / 内核.out.log (命令输出 + 崩溃原文) / 引导器.log (引导器动作) 驱动日志每轮启动前插分隔头; 门槛 log_level / 轮转 log_max_mb + log_keep / -n · -f · --级别 · -g · --json · --全部 实现 内核/日志.py (内核与引导器共用一份); 修掉"命令输出混进日志文件" (实证存档 归档/20260916-日志系统重做前/) 驱动样板 (也是写驱动的示范): Json解码 (oneshot) / 样板常驻 (provides 样板:心跳) / 样例消费器 (needs 只发契约名) 代码标点统一为 ASCII (保留界面用的框线 ─│◄▶ 与表格占位符 —); 已用 AST 等价对比证明逻辑零改动 自测 (真机, 不 mock): 进程 (真起进程真收子树) / 内核 58 项 / 配置 / db 182 项 / 日志 86 项 验收: 试跑引导器.py PASS 11 / FAIL 0 / 残留无; uvx pyright 与 uvx basedpyright 均 0 errors 0 warnings
263 lines
14 KiB
Python
263 lines
14 KiB
Python
"""内核的纯逻辑自测: 校验 / 契约匹配 / 拓扑排序 / 状态机 / 拼命令.
|
|
|
|
用法: python3 内核/自测内核.py
|
|
|
|
为什么分成"纯逻辑"这一份:
|
|
端到端那套 (真起进程,真写库) 由 试跑引导器.py 和真机敲命令来验;
|
|
这一份只喂**脏配置**和**假状态行**, 不碰 PG,不写注册表 -- 所以可以反复跑, 不留残渣.
|
|
老板口径: "单元测试不算测试" -- 所以这里除了 状态.复核() 那一组会真起一个 python 进程去
|
|
验判活, 其余都是纯函数; 真流程由真机命令验 (见技能里的 试跑引导器.py + 启动/停止实测).
|
|
|
|
跑完 0 退出; 有 FAIL 退出 1.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
项目根 = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(项目根 / "内核"))
|
|
|
|
import 内核 # noqa: E402
|
|
import 扫描 # noqa: E402
|
|
import 状态 # noqa: E402
|
|
|
|
通过 = 0
|
|
失败 = 0
|
|
|
|
|
|
def 核对(名称: str, 实际: object, 期望: object) -> None:
|
|
"""比一下, 顺手把结果打出来 (老板要看得到逐条的实测值)."""
|
|
global 通过, 失败
|
|
if 实际 == 期望:
|
|
通过 += 1
|
|
print(f" [OK] {名称}: {实际!r}")
|
|
return
|
|
失败 += 1
|
|
print(f" [FAIL] {名称}: 实际 {实际!r} != 期望 {期望!r}")
|
|
|
|
|
|
def 造驱动(根: Path, 目录名: str, 配置: dict[str, Any] | None, 入口名: str = "跑.py", 建入口: bool = True,
|
|
入口可执行: bool = False, 配置原文: str | None = None) -> Path:
|
|
"""在临时驱动根里造一个驱动文件夹 (配置可以给字典 / 原始文本 / 不给)."""
|
|
目录 = 根 / 目录名
|
|
目录.mkdir(parents=True, exist_ok=True)
|
|
if 配置原文 is not None:
|
|
(目录 / 状态.配置名).write_text(配置原文, encoding="utf-8")
|
|
elif 配置 is not None:
|
|
(目录 / 状态.配置名).write_text(json.dumps(配置, ensure_ascii=False), encoding="utf-8")
|
|
if 建入口:
|
|
入口 = 目录 / 入口名
|
|
入口.write_text("# 测试用\nprint('hi')\n", encoding="utf-8")
|
|
入口.chmod(0o755 if 入口可执行 else 0o644)
|
|
return 目录
|
|
|
|
|
|
def 测扫目录(根: Path) -> None:
|
|
"""扫目录: 只认根目录有 配置.efi.json 的一级子目录 (隐藏目录/没配置的都不算)."""
|
|
print("\n== 扫目录 ==")
|
|
造驱动(根, "有配置", {"efi": 1, "name": "有配置", "runtime": "python", "entry": "跑.py"})
|
|
造驱动(根, "没配置", None)
|
|
造驱动(根, ".藏起来的", {"efi": 1, "name": "藏", "runtime": "python", "entry": "跑.py"})
|
|
(根 / "一个文件.txt").write_text("x", encoding="utf-8")
|
|
出 = 扫描.扫目录(根)
|
|
核对("认出来的驱动数", len(出), 1)
|
|
核对("认出来的是谁", [项.name for 项 in 出], ["有配置"])
|
|
|
|
|
|
def 测校验(根: Path) -> None:
|
|
"""9 条校验里的静态部分 (1-6): 每种脏配置都要被拦下, 且原因写在 error 里."""
|
|
print("\n== 校验单个 (脏配置逐条) ==")
|
|
好 = 造驱动(根, "甲", {"efi": 1, "name": "甲", "runtime": "python", "entry": "跑.py"})
|
|
行 = 扫描.校验单个(好)
|
|
核对("好驱动 valid", 行["valid"], True)
|
|
核对("好驱动 entry_hash 非空", bool(行["entry_hash"]), True)
|
|
核对("好驱动 config_hash 非空", bool(行["config_hash"]), True)
|
|
核对("name 缺省用文件夹名", 扫描.校验单个(造驱动(根, "没写名", {"efi": 1, "runtime": "python", "entry": "跑.py"}))["name"], "没写名")
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "坏json", None, 配置原文="{ 这不是 json"))
|
|
核对("坏 json 被拦", 行["valid"], False)
|
|
核对("坏 json 原因", "不是合法 JSON" in str(行["error"]), True)
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "版本不对", {"efi": 9, "name": "版本不对", "runtime": "python", "entry": "跑.py"}))
|
|
核对("协议版本被拦", 行["valid"], False)
|
|
核对("版本原因", "协议版本不支持" in str(行["error"]), True)
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "形态不对", {"efi": 1, "name": "形态不对", "runtime": "shell", "entry": "跑.py"}))
|
|
核对("runtime 被拦", 行["valid"], False)
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "越界", {"efi": 1, "name": "越界", "runtime": "python", "entry": "../../跑.py"}))
|
|
核对("entry 越界被拦", 行["valid"], False)
|
|
核对("越界原因", "越界" in str(行["error"]), True)
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "缺入口", {"efi": 1, "name": "缺入口", "runtime": "python", "entry": "没有这个.py"}, 建入口=False))
|
|
核对("入口不存在被拦", 行["valid"], False)
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "缺x位", {"efi": 1, "name": "缺x位", "runtime": "exec", "entry": "跑.py"}))
|
|
核对("exec 缺 x 位被拦", 行["valid"], False)
|
|
|
|
行 = 扫描.校验单个(造驱动(根, "解释器乱写", {"efi": 1, "name": "解释器乱写", "runtime": "python", "entry": "跑.py", "interpreter": "奇怪的写法"}))
|
|
核对("interpreter 乱写被拦", 行["valid"], False)
|
|
|
|
|
|
def 测重名(根: Path) -> None:
|
|
"""第 7 条: 驱动名全局唯一, 后到的让位 (按目录名排序扫描, 结果稳定)."""
|
|
print("\n== 查重名 ==")
|
|
记录表 = [
|
|
扫描.校验单个(造驱动(根, "重名a", {"efi": 1, "name": "重名", "runtime": "python", "entry": "跑.py"})),
|
|
扫描.校验单个(造驱动(根, "重名b", {"efi": 1, "name": "重名", "runtime": "python", "entry": "跑.py"})),
|
|
]
|
|
扫描.查重名(记录表)
|
|
核对("先到的留", 记录表[0]["valid"], True)
|
|
核对("后到的被拒", 记录表[1]["valid"], False)
|
|
核对("原因写着重复", "驱动名重复" in str(记录表[1]["error"]), True)
|
|
|
|
|
|
def 测契约(根: Path) -> None:
|
|
"""第 8 条: 契约匹配 + 拓扑排序 + 成环拒载 (含"环上驱动的消费者跟着拒载")."""
|
|
print("\n== 契约匹配 / 拓扑 / 成环 ==")
|
|
记录表 = [
|
|
扫描.校验单个(造驱动(根, "供货", {"efi": 1, "name": "供货", "runtime": "python", "entry": "跑.py",
|
|
"provides": ["表格:网页"]})),
|
|
扫描.校验单个(造驱动(根, "消费", {"efi": 1, "name": "消费", "runtime": "python", "entry": "跑.py",
|
|
"needs": ["表格:网页"]})),
|
|
扫描.校验单个(造驱动(根, "孤儿", {"efi": 1, "name": "孤儿", "runtime": "python", "entry": "跑.py",
|
|
"needs": ["没人给的东西"]})),
|
|
扫描.校验单个(造驱动(根, "环甲", {"efi": 1, "name": "环甲", "runtime": "python", "entry": "跑.py",
|
|
"provides": ["环:a"], "needs": ["环:b"]})),
|
|
扫描.校验单个(造驱动(根, "环乙", {"efi": 1, "name": "环乙", "runtime": "python", "entry": "跑.py",
|
|
"provides": ["环:b"], "needs": ["环:a"]})),
|
|
扫描.校验单个(造驱动(根, "环下游", {"efi": 1, "name": "环下游", "runtime": "python", "entry": "跑.py",
|
|
"needs": ["环:a"]})),
|
|
]
|
|
契约, 问题 = 扫描.定契约(记录表)
|
|
拣 = {str(行["name"]): 行 for 行 in 记录表}
|
|
核对("供货 valid", 拣["供货"]["valid"], True)
|
|
核对("消费 valid", 拣["消费"]["valid"], True)
|
|
核对("契约表", 契约.get("表格:网页"), "供货")
|
|
核对("孤儿被拒", 拣["孤儿"]["valid"], False)
|
|
核对("孤儿原因", "契约无人提供" in str(拣["孤儿"]["error"]), True)
|
|
核对("环甲被拒", 拣["环甲"]["valid"], False)
|
|
核对("环乙被拒", 拣["环乙"]["valid"], False)
|
|
核对("环原因写着成环", "成环" in str(拣["环甲"]["error"]), True)
|
|
核对("环下游跟着被拒", 拣["环下游"]["valid"], False)
|
|
核对("问题表 (没有重复提供契约时为空)", 问题, [])
|
|
顺序 = 扫描.排顺序(记录表, 契约)
|
|
核对("供货排在消费前面", 顺序.index("供货") < 顺序.index("消费"), True)
|
|
|
|
|
|
def 测拼命令(根: Path) -> None:
|
|
"""拼命令: venv 缺要回落 system 并记警告; exec 形态直接给入口; EFI_DB 与配置 env 都要注入."""
|
|
print("\n== 拼命令 ==")
|
|
环境: dict[str, Any] = {"db": {"name": "efi_kernel", "host": "/tmp/socket", "port": 5432, "user": "lou"}}
|
|
目录 = 造驱动(根, "拼命令", {"efi": 1, "name": "拼命令", "runtime": "python", "entry": "跑.py",
|
|
"args": ["--x"], "env": {"我的变量": "1"}})
|
|
驱动 = 扫描.校验单个(目录)
|
|
argv, env, 警告表 = 内核.拼命令(驱动, 环境)
|
|
核对("venv 缺 -> 回落 python3", argv[0], "python3")
|
|
核对("argv 是 解释器 + 入口 + args", argv[1:], [str(目录 / "跑.py"), "--x"])
|
|
核对("有没有记警告", len(警告表) > 0, True)
|
|
核对("注入了 EFI_DB", "host=" in env.get("EFI_DB", ""), True)
|
|
核对("配置里的 env 生效", env.get("我的变量"), "1")
|
|
|
|
venv = 目录 / ".venv" / "bin"
|
|
venv.mkdir(parents=True, exist_ok=True)
|
|
(venv / "python").write_text("#!/bin/sh\n", encoding="utf-8")
|
|
argv, _env, 警告表 = 内核.拼命令(驱动, 环境)
|
|
核对("有 venv 就用 venv", argv[0], str(venv / "python"))
|
|
核对("这次没警告", 警告表, [])
|
|
|
|
exec驱动 = 扫描.校验单个(造驱动(根, "可执行形态", {"efi": 1, "name": "可执行形态", "runtime": "exec",
|
|
"entry": "跑.py"}, 入口可执行=True))
|
|
argv, _env, _警告 = 内核.拼命令(exec驱动, 环境)
|
|
核对("exec 形态直接给入口", argv[0], str(目录.parent / "可执行形态" / "跑.py"))
|
|
|
|
|
|
def 测取链() -> None:
|
|
"""调用链提取 (挡 A->B->A 用)."""
|
|
print("\n== 取链 ==")
|
|
核对("有链", 内核.取链({"args": {"chain": ["x", "y"]}}), ["x", "y"])
|
|
核对("没链", 内核.取链({"args": {}}), [])
|
|
核对("args 不是对象", 内核.取链({"args": "乱写"}), [])
|
|
|
|
|
|
def 测复核(根: Path) -> None:
|
|
"""状态机 / 收尸判定: 真起一个 python 进程去验 (不是纸面推演)."""
|
|
print("\n== 状态.复核 (真起进程) ==")
|
|
目录 = 造驱动(根, "活着的", {"efi": 1, "name": "活着的", "runtime": "python", "entry": "睡.py"}, 入口名="睡.py")
|
|
(目录 / "睡.py").write_text("import time\ntime.sleep(60)\n", encoding="utf-8")
|
|
驱动 = 扫描.校验单个(目录)
|
|
进程 = subprocess.Popen([sys.executable, str(目录 / "睡.py")])
|
|
|
|
try:
|
|
time.sleep(0.4)
|
|
核对("在跑的进程 -> 运行", 状态.复核(驱动, {"state": 状态.运行, "pid": 进程.pid})[0], 状态.运行)
|
|
核对("别人的 pid -> 已退出 (不认领不杀)",
|
|
状态.复核(驱动, {"state": 状态.运行, "pid": os.getpid()})[0], 状态.已退出)
|
|
核对("没在跑的状态维持原样", 状态.复核(驱动, {"state": 状态.停止, "pid": None})[0], 状态.停止)
|
|
核对("oneshot 进程没了算正常退出",
|
|
状态.复核({"mode": "oneshot", "dir": str(目录), "entry": "没这个.py"},
|
|
{"state": 状态.运行, "pid": 999999})[0], 状态.已退出)
|
|
核对("resident 进程没了算崩了",
|
|
状态.复核({"mode": "resident", "dir": str(目录), "entry": "没这个.py"},
|
|
{"state": 状态.运行, "pid": 999999})[0], 状态.崩了)
|
|
finally:
|
|
进程.send_signal(signal.SIGKILL)
|
|
进程.wait(timeout=5)
|
|
|
|
核对("被杀的进程 -> 崩了", 状态.复核(驱动, {"state": 状态.运行, "pid": 进程.pid})[0], 状态.崩了)
|
|
核对("没人记录的驱动但真在跑 -> 认领回运行",
|
|
状态.复核(驱动, {"state": 状态.停止, "pid": None})[0], 状态.停止)
|
|
|
|
|
|
def 测该拉起() -> None:
|
|
"""自动重拉的策略: autostart 管断电, restart=on-failure 管自己崩."""
|
|
print("\n== 该不该拉起 ==")
|
|
拉 = 内核.状态.该拉起
|
|
核对("断电 + autostart", bool(拉({"autostart": True}, {"exit_code": None}, 状态.崩了)), True)
|
|
核对("断电 + 没 autostart", 拉({"autostart": False}, {"exit_code": None}, 状态.崩了), "")
|
|
核对("自己崩 + on-failure", bool(拉({"restart": "on-failure"}, {"exit_code": 3}, 状态.崩了)), True)
|
|
核对("自己崩 + no", 拉({"restart": "no"}, {"exit_code": 3}, 状态.崩了), "")
|
|
核对("失败不自动重试", 拉({"restart": "on-failure"}, {"exit_code": 1}, 状态.失败), "")
|
|
核对("正常退出不拉", 拉({"autostart": True}, {"exit_code": 0}, 状态.已退出), "")
|
|
|
|
|
|
def 测待重启() -> None:
|
|
"""配置改了但还在跑 -> 列表要显示"待重启"."""
|
|
print("\n== 待重启 ==")
|
|
核对("改了要重启", 内核.状态.待重启({"config_hash": "b"}, {"state": 状态.运行, "boot_hash": "a"}), True)
|
|
核对("没改不用", 内核.状态.待重启({"config_hash": "a"}, {"state": 状态.运行, "boot_hash": "a"}), False)
|
|
核对("没在跑不算", 内核.状态.待重启({"config_hash": "b"}, {"state": 状态.停止, "boot_hash": "a"}), False)
|
|
核对("从没起过不算", 内核.状态.待重启({"config_hash": "b"}, {"state": 状态.运行, "boot_hash": None}), False)
|
|
|
|
|
|
def main() -> int:
|
|
"""把各组测试跑一遍, 打印汇总."""
|
|
print(f"内核纯逻辑自测 {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
with tempfile.TemporaryDirectory(prefix="内核自测-") as 临时:
|
|
根 = Path(临时)
|
|
测扫目录(根)
|
|
测校验(根)
|
|
测重名(根)
|
|
测契约(根)
|
|
测拼命令(根)
|
|
测复核(根)
|
|
# 取链/该拉起/待重启 是纯字典运算, 不需要工作目录
|
|
测取链()
|
|
测该拉起()
|
|
测待重启()
|
|
print(f"\n合计: 通过 {通过} / 失败 {失败}")
|
|
return 0 if 失败 == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|