lou
53e7d1e5f2
文档/ (2026-09-16; 老板定调: 这是 agent 底座, 所以必须扎实, 现在越扎实以后开发越简单)
00-索引 文档地图 + 30 秒概念速查 + 三条命令跑起来 + 事实源优先级
01-快速上手 体检 -> 建库 -> 起内核 -> 起驱动 -> 收工, 全带实测输出; 第一次最易踩的四个坑
02-写一个驱动 五分钟最小驱动 / 形态选择 / 能碰哪些表 / 汇报与调用两份模板 / 交付检查表
03-命令手册 两层每条命令 + 日志选项 + 退出码约定 + --json 样例 + 日常十条
04-契约与调用 一次调用的完整生命周期 / 六条仲裁 / 锁与按需拉起 / 排障表
05-日志与排障 三条道怎么读 + "症状->判据->处置"总表 + 断电收尸语义
06-架构与不变量 分层 / 14 条硬不变量 / 主流程表 / 双真相 / 为什么故意不做 / 已知薄弱点
07-模块与接口 逐模块职责与公开接口 + "想改 X -> 动哪几处"连带清单
08-数据模型 8 张表逐字段 (谁写谁读) + events.kind 字典 + 状态机 + 快照 + 排查 SQL
09-扩展指南 六个配方 (加子命令/加字段/加表/加日志来源/加自测/改判定) + 同步清单
10-验收与质量门 四道门 + 五份自测明细 + pyright 严格档 + 26 条已知坑总表 + 发布 checklist
规矩: 不重复设计文档 / 每条命令实测过再写 (含 jq 表达式) / 代码>设计>文档 的事实源优先级 /
改代码必须同步文档 (清单在 09 末尾) / 暂时没做到的事写成"已知边界"不含糊过去
修复: 同锁串行化原来是死的 (实测抓到的真缺陷)
旧行为: db.领调用 只领 pending (waiting 没人再碰) + db.同锁在跑 把 waiting 也算"占着锁"
-> 同一把锁上两条请求互相排队, 双双停在 waiting 谁也不跑 (实测 id 16/17);
而 收权超时 只收 running -> 排队连超时都没有 = 死锁
修法: ① db.领调用 的 SQL 改 state IN ('pending','waiting') -- 每轮把排队的领回来重判, 锁一空就推进
② db.同锁在跑 只认 state='running' (排队的还没拿到锁, 不挡人)
③ 内核.转发调用 waiting 分支补 deadline (排队也立期限); 内核.收权超时 遍历 running + waiting
④ 抽出 内核.期限文本() 统一算 deadline
实测: 两条同锁调用串行跑完 (19.started_at == 18.finished_at); 排队者超时被收权 (events 有记录)
回归: 自测db.py 调用组 +4 条断言 (waiting 不算占着锁 / waiting 会被重新领 / ...);
去掉一条依赖生产库全局计数的脆弱断言
其它: 内核 与 引导器 的 用法() 末尾加文档指引
验收: uvx pyright 0 errors / 0 warnings; 五份自测全过 (进程/内核 58/配置/db/日志 86);
试跑引导器.py PASS 11 / FAIL 0 / 残留无; 残留进程 0
778 lines
40 KiB
Python
778 lines
40 KiB
Python
"""数据库层真机自测: 43 个函数 / 8 张表, 全打真 PG, 不 mock.
|
|
|
|
用法: .venv/bin/python 内核/自测db.py
|
|
跑完 0 退出; 有 FAIL 退出 1.
|
|
|
|
为什么必须真机:
|
|
db.py 是**唯一碰 SQL 的文件**, 它对不对只有 PG 自己说了算 --
|
|
SQL 拼错 / 列名写错 / jsonb 强转漏了 / text[] 与 list 对不上, 纯逻辑测一律测不出来.
|
|
(自测内核.py 是纯逻辑那份, 这份补的就是"真写库"这一段.)
|
|
|
|
怎么保证不脏生产数据 (两条):
|
|
1. 自己造的行一律带 `_自测db_` 前缀 (驱动名 / source / cmd / caller / kernel / argv / detail),
|
|
跑完按前缀删干净; 末了两道终检: **前缀残留 0 条** + **8 张表行数与开跑前逐表一致**.
|
|
2. 三个**全局**函数 (收尸命令 / 收尾未结束 / 清不在空表) 会把别人的行一起改掉,
|
|
所以它们在自己的事务里跑 (autocommit = False), 验完 ROLLBACK -- 真 SQL 真 PG, 生产行不动.
|
|
|
|
开跑前会先按前缀清一次 (上次跑崩留下的残渣), 有残渣会打出来.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import traceback
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
项目根 = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(项目根 / "内核"))
|
|
|
|
import db # noqa: E402
|
|
import 文本 # noqa: E402
|
|
|
|
# 自测造的行一律带这个前缀 (删的时候只认它, 绝不碰生产行)
|
|
前缀 = "_自测db_"
|
|
# 8 张表 (表名来自这个固定清单, 不是外部输入, 所以下面敢拼进 SQL)
|
|
表名: list[str] = ["drivers", "driver_state", "events", "scans", "commands", "calls", "kernel_env", "kernel_runs"]
|
|
|
|
通过 = 0
|
|
失败 = 0
|
|
组结果: list[tuple[str, int, int]] = []
|
|
|
|
|
|
def 核对(名称: str, 实际: object, 期望: object) -> None:
|
|
"""比一下, 顺手把实测值打出来 (老板要看逐条的实测值, 不看'测试通过')."""
|
|
global 通过, 失败
|
|
if 实际 == 期望:
|
|
通过 += 1
|
|
print(f" [OK] {名称}: {实际!r}")
|
|
return
|
|
失败 += 1
|
|
print(f" [FAIL] {名称}: 实际 {实际!r} != 期望 {期望!r}")
|
|
|
|
|
|
def 核对真(名称: str, 条件: bool, 说明: str) -> None:
|
|
"""条件式核对 (判断不只是"等于某个值"时用, 说明写清楚实测到了什么)."""
|
|
global 通过, 失败
|
|
if 条件:
|
|
通过 += 1
|
|
print(f" [OK] {名称}: {说明}")
|
|
return
|
|
失败 += 1
|
|
print(f" [FAIL] {名称}: {说明}")
|
|
|
|
|
|
def 抓错(名称: str, 动作: Callable[[], None]) -> None:
|
|
"""期望它抛 ValueError (护栏不生效就是 FAIL). 抛别的类型算 FAIL, 不抛也算 FAIL."""
|
|
global 通过, 失败
|
|
try:
|
|
动作()
|
|
except ValueError as 错:
|
|
通过 += 1
|
|
print(f" [OK] {名称}: 按预期抛 ValueError ({错})")
|
|
return
|
|
except Exception as 错: # pylint: disable=broad-except
|
|
失败 += 1
|
|
print(f" [FAIL] {名称}: 抛的是 {type(错).__name__} 而不是 ValueError ({错})")
|
|
return
|
|
失败 += 1
|
|
print(f" [FAIL] {名称}: 该报错的没报错 (护栏失效)")
|
|
|
|
|
|
def 值(行: dict[str, Any] | None, 键: str) -> Any:
|
|
"""从可能为 None 的行里取值 (测试里到处是"行在就取某列, 不在就算了")."""
|
|
return 行.get(键) if 行 is not None else None
|
|
|
|
|
|
def 数行(连接: Any, 表: str) -> int:
|
|
"""某张表的行数 (借 db.查, 免得自测里再抄一份连接代码)."""
|
|
行表 = db.查(连接, f"SELECT count(*) AS 条数 FROM {表}", ())
|
|
return int(行表[0]["条数"]) if 行表 else -1
|
|
|
|
|
|
def 快照(连接: Any) -> dict[str, int]:
|
|
"""8 张表的行数快照 (开跑前 / 收工后各来一份, 逐表比)."""
|
|
return {表: 数行(连接, 表) for 表 in 表名}
|
|
|
|
|
|
def 按前缀数(连接: Any) -> dict[str, int]:
|
|
"""每张表里带自测前缀的残留行数 (收工终检用)."""
|
|
查法: dict[str, tuple[str, tuple[Any, ...]]] = {
|
|
"drivers": ("name LIKE %s", (f"{前缀}%",)),
|
|
"driver_state": ("name LIKE %s", (f"{前缀}%",)),
|
|
"events": ("source LIKE %s", (f"{前缀}%",)),
|
|
"scans": ("kernel LIKE %s", (f"{前缀}%",)),
|
|
"commands": ("source LIKE %s", (f"{前缀}%",)),
|
|
"calls": ("caller LIKE %s", (f"{前缀}%",)),
|
|
"kernel_env": ("detail LIKE %s", (f"{前缀}%",)),
|
|
"kernel_runs": ("argv LIKE %s", (f"{前缀}%",)),
|
|
}
|
|
出: dict[str, int] = {}
|
|
for 表, (条件, 参数) in 查法.items():
|
|
行表 = db.查(连接, f"SELECT count(*) AS 条数 FROM {表} WHERE {条件}", 参数)
|
|
出[表] = int(行表[0]["条数"]) if 行表 else -1
|
|
return 出
|
|
|
|
|
|
def 清场(连接: Any) -> int:
|
|
"""按前缀删掉自测自己造的行 (开跑前清残渣 + 收工后清场共用), 返回删了几行."""
|
|
总 = 0
|
|
删法: list[tuple[str, str]] = [
|
|
("drivers", "name LIKE %s"),
|
|
("events", "source LIKE %s"),
|
|
("scans", "kernel LIKE %s"),
|
|
("commands", "source LIKE %s"),
|
|
("calls", "caller LIKE %s"),
|
|
("kernel_env", "detail LIKE %s"),
|
|
("kernel_runs", "argv LIKE %s"),
|
|
]
|
|
for 表, 条件 in 删法:
|
|
with 连接.cursor() as 游标:
|
|
游标.execute(f"DELETE FROM {表} WHERE {条件}", (f"{前缀}%",))
|
|
总 += int(游标.rowcount)
|
|
return 总
|
|
|
|
|
|
def 现在() -> datetime.datetime:
|
|
"""带时区的时间戳 (timestamptz 列要 aware, 别喂 naive)."""
|
|
return datetime.datetime.now(datetime.timezone.utc)
|
|
|
|
|
|
def 造驱动(name: str, mode: str = "resident", valid: bool = True, error: str | None = None) -> dict[str, Any]:
|
|
"""造一条完整的 drivers 记录 (键名跟表列名一致, 用来验 upsert / 回读)."""
|
|
return {
|
|
"name": name,
|
|
"dir": f"/tmp/{name}",
|
|
"runtime": "python",
|
|
"entry": "跑.py",
|
|
"interpreter": "system",
|
|
"args": ["--自测"],
|
|
"env": {"自测": "1"},
|
|
"provides": ["自测db:能力"],
|
|
"needs": ["自测db:依赖"],
|
|
"autostart": False,
|
|
"restart": "no",
|
|
"mode": mode,
|
|
"config_hash": "哈希-初",
|
|
"entry_hash": "入口哈希-初",
|
|
"valid": valid,
|
|
"error": error,
|
|
"note": f"{前缀}造的行",
|
|
}
|
|
|
|
|
|
# ───────────────────────── ① 连接与建表 ─────────────────────────
|
|
|
|
|
|
def 测连接与建表(库: db.数据库, 连接: Any) -> None:
|
|
"""连 / 配置兜底 / 建表幂等 / 库存在 -- 后面 10 组都踩在这上面."""
|
|
默认 = db.从配置(None)
|
|
核对("从配置(None) 库名兜底", 默认.name, "efi_kernel")
|
|
核对("从配置(None) 端口兜底", 默认.port, 5432)
|
|
核对("从配置(空) 主机兜底", 默认.host, "/home/lou/pgdata/socket")
|
|
|
|
脏 = db.从配置({"name": "自测库", "port": "8080", "host": "1.2.3.4", "user": "谁"})
|
|
核对("从配置 吃配置值", (脏.name, 脏.port, 脏.host, 脏.user), ("自测库", 8080, "1.2.3.4", "谁"))
|
|
核对("从配置 端口是字符串也能转", 脏.port, 8080)
|
|
核对("从配置 端口是垃圾就兜底", db.从配置({"port": "垃圾"}).port, 5432)
|
|
核对("从配置 端口是 None 兜底", db.从配置({"port": None}).port, 5432)
|
|
核对("连接参数 库名覆盖", 默认.连接参数("别的库")["dbname"], "别的库")
|
|
核对("连接参数 默认吃自库", 默认.连接参数()["dbname"], "efi_kernel")
|
|
核对真("描述() 是人读的", 库.描述() == f"{库.name} @ {库.host}:{库.port} (user={库.user})", 库.描述())
|
|
|
|
核对真("连() autocommit 开着", 连接.autocommit is True, f"autocommit={连接.autocommit}")
|
|
行表 = db.查(连接, "SELECT 1 AS 一", ())
|
|
核对("连() 真能查 (SELECT 1)", 行表[0]["一"], 1)
|
|
|
|
管理 = db.连(库, 库名="postgres")
|
|
核对真("库存在(efi_kernel)", db.库存在(管理, 库.name), f"查 pg_database 命中 {库.name}")
|
|
核对真("库存在(瞎编的库名) 为假", not db.库存在(管理, f"{前缀}压根没有这个库"), "没这个库返回 False")
|
|
管理.close()
|
|
|
|
建的全 = db.建表(连接)
|
|
核对("建表() 对象数 (6 表 + 2 索引 + 2 张引导器台账)", len(建的全), 10)
|
|
核对真("建表() 报的是真表名", "drivers" in 建的全 and "events_ts_idx" in 建的全, f"清单 {建的全}")
|
|
核对("建表() 幂等 (再来一遍同样 10 个)", len(db.建表(连接)), 10)
|
|
核对("建引导器表() 只动自己两张", len(db.建引导器表(连接)), 2)
|
|
|
|
行表 = db.查(
|
|
连接,
|
|
"SELECT count(*) AS 表数 FROM information_schema.tables"
|
|
" WHERE table_schema = 'public' AND table_name = ANY(%s)",
|
|
(表名,),
|
|
)
|
|
核对("PG 里真有这 8 张表", int(行表[0]["表数"]), 8)
|
|
|
|
核对("清理函数能跑 (无残渣时删 0 行)", 清场(连接), 0)
|
|
|
|
|
|
# ───────────────────────── ② 注册表 drivers ─────────────────────────
|
|
|
|
|
|
def 测注册表(连接: Any, 生产名单: list[str]) -> None:
|
|
"""upsert / 回读 / 排序 / 清不在 (只删磁盘没了的)."""
|
|
def 造一条(name: str, **改: Any) -> dict[str, Any]:
|
|
记录 = 造驱动(name)
|
|
记录.update(改)
|
|
return 记录
|
|
|
|
甲 = f"{前缀}甲"
|
|
db.记驱动(连接, 造一条(甲))
|
|
行 = db.取驱动(连接, 甲)
|
|
核对真("记驱动 后取得到", 行 is not None, "行不是 None")
|
|
if 行 is not None:
|
|
核对("回读 name", 行["name"], 甲)
|
|
核对("回读 runtime", 行["runtime"], "python")
|
|
核对("回读 entry", 行["entry"], "跑.py")
|
|
核对("回读 mode", 行["mode"], "resident")
|
|
核对("回读 interpreter", 行["interpreter"], "system")
|
|
核对("回读 args (jsonb -> list)", 行["args"], ["--自测"])
|
|
核对("回读 env (jsonb -> dict)", 行["env"], {"自测": "1"})
|
|
核对("回读 provides (text[] -> list)", 行["provides"], ["自测db:能力"])
|
|
核对("回读 needs (text[] -> list)", 行["needs"], ["自测db:依赖"])
|
|
核对("回读 autostart", 行["autostart"], False)
|
|
核对("回读 valid", 行["valid"], True)
|
|
核对("回读 error 为空", 行["error"], None)
|
|
核对真("scanned_at 落了时间", 行["scanned_at"] is not None, str(行["scanned_at"]))
|
|
|
|
# upsert: 同一个名字再扫一次 = 覆盖, 不是插第二条
|
|
db.记驱动(连接, 造一条(甲, mode="oneshot", valid=False, error="自测-配置坏了", config_hash="哈希-新"))
|
|
全部 = db.取全部驱动(连接)
|
|
同名条数 = len([1 for 行 in 全部 if 行["name"] == 甲])
|
|
核对("upsert 不新增行 (同名仍只 1 条)", 同名条数, 1)
|
|
行 = db.取驱动(连接, 甲)
|
|
核对("upsert 覆盖了 mode", 行["mode"] if 行 else None, "oneshot")
|
|
核对("upsert 覆盖了 valid", 行["valid"] if 行 else None, False)
|
|
核对("upsert 覆盖了 error", 行["error"] if 行 else None, "自测-配置坏了")
|
|
核对("upsert 覆盖了 config_hash", 行["config_hash"] if 行 else None, "哈希-新")
|
|
|
|
名字表 = [str(行["name"]) for 行 in 全部]
|
|
核对("取全部驱动 按名字有序", 名字表, sorted(名字表))
|
|
核对真("取驱动(不存在) 给 None", db.取驱动(连接, f"{前缀}压根没有") is None, "返回 None")
|
|
|
|
# 清不在: 保留 = 生产 + 自测, 谁都不该删
|
|
核对("清不在(保留全) 不删任何行", db.清不在(连接, [*生产名单, 甲]), [])
|
|
核对真("清不在 后甲还在", db.取驱动(连接, 甲) is not None, "自测驱动还在")
|
|
|
|
# 清不在: 保留 = 只有生产 -> 自测那条该被删
|
|
db.确保状态行(连接, 甲, 1)
|
|
核对真("删之前 driver_state 有甲", db.取状态(连接, 甲) is not None, "状态行在")
|
|
被删 = db.清不在(连接, 生产名单)
|
|
核对("清不在 报出被删的名字", 被删, [甲])
|
|
核对真("被删驱动不在了", db.取驱动(连接, 甲) is None, "drivers 里没了")
|
|
核对真("driver_state 级联也删了 (不留孤儿)", db.取状态(连接, 甲) is None, "状态行跟着走")
|
|
核对真("生产驱动一个没少", all(db.取驱动(连接, 名) is not None for 名 in 生产名单), f"名单 {生产名单}")
|
|
|
|
|
|
# ───────────────────────── ③ 运行时状态 driver_state ─────────────────────────
|
|
|
|
|
|
def 测状态(连接: Any) -> None:
|
|
"""确保状态行幂等 / 白名单动态 UPDATE / 只改传来的列 (不抹别人的字段)."""
|
|
名 = f"{前缀}状态"
|
|
db.记驱动(连接, 造驱动(名))
|
|
|
|
db.确保状态行(连接, 名, 7)
|
|
db.确保状态行(连接, 名, 99)
|
|
行表 = db.查(连接, "SELECT count(*) AS 条数 FROM driver_state WHERE name = %s", (名,))
|
|
核对("确保状态行 幂等 (只有 1 行)", int(行表[0]["条数"]), 1)
|
|
行 = db.取状态(连接, 名)
|
|
核对("新状态行默认 stopped", 行["state"] if 行 else None, "stopped")
|
|
核对("新状态行 list_version 记的是第一次那版 (不覆盖)", 行["list_version"] if 行 else None, 7)
|
|
|
|
db.写状态(连接, 名, {"state": "running", "pid": 4242, "pgid": 4242, "started_at": 现在(), "restarts": 3,
|
|
"boot_hash": "启动指纹"})
|
|
行 = db.取状态(连接, 名)
|
|
核对("写状态 state", 行["state"] if 行 else None, "running")
|
|
核对("写状态 pid", 行["pid"] if 行 else None, 4242)
|
|
核对("写状态 pgid", 行["pgid"] if 行 else None, 4242)
|
|
核对("写状态 restarts", 行["restarts"] if 行 else None, 3)
|
|
核对("写状态 boot_hash", 行["boot_hash"] if 行 else None, "启动指纹")
|
|
核对真("写状态 started_at 落库", 值(行, "started_at") is not None, str(值(行, "started_at")))
|
|
|
|
# 关键: 只改 pid, 别的列一个都不能被动 (状态就是这么丢的)
|
|
db.写状态(连接, 名, {"pid": 999})
|
|
行 = db.取状态(连接, 名)
|
|
核对("只改 pid 时 boot_hash 不被抹", 行["boot_hash"] if 行 else None, "启动指纹")
|
|
核对("只改 pid 时 state 不变", 行["state"] if 行 else None, "running")
|
|
核对("只改 pid 时 restarts 不变", 行["restarts"] if 行 else None, 3)
|
|
核对("只改 pid 时 started_at 不变", 行["started_at"] is not None if 行 else False, True)
|
|
核对("只改 pid 生效", 行["pid"] if 行 else None, 999)
|
|
|
|
db.写状态(连接, 名, {"state": "stopped", "pid": None, "pgid": None, "exit_code": 0, "stopped_at": 现在()})
|
|
行 = db.取状态(连接, 名)
|
|
核对("写 pid=None 能真写进去", 行["pid"] if 行 else "取不到", None)
|
|
核对("写 exit_code=0 能真写进去", 行["exit_code"] if 行 else None, 0)
|
|
|
|
db.写状态(连接, 名, {})
|
|
核对("写状态({}) 不报错 (空改动直接返回)", db.取状态(连接, 名) is not None, True)
|
|
抓错("写状态(拼错的列名) 被白名单挡住", lambda: db.写状态(连接, 名, {"列名拼错了": 1}))
|
|
抓错("写状态(想改 updated_at) 被挡住", lambda: db.写状态(连接, 名, {"updated_at": 现在()}))
|
|
抓错("写状态(想改主键 name) 被挡住", lambda: db.写状态(连接, 名, {"name": "改名"}))
|
|
|
|
db.写状态(连接, f"{前缀}压根没有这行", {"pid": 1})
|
|
核对("写状态(没有这行) 静默 0 行 (调用方先 确保状态行)", db.取状态(连接, f"{前缀}压根没有这行"), None)
|
|
|
|
全部 = db.取全部状态(连接)
|
|
名字表 = [str(行["name"]) for 行 in 全部]
|
|
核对("取全部状态 按名字有序", 名字表, sorted(名字表))
|
|
核对真("取全部状态 含自测那行", 名 in 名字表, f"共 {len(全部)} 行")
|
|
|
|
|
|
# ───────────────────────── ④ 事件总线 events ─────────────────────────
|
|
|
|
|
|
def 测事件(连接: Any) -> None:
|
|
"""写事件 / 读事件 (条数 / 按驱动过滤) -- 这就是"没有通信协议"的那张总线表."""
|
|
名 = f"{前缀}状态"
|
|
db.写事件(连接, source=f"{前缀}甲", kind="start", message="自测: 起来了", driver=名, data={"pid": 4242})
|
|
db.写事件(连接, source=f"{前缀}甲", kind="error", message="自测: 报了个错", driver=名, level="error",
|
|
data={"原因": "自测"})
|
|
|
|
最近 = db.读事件(连接, 1)
|
|
核对("读事件 新的在前", 最近[0]["message"], "自测: 报了个错")
|
|
核对("读事件 level 回读", 最近[0]["level"], "error")
|
|
核对("读事件 driver 回读", 最近[0]["driver"], 名)
|
|
核对("读事件 data (jsonb) 回读", 最近[0]["data"], {"原因": "自测"})
|
|
|
|
我们 = db.读事件(连接, 10, 名)
|
|
核对("按驱动过滤 只出这个驱动的", [str(行["driver"]) for 行 in 我们], [名, 名])
|
|
核对("按驱动过滤 新的在前", [str(行["kind"]) for 行 in 我们], ["error", "start"])
|
|
核对("按驱动过滤 拿到 data", 我们[1]["data"], {"pid": 4242})
|
|
核对("默认 level 是 info", 我们[1]["level"], "info")
|
|
|
|
核对("按不存在的驱动过滤 = 空表", db.读事件(连接, 10, f"{前缀}压根没有"), [])
|
|
核对("读事件(条数=1) 只给 1 条", len(db.读事件(连接, 1)), 1)
|
|
核对真("读事件 带 id (主键)", isinstance(最近[0]["id"], int), f"id={最近[0]['id']}")
|
|
核对真("读事件 带 ts (时间戳)", 最近[0]["ts"] is not None, str(最近[0]["ts"]))
|
|
|
|
|
|
# ───────────────────────── ⑤ 扫描批次 scans ─────────────────────────
|
|
|
|
|
|
def 测扫描批次(连接: Any) -> None:
|
|
"""list_version 自增 + 最近一次能取回来 (列表页脚那个 扫描 #N)."""
|
|
第一次 = db.记扫描批次(连接, f"{前缀}内核", 3, 3, 0, 2)
|
|
第二次 = db.记扫描批次(连接, f"{前缀}内核", 3, 2, 1, 1)
|
|
核对真("记扫描批次 给自增 list_version", 第二次 > 第一次 > 0, f"第一次 {第一次} 第二次 {第二次}")
|
|
|
|
最近 = db.取最近扫描(连接)
|
|
核对真("取最近扫描 就是刚写那条", 最近 is not None and 最近["list_version"] == 第二次, str(最近))
|
|
if 最近 is not None:
|
|
核对("最近扫描 kernel 字段", 最近["kernel"], f"{前缀}内核")
|
|
核对("最近扫描 total", 最近["total"], 3)
|
|
核对("最近扫描 valid", 最近["valid"], 2)
|
|
核对("最近扫描 invalid", 最近["invalid"], 1)
|
|
核对("最近扫描 running", 最近["running"], 1)
|
|
核对真("最近扫描 带 started_at", 最近["started_at"] is not None, str(最近["started_at"]))
|
|
|
|
|
|
# ───────────────────────── ⑥ 命令 commands (事务里跑) ─────────────────────────
|
|
|
|
|
|
def 测命令(事务连接: Any) -> None:
|
|
"""待领 -> 领走 -> 回结果, 外加"running 的命令不会被重复领"和收尸.
|
|
|
|
这一组在事务里跑 (autocommit=False, 末尾 rollback): 领命令 是全表找第一条 pending,
|
|
生产库里真有待领命令的话会被我们顺手领走 -- 回滚保证它原封不动.
|
|
"""
|
|
口令 = f"{前缀}命令"
|
|
我的id = db.记命令(事务连接, 口令, "列表", {"自测": 1})
|
|
核对真("记命令 返回 id", 我的id > 0, f"id={我的id}")
|
|
|
|
行 = db.读命令(事务连接, 我的id)
|
|
核对("新命令 state=pending", 行["state"] if 行 else None, "pending")
|
|
核对("新命令 args (jsonb) 回读", 行["args"] if 行 else None, {"自测": 1})
|
|
核对("pending 命令 started_at 还是空", 行["started_at"] if 行 else "取不到", None)
|
|
|
|
# 领: pending -> running. 生产里若有 pending, 会先被领到 (回滚掉, 无副作用).
|
|
待领数 = int(db.查(事务连接, "SELECT count(*) AS 条数 FROM commands WHERE state = 'pending'", ())[0]["条数"])
|
|
领到的: list[int] = []
|
|
for _ in range(待领数 + 1):
|
|
条目 = db.领命令(事务连接)
|
|
if 条目 is None:
|
|
break
|
|
领到的.append(int(条目["id"]))
|
|
核对真("领到的命令 state=running", 条目["state"] == "running", f"#{条目['id']} state={条目['state']}")
|
|
核对真("领到的命令 started_at 落了", 条目["started_at"] is not None, str(条目["started_at"]))
|
|
核对真("领命令 能领到我们自己那条", 我的id in 领到的, f"领到 {领到的}")
|
|
|
|
for _ in range(3):
|
|
db.领命令(事务连接)
|
|
行 = db.读命令(事务连接, 我的id)
|
|
核对("被领过的命令不会被重复领 (已 running)", 行["state"] if 行 else None, "running")
|
|
|
|
# CLI 自己执行的命令用 state='running' 落行, 常驻内核不许再领
|
|
我自己跑 = db.记命令(事务连接, 口令, "启动 Json解码", state="running")
|
|
行 = db.读命令(事务连接, 我自己跑)
|
|
核对("记命令(state=running) 立刻有 started_at", 行["started_at"] is not None if 行 else False, True)
|
|
再领: list[int] = []
|
|
for _ in range(3):
|
|
条目 = db.领命令(事务连接)
|
|
if 条目 is not None:
|
|
再领.append(int(条目["id"]))
|
|
核对真("running 的命令不会被领走", 我自己跑 not in 再领, f"又领到 {再领}")
|
|
|
|
db.记命令结果(事务连接, 我的id, "done", {"输出": "自测跑完了"})
|
|
行 = db.读命令(事务连接, 我的id)
|
|
核对("记命令结果 state", 行["state"] if 行 else None, "done")
|
|
核对("记命令结果 result", 行["result"] if 行 else None, {"输出": "自测跑完了"})
|
|
核对真("记命令结果 落了 finished_at", 值(行, "finished_at") is not None, str(值(行, "finished_at")))
|
|
|
|
核对("读命令(不存在) 给 None", db.读命令(事务连接, 99999999), None)
|
|
|
|
# 收尸: 断电留下的 running (finished_at 空) 要能收尾
|
|
db.记命令(事务连接, 口令, "谁都没管它", state="running")
|
|
收了几条 = db.收尸命令(事务连接, "自测收尸")
|
|
核对真("收尸命令 收到了 running 的行", 收了几条 >= 1, f"收了 {收了几条} 条")
|
|
行 = db.读命令(事务连接, 我自己跑)
|
|
核对("被收尸的命令 state=failed", 行["state"] if 行 else None, "failed")
|
|
核对真("收尸 落了 finished_at", 值(行, "finished_at") is not None, str(值(行, "finished_at")))
|
|
核对("收尸 把原因写进 result", 行["result"] if 行 else None, {"detail": "自测收尸"})
|
|
再收 = db.收尸命令(事务连接, "自测再收一次")
|
|
核对("已收过的不会再收 (幂等)", 再收, 0)
|
|
|
|
|
|
# ───────────────────────── ⑦ 调用 calls (事务里跑) ─────────────────────────
|
|
|
|
|
|
def 测调用(事务连接: Any) -> None:
|
|
"""驱动写 pending -> 内核领 -> 转发 -> 回填. 这里照抄 驱动/样例消费器/请求.py 那条 INSERT,
|
|
顺便验"驱动自己写的行, 内核领得到" (db.py 没有插 calls 的函数 -- 入口在驱动那边).
|
|
"""
|
|
我的 = f"{前缀}调用"
|
|
契约 = "自测db:能力"
|
|
锁 = "自测db:锁"
|
|
游标 = 事务连接.cursor()
|
|
游标.execute(
|
|
"INSERT INTO calls (caller, want, args, lock_key)"
|
|
" VALUES (%s, %s, %s::jsonb, %s) RETURNING id",
|
|
(我的, 契约, json.dumps({"chain": [我的]}, ensure_ascii=False), 锁),
|
|
)
|
|
行 = 游标.fetchone()
|
|
游标.close()
|
|
我的id = int(行[0]) if 行 else 0
|
|
核对真("驱动那条 INSERT 写进去了 (真集成)", 我的id > 0, f"calls id={我的id}")
|
|
|
|
领到的 = db.领调用(事务连接)
|
|
核对真("领调用 能领到驱动写的行", 我的id in [int(行["id"]) for 行 in 领到的], f"领到 {len(领到的)} 条")
|
|
# 这里不核"条数": 领调用 现在把 pending 与 waiting 一起领回来 (见下面回归组), 生产库里若正好有
|
|
# 待处理的行, 领到几条不由我们说了算 -- 只核对"我们那条一定在里面" (上面那条足够).
|
|
行 = db.读调用(事务连接, 我的id)
|
|
核对("领走后 state=waiting", 行["state"] if 行 else None, "waiting")
|
|
|
|
db.写调用(事务连接, 我的id, {"state": "running", "provider": f"{前缀}状态", "started_at": 现在()})
|
|
行 = db.读调用(事务连接, 我的id)
|
|
核对("写调用 state", 行["state"] if 行 else None, "running")
|
|
核对("写调用 provider (内核自己记的账)", 行["provider"] if 行 else None, f"{前缀}状态")
|
|
核对("写调用 caller 没被动", 行["caller"] if 行 else None, 我的)
|
|
核对("写调用 want 没被动", 行["want"] if 行 else None, 契约)
|
|
|
|
核对真("同锁在跑 (别的 id 占着同一把锁)", db.同锁在跑(事务连接, 锁, 0), f"lock_key={锁}")
|
|
核对真("同锁在跑 (排除自己就没有别人)", not db.同锁在跑(事务连接, 锁, 我的id), "排除自己 -> False")
|
|
核对真("同锁在跑 (没人用的锁)", not db.同锁在跑(事务连接, "自测db:没人用的锁", 0), "False")
|
|
|
|
运行中的 = db.取调用(事务连接, "running")
|
|
核对真("取调用(running) 含我们那条", 我的id in [int(行["id"]) for 行 in 运行中的], f"{len(运行中的)} 条")
|
|
待办 = db.取调用(事务连接, "pending")
|
|
核对真("取调用(pending) 不含我们那条", 我的id not in [int(行["id"]) for 行 in 待办], f"{len(待办)} 条")
|
|
|
|
抓错("写调用(拼错的列名) 被白名单挡住", lambda: db.写调用(事务连接, 我的id, {"列名拼错了": 1}))
|
|
抓错("写调用(想改主键 id) 被挡住", lambda: db.写调用(事务连接, 我的id, {"id": 1}))
|
|
db.写调用(事务连接, 我的id, {})
|
|
核对("写调用({}) 不报错", db.读调用(事务连接, 我的id) is not None, True)
|
|
|
|
# ── 回归 (2026-09-16 实测踩到的真缺陷, 修完必须有这条守着) ──
|
|
# 旧行为: waiting 的行既不会被重新领、又在 同锁在跑 里被当成"占着锁" ->
|
|
# 同一把锁上两条排队请求**互相排队**, 双双卡在 waiting, 谁也不跑 (实测: 两条都停在 waiting).
|
|
# 新行为: waiting 每轮被领回来重判; 只认 state='running' 算占锁 -> 锁一空就推进.
|
|
锁2 = f"{前缀}同锁"
|
|
两个id: list[int] = []
|
|
for _序 in range(2):
|
|
游标2 = 事务连接.cursor()
|
|
游标2.execute(
|
|
"INSERT INTO calls (caller, want, args, lock_key) VALUES (%s, %s, '{}'::jsonb, %s) RETURNING id",
|
|
(我的, 契约, 锁2),
|
|
)
|
|
行2 = 游标2.fetchone()
|
|
游标2.close()
|
|
两个id.append(int(行2[0]) if 行2 else 0)
|
|
领到的2 = [int(行2["id"]) for 行2 in db.领调用(事务连接)]
|
|
核对真("同锁两条都被领走 (变成 waiting)", all(i in 领到的2 for i in 两个id), f"ids={两个id} 领到={领到的2}")
|
|
核对真("waiting 不算'占着锁' (否则两条互相排队)", not db.同锁在跑(事务连接, 锁2, 两个id[0]), "排除自己后应为 False")
|
|
db.写调用(事务连接, 两个id[0], {"state": "running"})
|
|
核对真("第一条拿到锁后, 第二条才被挡住", db.同锁在跑(事务连接, 锁2, 两个id[1]), f"lock_key={锁2}")
|
|
领到的3 = [int(行2["id"]) for 行2 in db.领调用(事务连接)]
|
|
核对真("waiting 会被重新领 (锁一空就能推进)", 两个id[1] in 领到的3, f"领到={领到的3}")
|
|
|
|
db.写调用(事务连接, 我的id, {"state": "done", "result": {"产出": "自测的产出"}, "finished_at": 现在()})
|
|
行 = db.读调用(事务连接, 我的id)
|
|
核对("回填结果 state", 行["state"] if 行 else None, "done")
|
|
核对("回填结果 result", 行["result"] if 行 else None, {"产出": "自测的产出"})
|
|
核对("读调用(不存在) 给 None", db.读调用(事务连接, 99999999), None)
|
|
|
|
|
|
def 等通知(连接: Any, 超时: float = 3.0) -> list[tuple[str, str]]:
|
|
"""等通知到货 (最多等 超时 秒), 取到就返回.
|
|
|
|
为什么不能 通知完就立刻 poll: NOTIFY 要等发方那条命令提交后才真正投递, 收方这边
|
|
网络 + poll() 都要一点点时间 -- 立刻取会间歇性取到空 (2026-09-16 自测第一次跑就误报 3 条 FAIL).
|
|
"""
|
|
线 = time.monotonic() + 超时
|
|
while True:
|
|
try:
|
|
连接.poll()
|
|
except Exception: # pylint: disable=broad-except
|
|
pass
|
|
收到 = db.收通知(连接)
|
|
if 收到:
|
|
return 收到
|
|
if time.monotonic() >= 线:
|
|
return []
|
|
time.sleep(0.05)
|
|
|
|
|
|
# ───────────────────────── ⑧ LISTEN / NOTIFY (两条真连接) ─────────────────────────
|
|
|
|
|
|
def 测通知(库: db.数据库) -> None:
|
|
"""真 LISTEN / 真 NOTIFY: 一条连接听, 另一条发, 听得见才算通 (这是内核被唤醒的唯一机制)."""
|
|
听者 = db.连(库)
|
|
喊者 = db.连(库)
|
|
通道 = f"{前缀}通道"
|
|
db.监听(听者, 通道)
|
|
db.通知(喊者, 通道, "喂")
|
|
核对("LISTEN/NOTIFY 收到通知", 等通知(听者), [(通道, "喂")])
|
|
核对("收通知 取走就没了 (第二次空)", db.收通知(听者), [])
|
|
|
|
db.通知(喊者, 通道, "")
|
|
核对("空载荷也能收到 (只有叫醒作用)", 等通知(听者), [(通道, "")])
|
|
|
|
db.监听(听者, f"{前缀}中文通道")
|
|
db.通知(喊者, f"{前缀}中文通道", "中文")
|
|
核对("中文通道名 (驱动名有中文, 通道也有)", 等通知(听者), [(f"{前缀}中文通道", "中文")])
|
|
|
|
# 一条连接自己发的自己能不能听着? PG 会把通知投给包括自己在内的所有监听会话 -- 顺手记下真行为
|
|
db.监听(喊者, f"{前缀}自听")
|
|
db.通知(喊者, f"{前缀}自听", "自己喊自己")
|
|
核对("同一连接自己喊自己 (PG 行为, 如实记下)", 等通知(喊者), [(f"{前缀}自听", "自己喊自己")])
|
|
|
|
抓错("监听(带连字符的通道名) 被挡住", lambda: db.监听(听者, "坏-通道"))
|
|
抓错("监听(空格) 被挡住", lambda: db.监听(听者, "坏 通道"))
|
|
抓错("监听(空串) 被挡住", lambda: db.监听(听者, ""))
|
|
听者.close()
|
|
喊者.close()
|
|
|
|
|
|
# ───────────────────────── ⑨ 咨询锁 (独一份调度器) ─────────────────────────
|
|
|
|
|
|
def 测咨询锁(库: db.数据库) -> None:
|
|
"""会话级咨询锁: 抢得到 / 抢不到 / 连接一断自动放 (断电不留死锁)."""
|
|
键 = 0x7A7A7A01 # 自测专用键 (别抢 0x65666901, 那是常驻内核的)
|
|
甲 = db.连(库)
|
|
乙 = db.连(库)
|
|
核对真("甲 抢到锁", db.试锁(甲, 键), f"键 {键}")
|
|
核对真("乙 抢不到 (同一份 PG 内存只能有一个调度器)", not db.试锁(乙, 键), "False")
|
|
核对真("甲 再抢同一把还是 True (会重入计数)", db.试锁(甲, 键), "会话级锁可重入")
|
|
核对真("乙 换一把键就能抢到", db.试锁(乙, 键 + 1), f"键 {键 + 1}")
|
|
甲.close()
|
|
time.sleep(0.2)
|
|
核对真("甲 连接一断, 锁自动放掉 (断电不留死锁)", db.试锁(乙, 键), "乙现在抢到了")
|
|
乙.close()
|
|
|
|
|
|
# ───────────────────────── ⑩ 引导器两张台账 ─────────────────────────
|
|
|
|
|
|
def 测台账(连接: Any, 事务连接: Any) -> None:
|
|
"""kernel_env (体检快照) + kernel_runs (运行台账) -- 引导器那两张."""
|
|
核对("取bool(None)", db.取bool(None), False)
|
|
核对("取bool(True)", db.取bool(True), True)
|
|
核对("取bool(False)", db.取bool(False), False)
|
|
核对("取bool('true')", db.取bool("true"), True)
|
|
核对("取bool('TRUE')", db.取bool("TRUE"), True)
|
|
核对("取bool('1')", db.取bool("1"), True)
|
|
核对("取bool('yes')", db.取bool("yes"), True)
|
|
核对("取bool('ok')", db.取bool("ok"), True)
|
|
核对("取bool('no')", db.取bool("no"), False)
|
|
核对("取bool(数字) 不算真", db.取bool(3), False)
|
|
|
|
# 快照里故意混 None / 字符串 / 真 bool (json 落盘回来就是这样)
|
|
db.记体检(连接, {
|
|
"python": {"version": "3.11.15"},
|
|
"venv": {"path": "/tmp/自测venv", "healthy": "true"},
|
|
"packages": [{"name": "自测包", "want": ">=1", "got": "1.0", "ok": True, "required": True}],
|
|
"pg": {"ok": None},
|
|
"driver_root": {"ok": True},
|
|
"blocking_ok": True,
|
|
"detail": f"{前缀}体检快照",
|
|
})
|
|
行表 = db.查(连接, "SELECT * FROM kernel_env WHERE detail = %s ORDER BY id DESC LIMIT 1", (f"{前缀}体检快照",))
|
|
核对真("记体检 落了一行", len(行表) == 1, "1 行")
|
|
if 行表:
|
|
行 = 行表[0]
|
|
核对("体检 python_version", 行["python_version"], "3.11.15")
|
|
核对("体检 venv_path", 行["venv_path"], "/tmp/自测venv")
|
|
核对("体检 venv_healthy ('true' -> True)", 行["venv_healthy"], True)
|
|
核对("体检 pg_ok (None -> False)", 行["pg_ok"], False)
|
|
核对("体检 driver_root_ok", 行["driver_root_ok"], True)
|
|
核对("体检 ok", 行["ok"], True)
|
|
核对("体检 packages (jsonb 数组)", 行["packages"][0]["name"] if 行["packages"] else None, "自测包")
|
|
|
|
# kernel_runs: 开跑 -> 补 pid -> 收尾
|
|
今日前 = db.今日运行次数(连接)
|
|
运行id = db.记运行开始(连接, f"{前缀}运行", "daemon", os.getpid())
|
|
核对真("记运行开始 返回 id", 运行id > 0, f"id={运行id}")
|
|
核对("今日运行次数 加一", db.今日运行次数(连接), 今日前 + 1)
|
|
|
|
未结束 = db.未结束运行(连接)
|
|
核对真("未结束运行 含刚开的那条", 运行id in [int(行["id"]) for 行 in 未结束], f"共 {len(未结束)} 条")
|
|
最近 = db.最近运行(连接)
|
|
核对真("最近运行 就是刚开那条", 最近 is not None and int(最近["id"]) == 运行id, str(最近))
|
|
|
|
db.记内核pid(连接, 运行id, 4321)
|
|
最近 = db.最近运行(连接)
|
|
核对("记内核pid 补上了", 最近["pid"] if 最近 else None, 4321)
|
|
|
|
db.记运行结束(连接, 运行id, 0, 0.25, True, "自测收尾")
|
|
最近 = db.最近运行(连接)
|
|
核对("记运行结束 exit_code", 最近["exit_code"] if 最近 else None, 0)
|
|
核对("记运行结束 ok", 最近["ok"] if 最近 else None, True)
|
|
核对("记运行结束 detail", 最近["detail"] if 最近 else None, "自测收尾")
|
|
核对真("记运行结束 seconds 落库", abs(float(最近["seconds"]) - 0.25) < 0.01 if 最近 else False,
|
|
f"{最近['seconds'] if 最近 else '?'}")
|
|
核对真("记运行结束 落了 finished_at", 最近["finished_at"] is not None if 最近 else False, "有值")
|
|
核对真("收尾后不在未结束列表里", 运行id not in [int(行["id"]) for 行 in db.未结束运行(连接)], "已收尾")
|
|
|
|
# 收尾未结束 是全局函数 (会把所有没收尾的行改掉) -> 在事务里验, 末尾回滚
|
|
in_tx = db.记运行开始(事务连接, f"{前缀}断电残留", "oneshot", None)
|
|
收到了 = db.收尾未结束(事务连接, "自测收尾未结束")
|
|
核对真("收尾未结束 收到了没收尾的行", 收到了 >= 1, f"收了 {收到了} 条")
|
|
行表 = db.查(事务连接, "SELECT * FROM kernel_runs WHERE id = %s", (in_tx,))
|
|
核对("收尾未结束 给补了 finished_at", 行表[0]["finished_at"] is not None if 行表 else False, True)
|
|
核对("收尾未结束 标 ok=false", 行表[0]["ok"] if 行表 else None, False)
|
|
核对("收尾未结束 写明是谁收的", 行表[0]["detail"] if 行表 else None, "自测收尾未结束")
|
|
|
|
|
|
# ───────────────────────── ⑪ 收工: 清场 + 两道终检 ─────────────────────────
|
|
|
|
|
|
def 测清理与残留(连接: Any, 开跑行数: dict[str, int]) -> None:
|
|
"""自己的行删干净, 再逐表核对: 残留 0 条 + 行数与开跑前一致."""
|
|
名 = f"{前缀}状态"
|
|
db.确保状态行(连接, 名, 1)
|
|
核对真("清场前 自测驱动还在", db.取驱动(连接, 名) is not None, "在")
|
|
核对真("清场前 它的状态行也在", db.取状态(连接, 名) is not None, "在")
|
|
|
|
删了 = 清场(连接)
|
|
核对真("清场 删掉了自己造的行", 删了 >= 1, f"删了 {删了} 行")
|
|
核对真("清场 把自测驱动删了", db.取驱动(连接, 名) is None, "没了")
|
|
核对真("drivers 级联带走了 driver_state (不留孤儿)", db.取状态(连接, 名) is None, "状态行跟着走")
|
|
|
|
残留 = 按前缀数(连接)
|
|
核对("前缀残留 (8 张表逐表)", 残留, dict.fromkeys(表名, 0))
|
|
|
|
收工行数 = 快照(连接)
|
|
核对("8 张表行数与开跑前一致", 收工行数, 开跑行数)
|
|
|
|
|
|
# ───────────────────────── 收口 ─────────────────────────
|
|
|
|
|
|
def 跑组(名称: str, 动作: Callable[[], None]) -> None:
|
|
"""跑一组并记账; 某一组炸了不影响后面的组 (后面的组照样能报问题)."""
|
|
global 通过, 失败
|
|
前的通过, 前的失败 = 通过, 失败
|
|
print(f"\n== {名称} ==")
|
|
try:
|
|
动作()
|
|
except Exception as 错: # pylint: disable=broad-except
|
|
失败 += 1
|
|
print(f" [FAIL] 这一组抛异常: {type(错).__name__}: {错}")
|
|
traceback.print_exc()
|
|
组结果.append((名称, 通过 - 前的通过, 失败 - 前的失败))
|
|
|
|
|
|
def main() -> int:
|
|
"""连真库, 逐组跑, 出汇总表."""
|
|
print(f"db 层真机自测 {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
|
|
配置 = json.loads((项目根 / "环境.efi.json").read_text(encoding="utf-8"))
|
|
库 = db.从配置(配置.get("db"))
|
|
print(f"库: {库.描述()} (测试造的行一律带前缀 {前缀!r}, 只删自己造的 -- 见文件头)")
|
|
|
|
if not db.有psycopg2:
|
|
print("[FAIL] 这个 python 没装 psycopg2, db 层没法验 (引导器可以走降级分支, 自测不行)")
|
|
return 1
|
|
|
|
连接 = db.连(库)
|
|
开跑行数 = 快照(连接)
|
|
开跑前残留 = 按前缀数(连接)
|
|
残渣 = sum(开跑前残留.values())
|
|
if 残渣:
|
|
print(f"[WARN] 上次跑崩留下的残渣 {残渣} 条 (逐表 {开跑前残留}), 先清掉再开始")
|
|
清场(连接)
|
|
开跑行数 = 快照(连接)
|
|
|
|
事务连接 = db.连(库)
|
|
事务连接.autocommit = False
|
|
|
|
生产名单 = [str(行["name"]) for 行 in db.取全部驱动(连接)]
|
|
print(f"生产注册表: {len(生产名单)} 个驱动 {生产名单} (自测绝不删它们)")
|
|
|
|
跑组("① 连接与建表", lambda: 测连接与建表(库, 连接))
|
|
跑组("② 注册表 drivers", lambda: 测注册表(连接, 生产名单))
|
|
跑组("③ 运行时状态 driver_state", lambda: 测状态(连接))
|
|
跑组("④ 事件总线 events", lambda: 测事件(连接))
|
|
跑组("⑤ 扫描批次 scans", lambda: 测扫描批次(连接))
|
|
|
|
def 命令组() -> None:
|
|
测命令(事务连接)
|
|
事务连接.rollback()
|
|
print(" [提示] 本组在事务里跑, 已 ROLLBACK (生产库的 pending 命令原封不动)")
|
|
|
|
def 调用组() -> None:
|
|
测调用(事务连接)
|
|
事务连接.rollback()
|
|
print(" [提示] 本组在事务里跑, 已 ROLLBACK (生产库的 calls 原封不动)")
|
|
|
|
跑组("⑥ 命令 commands (事务)", 命令组)
|
|
跑组("⑦ 调用 calls (事务)", 调用组)
|
|
跑组("⑧ LISTEN / NOTIFY", lambda: 测通知(库))
|
|
|
|
def 锁组() -> None:
|
|
测咨询锁(库)
|
|
|
|
跑组("⑨ 咨询锁 (独一份)", 锁组)
|
|
|
|
def 台账组() -> None:
|
|
测台账(连接, 事务连接)
|
|
|
|
跑组("⑩ 引导器台账 kernel_env / kernel_runs", 台账组)
|
|
事务连接.close()
|
|
跑组("⑪ 清场与残留终检", lambda: 测清理与残留(连接, 开跑行数))
|
|
|
|
连接.close()
|
|
|
|
print("\n" + "=" * 110)
|
|
print("汇总")
|
|
print("=" * 110)
|
|
行表 = [[str(序), 名称, str(过), str(挂), "全过" if 挂 == 0 else "有问题"]
|
|
for 序, (名称, 过, 挂) in enumerate(组结果, 1)]
|
|
文本.打印(文本.表格(["#", "测试组", "通过", "失败", "判定"], 行表,
|
|
[文本.右, 文本.左, 文本.右, 文本.右, 文本.左]))
|
|
print(f"\n 合计: 通过 {通过} / 失败 {失败}")
|
|
print(f" 残留: {'无' if 失败 == 0 else '见上面 FAIL 行'}")
|
|
print(f" 结论: {'全过' if 失败 == 0 else '有 FAIL, 上面逐条看'}")
|
|
return 0 if 失败 == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|