252 lines
11 KiB
Python
252 lines
11 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
|
||
"""
|
||
|
||
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",
|
||
}
|
||
|
||
results = [] # (ok: bool, line: str)
|
||
|
||
|
||
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)
|
||
|
||
|
||
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}")
|
||
# 交叉核对：代码里的列名全部在册
|
||
sql_path = os.path.join(WS_ROOT, SQL_MODULE)
|
||
if os.path.isfile(sql_path) and tname in COLS_IN_CODE:
|
||
src = open(sql_path, "r", encoding="utf-8").read()
|
||
m = re.search(COLS_IN_CODE[tname] + r"\s*=\s*\((.*?)\)", src, re.S)
|
||
if m:
|
||
code_cols = re.findall(r'"([a-zA-Z_][a-zA-Z0-9_]*)"', m.group(1))
|
||
missing = [c for c in code_cols if c not in names]
|
||
check(not missing, f"[cross-check-sql] {rel} 覆盖 {COLS_IN_CODE[tname]} {len(code_cols)} 列"
|
||
+ ("" if not missing else f" 缺列={missing}"))
|
||
else:
|
||
check(False, f"[cross-check-sql] {rel} 未能解析 {COLS_IN_CODE[tname]}")
|
||
|
||
# --- 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()
|
||
|
||
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())
|