397 lines
18 KiB
Python
397 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-2b 表定义自检（database-table-definition-spec 机械断言）。
|
||
|
||
覆盖范围（QC #4 要求）:
|
||
1. 全部 4 个落点都跑同一套 spec 断言：
|
||
- modules/world_sync/models/pbl_runtime_event.json (唯一真源)
|
||
- modules/world_sync/models/pbl_entity_state.json (唯一真源)
|
||
- apps/scense/pkgs/world_sync/models/pbl_runtime_event.json (打包镜像)
|
||
- apps/scense/pkgs/world_sync/models/pbl_entity_state.json (打包镜像)
|
||
2. 真源与镜像 sha256 两两相同的一致性断言（消除内容级分叉）。
|
||
3. 各段键集合 ⊆ spec 白名单的断言（summary/fields/indexes/codes 均不得出现未定义键）。
|
||
4. 抽象类型断言（禁止 VARCHAR/BIGINT/DATETIME(3)/JSON 等方言具体类型）。
|
||
5. 主键 id str(32) 断言 + 与 pbl_runtime_sql.py 实际列名交叉核对。
|
||
|
||
任一断言不符即 RESULT: FAIL 且退出码 1。
|
||
用法: python3 modules/world_sync/scripts/m11b2_selftest.py
|
||
|
||
────────────────────────────────────────────────────────────────────────────
|
||
[cross-check-sql] 提取器的列名解析规则（M11b-2d-B 固化，下次结构变更照此扩展）
|
||
────────────────────────────────────────────────────────────────────────────
|
||
pbl_runtime_sql.py 里 ``EVENT_COLUMNS`` / ``STATE_COLUMNS`` 有两种合法真源形态，
|
||
提取器必须**同时**兼容，否则自检会在代码演进时假阳（M11b-2c 实跑 4 FAIL 即此因）：
|
||
|
||
形态 A（旧·两段式手抄字面量）
|
||
EVENT_COLUMNS = (
|
||
"id", "event_id", ...
|
||
)
|
||
→ 正则 ``NAME\\s*=\\s*\\((.*?)\\)`` + 引号内标识符，直接取字面量列名。
|
||
|
||
形态 B（M11b-2b 起·四段式 models/*.json 机械派生，当前真态）
|
||
STATE_TABLE = "pbl_entity_state"
|
||
STATE_COLUMNS = columns_of(STATE_TABLE)
|
||
列名不再手抄，由 :mod:`world_sync.pbl_runtime_authority` 按
|
||
``models/<table>.json`` 的 ``fields[].name``（spec 四段式第三段之前的
|
||
fields 段）有序导出，fail-closed（模型缺失/非四段式/fields 空 →
|
||
AuthorityResolutionError，不退回旧常量）。
|
||
解析优先级：
|
||
B1 运行时取真值：sys.path 注入 ``modules/world_sync``，import
|
||
``world_sync.pbl_runtime_sql``，读模块属性（= 生产代码实际使用的列契约，
|
||
与「手抄正则」相比更强：常量表达式怎么变都能拿到最终值）。
|
||
B2 静态回落：import 失败（依赖/环境差异）时，按形态 B 正则解析出
|
||
``columns_of(<TABLE_CONST>)`` → ``<TABLE_CONST> = "<table>"`` → 读唯一真源
|
||
``modules/world_sync/models/<table>.json`` 的 ``fields[].name``。
|
||
A/B 都解析不出 → 记 FAIL（不静默跳过），保证门禁不失效。
|
||
|
||
覆盖强度说明（未放宽任何断言）：形态 A 原本只断「代码列 ⊆ fields 列」单向；
|
||
形态 B 下列集由 fields 派生，单向断言退化为恒真，故补两条等价替代断言把强度
|
||
恢复到「代码契约 == 表契约」双向：
|
||
- [cross-check-sql] 代码列 ⊆ fields 列（原有，语义不变）
|
||
- [cross-check-sql-nocov] fields 列 ⊆ 代码列（真源不得声明代码不认的幽灵列）
|
||
- [cross-check-sql-order] 代码列顺序 == fields 声明顺序（columns_of 有序导出契约）
|
||
- [cross-check-sql-count] 列数与代码注释宣称的列数一致（防静默丢列）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
MODULE_DIR = os.path.abspath(os.path.join(HERE, "..")) # modules/world_sync
|
||
WS_ROOT = os.path.abspath(os.path.join(MODULE_DIR, "..", "..")) # 机构工作空间根
|
||
|
||
SRC_DIR = os.path.join("modules", "world_sync", "models")
|
||
MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models")
|
||
|
||
TABLES = ("pbl_runtime_event", "pbl_entity_state")
|
||
|
||
# ---- spec 白名单键（database-table-definition-spec）----
|
||
ROOT_KEYS = {"summary", "fields", "indexes", "codes"}
|
||
SUMMARY_KEYS = {"name", "title", "primary", "catelog", "comment"}
|
||
FIELD_KEYS = {"name", "title", "type", "length", "dec", "nullable", "default"}
|
||
INDEX_KEYS = {"name", "idxtype", "idxfields"}
|
||
CODE_KEYS = {"field", "table", "valuefield", "textfield", "cond"}
|
||
|
||
# 抽象类型（spec 表格）
|
||
ABSTRACT_TYPES = {
|
||
"str", "char", "short", "int", "long", "float", "double", "ddouble",
|
||
"decimal", "date", "time", "datetime", "timestamp", "text", "bin",
|
||
}
|
||
NEEDS_LENGTH = {"str", "char", "float", "double", "ddouble", "decimal"}
|
||
NEEDS_DEC = {"float", "double", "ddouble", "decimal"}
|
||
|
||
# 真源 fields 必须覆盖的代码侧列名（pbl_runtime_sql.py）
|
||
SQL_MODULE = os.path.join("modules", "world_sync", "world_sync", "pbl_runtime_sql.py")
|
||
COLS_IN_CODE = {
|
||
"pbl_runtime_event": "EVENT_COLUMNS",
|
||
"pbl_entity_state": "STATE_COLUMNS",
|
||
}
|
||
# 代码注释里宣称的列数（防静默丢列；改模型时同步更新此表，不一致即 FAIL）
|
||
DECLARED_COL_COUNT = {
|
||
"pbl_runtime_event": 25,
|
||
"pbl_entity_state": 10,
|
||
}
|
||
|
||
results = [] # (ok: bool, line: str)
|
||
_CODE_COLS_CACHE = {} # (tname, const) -> (cols|None, how)
|
||
|
||
|
||
def check(ok: bool, line: str) -> bool:
|
||
results.append((bool(ok), line))
|
||
return bool(ok)
|
||
|
||
|
||
def sha256_of(path: str) -> str:
|
||
h = hashlib.sha256()
|
||
with open(path, "rb") as fh:
|
||
for chunk in iter(lambda: fh.read(65536), b""):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
|
||
def load(path: str):
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
return json.load(fh)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 代码侧列名提取（解析规则见文件头「[cross-check-sql] 提取器的列名解析规则」）
|
||
# --------------------------------------------------------------------------
|
||
def _cols_from_literal(src: str, const: str):
|
||
"""形态 A：NAME = ("a", "b", ...) 手抄字面量元组。"""
|
||
m = re.search(const + r"\s*=\s*\((.*?)\)", src, re.S)
|
||
if not m:
|
||
return None
|
||
cols = re.findall(r'"([a-zA-Z_][a-zA-Z0-9_]*)"', m.group(1))
|
||
return cols or None
|
||
|
||
|
||
def _table_from_columns_of(src: str, const: str):
|
||
"""形态 B 静态解析：NAME = columns_of(<TABLE_CONST>) → <TABLE_CONST> = "<table>"。"""
|
||
m = re.search(const + r"\s*=\s*columns_of\(\s*([A-Za-z_][A-Za-z0-9_]*)\s*\)", src)
|
||
if not m:
|
||
return None
|
||
tconst = m.group(1)
|
||
m2 = re.search(r"\b" + re.escape(tconst) + r"\s*=\s*[\"']([a-zA-Z_][a-zA-Z0-9_]*)[\"']", src)
|
||
return m2.group(1) if m2 else None
|
||
|
||
|
||
def _cols_from_runtime(const: str):
|
||
"""形态 B1：import 生产模块，读运行时常量真值（最贴近实际 SQL 契约）。"""
|
||
try:
|
||
if MODULE_DIR not in sys.path:
|
||
sys.path.insert(0, MODULE_DIR)
|
||
mod = __import__("world_sync.pbl_runtime_sql", fromlist=["pbl_runtime_sql"])
|
||
cols = getattr(mod, const, None)
|
||
if isinstance(cols, (list, tuple)) and cols:
|
||
return [str(c) for c in cols]
|
||
except Exception: # noqa: BLE001 - 环境不具备时回落静态解析，不因此放宽断言
|
||
return None
|
||
return None
|
||
|
||
|
||
def _cols_from_authority_model(table: str):
|
||
"""形态 B2：读唯一真源 models/<table>.json 的 fields[].name（有序）。"""
|
||
p = os.path.join(WS_ROOT, SRC_DIR, table + ".json")
|
||
if not os.path.isfile(p):
|
||
return None
|
||
try:
|
||
doc = load(p)
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
fields = doc.get("fields") or []
|
||
cols = [str(f.get("name")) for f in fields if f.get("name")]
|
||
return cols or None
|
||
|
||
|
||
def extract_code_columns(tname: str, const: str):
|
||
"""返回 (cols|None, how)。cols=None 表示解析失败（调用方记 FAIL，不静默跳过）。"""
|
||
key = (tname, const)
|
||
if key in _CODE_COLS_CACHE:
|
||
return _CODE_COLS_CACHE[key]
|
||
|
||
out = (None, "unresolved")
|
||
sql_path = os.path.join(WS_ROOT, SQL_MODULE)
|
||
if os.path.isfile(sql_path):
|
||
src = open(sql_path, "r", encoding="utf-8").read()
|
||
# A) 旧两段式字面量
|
||
cols = _cols_from_literal(src, const)
|
||
if cols:
|
||
out = (cols, "literal-tuple(形态A)")
|
||
else:
|
||
# B1) 运行时真值
|
||
cols = _cols_from_runtime(const)
|
||
if cols:
|
||
out = (cols, "runtime-import columns_of(形态B1)")
|
||
else:
|
||
# B2) 静态解析表名 → 唯一真源 models/*.json
|
||
table = _table_from_columns_of(src, const)
|
||
if table:
|
||
cols = _cols_from_authority_model(table)
|
||
if cols:
|
||
out = (cols, "static columns_of(%s) -> models/%s.json fields[]"
|
||
% (table, table) + "(形态B2)")
|
||
_CODE_COLS_CACHE[key] = out
|
||
return out
|
||
|
||
|
||
def assert_spec(rel: str) -> None:
|
||
"""对单个落点文件跑全套 spec 断言。"""
|
||
path = os.path.join(WS_ROOT, rel)
|
||
if not check(os.path.isfile(path), f"[exists] {rel}"):
|
||
return
|
||
try:
|
||
doc = load(path)
|
||
check(True, f"[json-parse] {rel} OK")
|
||
except Exception as exc: # noqa: BLE001
|
||
check(False, f"[json-parse] {rel} FAIL: {exc}")
|
||
return
|
||
|
||
# --- 根键四段式 ---
|
||
root = set(doc.keys())
|
||
check(root <= ROOT_KEYS, f"[root-keys] {rel} ⊆ summary/fields/indexes/codes -> {sorted(root)}"
|
||
+ ("" if root <= ROOT_KEYS else f" 多余={sorted(root - ROOT_KEYS)}"))
|
||
|
||
# --- summary: 恰好一条，键白名单 ---
|
||
summary = doc.get("summary") or []
|
||
check(len(summary) == 1, f"[summary-count] {rel} 恰好 1 条 -> {len(summary)}")
|
||
if not summary:
|
||
return
|
||
s0 = summary[0]
|
||
skeys = set(s0.keys())
|
||
check(skeys <= SUMMARY_KEYS, f"[summary-keys] {rel} ⊆ {sorted(SUMMARY_KEYS)}"
|
||
+ ("" if skeys <= SUMMARY_KEYS else f" 非白名单键={sorted(skeys - SUMMARY_KEYS)}"))
|
||
tname = os.path.basename(rel)[: -len(".json")]
|
||
check(s0.get("name") == tname, f"[summary-name] {rel} name=={tname} -> {s0.get('name')}")
|
||
check(bool(str(s0.get("title") or "").strip()), f"[summary-title] {rel} title 非空")
|
||
primary = s0.get("primary")
|
||
check(isinstance(primary, list) and primary == ["id"],
|
||
f"[summary-primary] {rel} primary==['id'] (array) -> {primary!r}")
|
||
check(str(s0.get("catelog") or "") in {"entity", "relation", "dimession", "indication", ""},
|
||
f"[summary-catelog] {rel} catelog 合法 -> {s0.get('catelog')}")
|
||
|
||
# --- fields ---
|
||
fields = doc.get("fields") or []
|
||
check(len(fields) >= 1, f"[fields-count] {rel} >=1 -> {len(fields)}")
|
||
names = []
|
||
for f in fields:
|
||
fn = str(f.get("name") or "?")
|
||
names.append(fn)
|
||
tag = f"{rel}#{fn}"
|
||
fkeys = set(f.keys())
|
||
check(fkeys <= FIELD_KEYS, f"[field-keys] {tag} ⊆ {sorted(FIELD_KEYS)}"
|
||
+ ("" if fkeys <= FIELD_KEYS else f" 非白名单键={sorted(fkeys - FIELD_KEYS)}"))
|
||
t = f.get("type")
|
||
check(t in ABSTRACT_TYPES, f"[abstract-type] {tag} type 抽象 -> {t!r}")
|
||
# 方言具体类型黑名单形态（含括号/空格的大写写法）
|
||
check(bool(re.fullmatch(r"[a-z]+", str(t or ""))), f"[type-lower-no-dialect] {tag} -> {t!r}")
|
||
check(bool(str(f.get("title") or "").strip()), f"[field-title] {tag} title 非空(DDL COMMENT 来源)")
|
||
if t in NEEDS_LENGTH:
|
||
ln = f.get("length")
|
||
check(isinstance(ln, int) and not isinstance(ln, bool) and ln > 0,
|
||
f"[field-length] {tag} length 为正整数 -> {ln!r}")
|
||
if t in NEEDS_DEC:
|
||
dc = f.get("dec")
|
||
check(isinstance(dc, int) and not isinstance(dc, bool) and dc > 0,
|
||
f"[field-dec] {tag} dec 为正整数 -> {dc!r}")
|
||
check(str(f.get("nullable") or "yes") in {"yes", "no"}, f"[field-nullable] {tag} -> {f.get('nullable')}")
|
||
check("id" in names, f"[field-id] {rel} 含 id 主键列")
|
||
idf = next((f for f in fields if f.get("name") == "id"), None)
|
||
if idf:
|
||
check(idf.get("type") == "str" and idf.get("length") == 32,
|
||
f"[id-str32] {rel} id type=str length=32 -> {idf.get('type')}/{idf.get('length')}")
|
||
check(idf.get("nullable") == "no", f"[id-notnull] {rel} id nullable=no")
|
||
check(len(names) == len(set(names)), f"[field-unique] {rel} 列名无重复")
|
||
|
||
# --- indexes ---
|
||
indexes = doc.get("indexes") or []
|
||
inames = []
|
||
for ix in indexes:
|
||
nm = str(ix.get("name") or "?")
|
||
inames.append(nm)
|
||
tag = f"{rel}#{nm}"
|
||
ikeys = set(ix.keys())
|
||
check(ikeys <= INDEX_KEYS, f"[index-keys] {tag} ⊆ {sorted(INDEX_KEYS)}"
|
||
+ ("" if ikeys <= INDEX_KEYS else f" 非白名单键={sorted(ikeys - INDEX_KEYS)}"))
|
||
check(ix.get("idxtype") in {"unique", "index"}, f"[index-type] {tag} -> {ix.get('idxtype')}")
|
||
ixfs = ix.get("idxfields")
|
||
check(isinstance(ixfs, list) and len(ixfs) > 0, f"[index-idxfields-array] {tag} -> {ixfs!r}")
|
||
if isinstance(ixfs, list):
|
||
unknown = [c for c in ixfs if c not in names]
|
||
check(not unknown, f"[index-cols-exist] {tag} 索引列都在 fields 中 -> {ixfs}"
|
||
+ ("" if not unknown else f" 缺列={unknown}"))
|
||
check(len(inames) == len(set(inames)), f"[index-unique-name] {rel} 索引名无重复")
|
||
# 租户隔离：所有唯一索引以 tenant_id 打头
|
||
for ix in indexes:
|
||
if ix.get("idxtype") == "unique":
|
||
ixfs = ix.get("idxfields") or []
|
||
check(bool(ixfs) and ixfs[0] == "tenant_id",
|
||
f"[tenant-first] {rel}#{ix.get('name')} 唯一索引 tenant_id 打头 -> {ixfs}")
|
||
|
||
# --- 交叉核对：代码里的列名全部在册（兼容形态 A/B，见文件头解析规则）---
|
||
const = COLS_IN_CODE.get(tname)
|
||
if const:
|
||
sql_path = os.path.join(WS_ROOT, SQL_MODULE)
|
||
if not os.path.isfile(sql_path):
|
||
check(False, f"[cross-check-sql] {rel} 代码侧真源缺失 {SQL_MODULE}")
|
||
else:
|
||
code_cols, how = extract_code_columns(tname, const)
|
||
if not code_cols:
|
||
check(False, f"[cross-check-sql] {rel} 未能解析 {const}"
|
||
f"（形态A字面量/形态B运行时/形态B静态 三级解析均失败）")
|
||
else:
|
||
missing = [c for c in code_cols if c not in names]
|
||
check(not missing, f"[cross-check-sql] {rel} 覆盖 {const} {len(code_cols)} 列"
|
||
f" via {how}" + ("" if not missing else f" 缺列={missing}"))
|
||
# 等价替代断言（四段式下把单向覆盖升级为双向契约一致，强度不降反升）
|
||
nocov = [c for c in names if c not in set(code_cols)]
|
||
check(not nocov, f"[cross-check-sql-nocov] {rel} fields 无代码未声明列"
|
||
f"（fields⊆{const}）" + ("" if not nocov else f" 幽灵列={nocov}"))
|
||
check(list(code_cols) == list(names),
|
||
f"[cross-check-sql-order] {rel} {const} 顺序 == fields 声明顺序"
|
||
f"（columns_of 有序导出契约）")
|
||
declared = DECLARED_COL_COUNT.get(tname)
|
||
if declared is not None:
|
||
check(len(code_cols) == declared and len(names) == declared,
|
||
f"[cross-check-sql-count] {rel} 列数==宣称 {declared}"
|
||
f" -> code={len(code_cols)} fields={len(names)}")
|
||
|
||
# --- codes ---
|
||
codes = doc.get("codes") or []
|
||
seen_fields = set()
|
||
for c in codes:
|
||
ckeys = set(c.keys())
|
||
check(ckeys <= CODE_KEYS, f"[code-keys] {rel}#{c.get('field')} ⊆ {sorted(CODE_KEYS)}"
|
||
+ ("" if ckeys <= CODE_KEYS else f" 非白名单键={sorted(ckeys - CODE_KEYS)}"))
|
||
check(str(c.get("table") or "") and "." not in str(c.get("table")),
|
||
f"[code-table-no-dot] {rel}#{c.get('field')} table={c.get('table')}")
|
||
if str(c.get("table")) == "appcodes_kv":
|
||
check(str(c.get("cond") or "").startswith("parentid="),
|
||
f"[code-cond-parentid] {rel}#{c.get('field')} cond={c.get('cond')}")
|
||
cf = c.get("field")
|
||
check(cf not in seen_fields, f"[code-no-dup] {rel} field={cf} 不重复")
|
||
seen_fields.add(cf)
|
||
check(cf in names, f"[code-field-exists] {rel} field={cf} 在 fields 中")
|
||
|
||
# 全文无方言具体类型字样（type 值层面已断言，这里兜底扫 type 键）
|
||
dialect = [f.get("name") for f in fields
|
||
if re.search(r"(VARCHAR|BIGINT|DATETIME\(|TIMESTAMP\(|NVARCHAR|INT4|INT8)", str(f.get("type")))]
|
||
check(not dialect, f"[no-dialect-types] {rel} 无方言具体类型" + ("" if not dialect else f" 违规={dialect}"))
|
||
|
||
|
||
def main() -> int:
|
||
paths = []
|
||
for t in TABLES:
|
||
paths.append(os.path.join(SRC_DIR, t + ".json"))
|
||
for t in TABLES:
|
||
paths.append(os.path.join(MIRROR_DIR, t + ".json"))
|
||
|
||
print("== 自检落点（4 个路径，真源 + 镜像同一套 spec 断言）==")
|
||
for p in paths:
|
||
print(" - " + p)
|
||
print()
|
||
|
||
for rel in paths:
|
||
assert_spec(rel)
|
||
|
||
print("== 真源 vs 镜像 sha256 一致性断言 ==")
|
||
for t in TABLES:
|
||
src_rel = os.path.join(SRC_DIR, t + ".json")
|
||
mir_rel = os.path.join(MIRROR_DIR, t + ".json")
|
||
sp, mp = os.path.join(WS_ROOT, src_rel), os.path.join(WS_ROOT, mir_rel)
|
||
if os.path.isfile(sp) and os.path.isfile(mp):
|
||
hs, hm = sha256_of(sp), sha256_of(mp)
|
||
print(f" {t}: src={hs}")
|
||
print(f" {t}: mir={hm}")
|
||
check(hs == hm, f"[mirror-sha256-equal] {t} 真源与镜像字节级一致 -> {hs == hm}")
|
||
else:
|
||
check(False, f"[mirror-sha256-equal] {t} 文件缺失，无法比对")
|
||
print()
|
||
|
||
print("== 代码侧列名解析方式（形态 A/B 兼容诊断）==")
|
||
for t in TABLES:
|
||
cols, how = extract_code_columns(t, COLS_IN_CODE[t])
|
||
print(f" {COLS_IN_CODE[t]}: n={len(cols) if cols else 0} via {how}")
|
||
print()
|
||
|
||
passed = sum(1 for ok, _ in results if ok)
|
||
failed = [(line) for ok, line in results if not ok]
|
||
for ok, line in results:
|
||
print(("PASS " if ok else "FAIL ") + line)
|
||
print()
|
||
print(f"断言合计 {len(results)}：PASS {passed} / FAIL {len(failed)}")
|
||
if failed:
|
||
print("RESULT: FAIL")
|
||
return 1
|
||
print("RESULT: PASS")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|