338 lines
14 KiB
Python
338 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-2b 自测：表定义 JSON 符合 database-table-definition-spec（四段式内联口径）。
|
||
|
||
口径变更（M11b-2b-B1）：world_sync/models/pbl_runtime_event.json、pbl_entity_state.json
|
||
已由「key=schema 引用式」改为**四段式内联**（summary/fields/indexes/codes），列集与权威真源
|
||
modules/pbl_runtime_ext/models/*.json 内容级对齐，由 m11b2b_column_diff.py 做双向列集 diff 留证，
|
||
不再靠 authority_path 运行时引用。本脚本断言口径随之更新。
|
||
|
||
覆盖全部 4 个落点（world_sync 真源 2 + 应用打包镜像 2），任一不符即 RESULT: FAIL。
|
||
|
||
断言集（判定项，计入 RESULT）：
|
||
A1 JSON 可解析且根为对象；
|
||
A2 根键 == {"summary","fields","indexes","codes"}（四段式，不再带 key/schema 引用外壳）；
|
||
A3 fields 段非空（四段式产物必须自带列结构体，列集内联落地；旧「不得自带结构体」检查废除）；
|
||
A4 fields[]/indexes[]/codes[] 逐条键集合 ⊆ spec 白名单
|
||
（SPEC_FIELD_KEYS / SPEC_INDEX_KEYS / SPEC_CODE_KEYS；禁止把私有元信息藏进标准结构体条目里）；
|
||
A5 summary[0].name 非空且 == 文件名去 .json（表名与文件名一致；旧 authority_module/
|
||
authority_path/table 必填检查随引用式口径一并废除）；
|
||
A7 真源与镜像 sha256 两两相同（禁止内容级分叉）。
|
||
|
||
U 段（上游体检项，不计入本任务 RESULT，但逐条打印 DEFECT 供冒泡留证）：
|
||
A6 权威真源 modules/pbl_runtime_ext/models/*.json 自身对 spec 四段式白名单/抽象类型的符合性。
|
||
—— 该侧为 M11b-1 已批准产物，本任务按 PM 裁决「不得反向修改」，故只体检登记不判定。
|
||
|
||
信息性打印（不计入判定）：summary 段条目上超出 SPEC_SUMMARY_KEYS 的表级元信息键
|
||
（owner/milestone/decision_ref/decision_basis/structure_note/consumers/mirror_note 等）
|
||
只打印 info 提示——表级元信息允许登记在 summary 记录上，不作为失败判定。
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
SOURCE_TABLES = [
|
||
"pbl_runtime_event.json",
|
||
"pbl_entity_state.json",
|
||
]
|
||
|
||
# 落点 1：world_sync 真源（modules 侧，唯一手改入口）
|
||
MODULE_SRC_DIR = "modules/world_sync/models"
|
||
# 落点 2：应用打包镜像（只由 sync_models_mirror.py 生成，禁止手工双写）
|
||
APP_MIRROR_DIR = os.path.join("apps", "scense", "pkgs", "world_sync", "models")
|
||
# 权威真源（pbl_runtime_ext，M11b-1 已批准产物，本任务只读体检）
|
||
AUTHORITY_DIR = "modules/pbl_runtime_ext/models"
|
||
|
||
# database-table-definition-spec 四段式与各段键白名单
|
||
SPEC_ROOT_KEYS = {"summary", "fields", "indexes", "codes"}
|
||
SPEC_SUMMARY_KEYS = {"name", "title", "primary", "catelog"}
|
||
SPEC_FIELD_KEYS = {"name", "title", "type", "length", "dec", "nullable", "default"}
|
||
SPEC_INDEX_KEYS = {"name", "idxtype", "idxfields"}
|
||
SPEC_CODE_KEYS = {"field", "table", "valuefield", "textfield", "cond"}
|
||
|
||
ABSTRACT_TYPES = {
|
||
"str", "char", "short", "int", "long", "float", "double", "ddouble",
|
||
"decimal", "date", "time", "datetime", "timestamp", "text", "bin",
|
||
}
|
||
NEED_LENGTH = {"str", "char", "float", "double", "ddouble", "decimal"}
|
||
NEED_DEC = {"float", "double", "ddouble", "decimal"}
|
||
|
||
_failures = [] # 判定项失败
|
||
_upstream_defects = [] # 上游体检项缺陷（登记不判定）
|
||
_assert_count = 0
|
||
_upstream_count = 0
|
||
|
||
|
||
def check(cond, label):
|
||
global _assert_count
|
||
_assert_count += 1
|
||
if not cond:
|
||
_failures.append(label)
|
||
print(" FAIL %s" % label)
|
||
else:
|
||
print(" ok %s" % label)
|
||
return bool(cond)
|
||
|
||
|
||
def check_upstream(cond, label):
|
||
global _upstream_count
|
||
_upstream_count += 1
|
||
if not cond:
|
||
_upstream_defects.append(label)
|
||
print(" DEFECT %s" % label)
|
||
else:
|
||
print(" ok %s" % label)
|
||
return bool(cond)
|
||
|
||
|
||
def find_root():
|
||
here = os.path.dirname(os.path.abspath(__file__))
|
||
cur = here
|
||
for _ in range(10):
|
||
if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")):
|
||
return cur
|
||
parent = os.path.dirname(cur)
|
||
if parent == cur:
|
||
break
|
||
cur = parent
|
||
return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir))
|
||
|
||
|
||
def sha256_of(path):
|
||
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_json(path):
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
return json.load(fh)
|
||
|
||
|
||
def as_list(val):
|
||
return val if isinstance(val, list) else []
|
||
|
||
|
||
def stem_of(path):
|
||
base = os.path.basename(path)
|
||
return base[:-5] if base.endswith(".json") else base
|
||
|
||
|
||
def validate_spec(path):
|
||
"""A1~A5：world_sync 侧四段式表定义（真源与镜像执行同一套断言）。"""
|
||
print("== spec(inline four-section): %s" % path)
|
||
try:
|
||
data = load_json(path)
|
||
except Exception as exc: # noqa: BLE001
|
||
check(False, "A1 json-parseable (%s)" % exc)
|
||
return None
|
||
if not check(isinstance(data, dict), "A1 json-is-object"):
|
||
return None
|
||
|
||
# --- A2 根键必须恰好是四段式（引用式外壳 key/schema 视为不合规）---
|
||
check(set(data.keys()) == SPEC_ROOT_KEYS,
|
||
"A2 root-keys == {summary,fields,indexes,codes} (got %s)" % sorted(data.keys()))
|
||
|
||
summary = as_list(data.get("summary"))
|
||
fields = as_list(data.get("fields"))
|
||
indexes = as_list(data.get("indexes"))
|
||
codes = as_list(data.get("codes"))
|
||
|
||
check(len(summary) >= 1, "A2 summary non-empty array (got %d)" % len(summary))
|
||
check(isinstance(data.get("fields"), list), "A2 fields is array")
|
||
check(isinstance(data.get("indexes"), list), "A2 indexes is array")
|
||
check(isinstance(data.get("codes"), list), "A2 codes is array")
|
||
|
||
# --- A3 列结构体内联落地：fields 必须非空（旧口径「引用体内不含结构体」检查已废除）---
|
||
if not check(len(fields) > 0, "A3 fields non-empty (inline column set, got %d)" % len(fields)):
|
||
return data
|
||
|
||
# --- A4 各段条目键集合 ⊆ spec 白名单 ---
|
||
for s_i, s_rec in enumerate(summary):
|
||
if not isinstance(s_rec, dict):
|
||
check(False, "A4 summary[%d] is object" % s_i)
|
||
continue
|
||
extra = set(s_rec.keys()) - SPEC_SUMMARY_KEYS
|
||
if extra:
|
||
print(" info summary[%d] table-level meta keys (not judged): %s" % (s_i, sorted(extra)))
|
||
|
||
field_names = []
|
||
for f in fields:
|
||
if not isinstance(f, dict):
|
||
check(False, "A4 field entry is object (%r)" % f)
|
||
continue
|
||
fname = str(f.get("name", "?"))
|
||
field_names.append(fname)
|
||
extra = set(f.keys()) - SPEC_FIELD_KEYS
|
||
check(not extra, "A4[%s] field keys subset whitelist (extra=%s)" % (fname, sorted(extra)))
|
||
check(fname != "?", "A4[%s] field.name present" % fname)
|
||
|
||
for idx in indexes:
|
||
if not isinstance(idx, dict):
|
||
check(False, "A4 index entry is object (%r)" % idx)
|
||
continue
|
||
iname = str(idx.get("name", "?"))
|
||
extra = set(idx.keys()) - SPEC_INDEX_KEYS
|
||
check(not extra, "A4[%s] index keys subset whitelist (extra=%s)" % (iname, sorted(extra)))
|
||
|
||
for code in codes:
|
||
if not isinstance(code, dict):
|
||
check(False, "A4 code entry is object (%r)" % code)
|
||
continue
|
||
extra = set(code.keys()) - SPEC_CODE_KEYS
|
||
check(not extra, "A4[code:%s] keys subset whitelist (extra=%s)" % (code.get("field", "?"), sorted(extra)))
|
||
|
||
# 列集自洽性（附加判定：内联后本文件自身必须可独立建表）
|
||
for idx in indexes:
|
||
if not isinstance(idx, dict):
|
||
continue
|
||
iname = str(idx.get("name", "?"))
|
||
for ifld in as_list(idx.get("idxfields")):
|
||
check(ifld in field_names, "A4[%s] idxfield present in fields: %s" % (iname, ifld))
|
||
check(idx.get("idxtype") in ("unique", "index"), "A4[%s] idxtype unique|index (%r)" % (iname, idx.get("idxtype")))
|
||
|
||
for f in fields:
|
||
if not isinstance(f, dict):
|
||
continue
|
||
fname = str(f.get("name", "?"))
|
||
t = f.get("type")
|
||
check(t in ABSTRACT_TYPES, "A3[%s] abstract type (%r)" % (fname, t))
|
||
if t in NEED_LENGTH:
|
||
check(isinstance(f.get("length"), int) and f.get("length") > 0, "A3[%s] positive int length" % fname)
|
||
if t in NEED_DEC:
|
||
check(isinstance(f.get("dec"), int) and f.get("dec") > 0, "A3[%s] positive int dec" % fname)
|
||
check(str(f.get("nullable", "no")) in ("yes", "no"), "A3[%s] nullable in yes|no" % fname)
|
||
|
||
# --- A5 表名一致性：summary[0].name 非空且 == 文件名去 .json（旧引用字段检查已废除）---
|
||
expected_name = stem_of(path)
|
||
if not check(len(summary) >= 1 and isinstance(summary[0], dict), "A5 summary[0] is object"):
|
||
return data
|
||
s0_name = str(summary[0].get("name", "")).strip()
|
||
check(bool(s0_name), "A5 summary[0].name non-empty")
|
||
check(s0_name == expected_name,
|
||
"A5 summary[0].name == filename stem (%r vs %r)" % (s0_name, expected_name))
|
||
return data
|
||
|
||
|
||
def validate_authority(path):
|
||
"""A6（上游体检）：pbl_runtime_ext 权威真源对 spec 的符合性，只登记不判定。"""
|
||
print("== authority(upstream audit): %s" % path)
|
||
if not os.path.isfile(path):
|
||
check_upstream(False, "A6 authority file present: %s" % path)
|
||
return
|
||
try:
|
||
data = load_json(path)
|
||
except Exception as exc: # noqa: BLE001
|
||
check_upstream(False, "A6 json-parseable (%s)" % exc)
|
||
return
|
||
check_upstream(isinstance(data, dict), "A6 root-is-object")
|
||
check_upstream(set(data.keys()) <= SPEC_ROOT_KEYS,
|
||
"A6 root-keys subset of four-section (got %s)" % sorted(data.keys()))
|
||
|
||
summary = data.get("summary") or []
|
||
check_upstream(len(summary) == 1, "A6 summary has exactly one record (got %d)" % len(summary))
|
||
if summary and isinstance(summary[0], dict):
|
||
s0 = summary[0]
|
||
check_upstream(set(s0.keys()) <= SPEC_SUMMARY_KEYS,
|
||
"A6 summary keys subset whitelist (got %s)" % sorted(s0.keys()))
|
||
check_upstream(bool(s0.get("name")), "A6 summary.name non-empty")
|
||
check_upstream(isinstance(s0.get("primary"), list) and len(s0.get("primary") or []) > 0,
|
||
"A6 summary.primary is non-empty array")
|
||
|
||
fields = data.get("fields") or []
|
||
check_upstream(len(fields) > 0, "A6 fields non-empty")
|
||
names = []
|
||
for f in fields:
|
||
if not isinstance(f, dict):
|
||
check_upstream(False, "A6 field-is-object")
|
||
continue
|
||
fname = f.get("name", "?")
|
||
names.append(fname)
|
||
check_upstream(set(f.keys()) <= SPEC_FIELD_KEYS,
|
||
"A6[%s] field keys subset whitelist (got %s)" % (fname, sorted(f.keys())))
|
||
t = f.get("type")
|
||
check_upstream(t in ABSTRACT_TYPES, "A6[%s] abstract type (%r)" % (fname, t))
|
||
if t in NEED_LENGTH:
|
||
check_upstream(isinstance(f.get("length"), int) and f.get("length") > 0,
|
||
"A6[%s] positive int length" % fname)
|
||
if t in NEED_DEC:
|
||
check_upstream(isinstance(f.get("dec"), int) and f.get("dec") > 0, "A6[%s] positive int dec" % fname)
|
||
check_upstream(str(f.get("nullable", "no")) in ("yes", "no"), "A6[%s] nullable in yes|no" % fname)
|
||
|
||
for idx in (data.get("indexes") or []):
|
||
if not isinstance(idx, dict):
|
||
check_upstream(False, "A6 index-is-object")
|
||
continue
|
||
iname = idx.get("name", "?")
|
||
check_upstream(set(idx.keys()) <= SPEC_INDEX_KEYS,
|
||
"A6[%s] index keys subset whitelist (got %s)" % (iname, sorted(idx.keys())))
|
||
check_upstream(isinstance(idx.get("idxfields"), list) and len(idx.get("idxfields") or []) > 0,
|
||
"A6[%s] idxfields non-empty array" % iname)
|
||
check_upstream(idx.get("idxtype") in ("unique", "index"), "A6[%s] idxtype unique|index" % iname)
|
||
for ifld in (idx.get("idxfields") or []):
|
||
check_upstream(ifld in names, "A6[%s] idxfield exists in fields: %s" % (iname, ifld))
|
||
|
||
for code in (data.get("codes") or []):
|
||
if not isinstance(code, dict):
|
||
check_upstream(False, "A6 code-is-object")
|
||
continue
|
||
check_upstream(set(code.keys()) <= SPEC_CODE_KEYS,
|
||
"A6 code keys subset whitelist (got %s)" % sorted(code.keys()))
|
||
if code.get("table") == "appcodes_kv":
|
||
check_upstream("parentid=" in str(code.get("cond", "")), "A6 appcodes_kv cond uses parentid=")
|
||
check_upstream("." not in str(code.get("table", "")), "A6 codes.table has no dot notation")
|
||
|
||
|
||
def main():
|
||
root = find_root()
|
||
print("workspace root: %s" % root)
|
||
paths = []
|
||
for name in SOURCE_TABLES:
|
||
paths.append(os.path.join(root, MODULE_SRC_DIR, name))
|
||
paths.append(os.path.join(root, APP_MIRROR_DIR, name))
|
||
|
||
print("\n--- A1~A5: 4 个落点全部执行同一套四段式 spec 断言 ---")
|
||
for p in paths:
|
||
if not os.path.isfile(p):
|
||
check(False, "file exists: %s" % p)
|
||
continue
|
||
validate_spec(p)
|
||
|
||
print("\n--- A7: 真源与镜像 sha256 两两相同 ---")
|
||
for name in SOURCE_TABLES:
|
||
src = os.path.join(root, MODULE_SRC_DIR, name)
|
||
dst = os.path.join(root, APP_MIRROR_DIR, name)
|
||
if os.path.isfile(src) and os.path.isfile(dst):
|
||
hs, hd = sha256_of(src), sha256_of(dst)
|
||
print(" %s\n src=%s\n dst=%s" % (name, hs, hd))
|
||
check(hs == hd, "A7 sha256 equal: %s" % name)
|
||
else:
|
||
check(False, "A7 both files present for sha256: %s" % name)
|
||
|
||
print("\n--- A6: 权威真源（pbl_runtime_ext）上游体检，只登记不判定 ---")
|
||
for name in SOURCE_TABLES:
|
||
validate_authority(os.path.join(root, AUTHORITY_DIR, name))
|
||
|
||
print("\njudged assertions executed: %d" % _assert_count)
|
||
print("upstream audit checks executed: %d" % _upstream_count)
|
||
if _upstream_defects:
|
||
print("UPSTREAM DEFECTS (%d) -> 已登记，需冒泡 agent.pm / pbl_runtime_ext owner：" % len(_upstream_defects))
|
||
for d in _upstream_defects:
|
||
print(" * %s" % d)
|
||
if _failures:
|
||
print("FAILURES (%d):" % len(_failures))
|
||
for f in _failures:
|
||
print(" - %s" % f)
|
||
print("RESULT: FAIL")
|
||
return 1
|
||
print("RESULT: PASS")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|