256 lines
11 KiB
Python
256 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-2b 自测：表定义 JSON 符合 database-table-definition-spec（PM 裁决 B：key=schema 引用）。
|
||
|
||
覆盖全部 4 个落点（world_sync 真源 2 + 应用打包镜像 2），任一不符即 RESULT: FAIL。
|
||
|
||
断言集（判定项，计入 RESULT）：
|
||
A1 JSON 可解析；
|
||
A2 根键 == {"key","schema"} 且 key == "schema"（引用形态，不再自带结构体）；
|
||
A3 引用体内不含 summary/fields/indexes/codes 结构体（列集以 authority_path 为唯一真源）；
|
||
A4 各段键集合 ⊆ 白名单（引用体键白名单；禁止把元信息藏在标准结构体里）；
|
||
A5 authority_module/authority_path/table 必填，且 authority_path 指向的真源文件真实存在；
|
||
A7 真源与镜像 sha256 两两相同（禁止内容级分叉）。
|
||
|
||
U 段（上游体检项，不计入本任务 RESULT，但逐条打印 DEFECT 供冒泡留证）：
|
||
A6 权威真源 modules/pbl_runtime_ext/models/*.json 自身对 spec 四段式白名单/抽象类型的符合性。
|
||
—— 该侧为 M11b-1 已批准产物，本任务按 PM 裁决「不得反向修改」，故只体检登记不判定。
|
||
"""
|
||
|
||
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"
|
||
|
||
# 引用体允许出现的键（表级元信息全部登记在此白名单内，不得散落到结构体里）
|
||
REFERENCE_KEY_WHITELIST = {
|
||
"table", "authority_module", "authority_path", "owner", "milestone",
|
||
"decision_ref", "decision_basis", "structure_note", "consumers", "mirror_note",
|
||
"comment", "note",
|
||
}
|
||
|
||
# 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 validate_reference(path, root):
|
||
"""A1~A5：world_sync 侧引用文件（真源与镜像执行同一套断言）。"""
|
||
print("== reference: %s" % path)
|
||
try:
|
||
data = load_json(path)
|
||
except Exception as exc: # noqa: BLE001
|
||
check(False, "A1 json-parseable (%s)" % exc)
|
||
return None
|
||
check(isinstance(data, dict), "A1 json-is-object")
|
||
check(set(data.keys()) == {"key", "schema"}, "A2 root-keys == {key,schema} (got %s)" % sorted(data.keys()))
|
||
check(data.get("key") == "schema", 'A2 key == "schema"')
|
||
body = data.get("schema")
|
||
if not check(isinstance(body, dict), "A2 schema-is-object"):
|
||
return None
|
||
for seg in ("summary", "fields", "indexes", "codes"):
|
||
check(seg not in body, "A3 no self-carried struct: %s" % seg)
|
||
extra = set(body.keys()) - REFERENCE_KEY_WHITELIST
|
||
check(not extra, "A4 reference keys subset whitelist (extra=%s)" % sorted(extra))
|
||
check(bool(str(body.get("table", "")).strip()), "A5 table non-empty")
|
||
check(body.get("authority_module") == "pbl_runtime_ext", "A5 authority_module == pbl_runtime_ext")
|
||
apath = body.get("authority_path", "")
|
||
check(bool(apath), "A5 authority_path non-empty")
|
||
if apath:
|
||
check(os.path.isfile(os.path.join(root, apath)), "A5 authority file exists: %s" % apath)
|
||
return body
|
||
|
||
|
||
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_reference(p, root)
|
||
|
||
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())
|