# -*- coding: utf-8 -*- """pbl_blueprint M1a 自查脚本（可在机构工作空间根目录复跑）。 用法： python3 modules/pbl_blueprint/scripts/selfcheck.py # 全量自查 python3 modules/pbl_blueprint/scripts/selfcheck.py --report PATH # 同时写报告 退出码：0=全部通过；1=存在失败项。 检查项（机械可验，全部基于真实落盘文件）： A 目录结构：包目录/__init__.py/init.py/pyproject.toml/scripts/load_path.py/skill/SKILL.md B 表定义：11 表 models/*.json 四段式(summary/fields/indexes/codes)、primary=["id"]、 id str32、tenant_id 首业务字段、金额 double(18,2)、索引 tenant_id 打头 C CRUD 定义：11 个 json/*.json 根键 tblname+params、editable 三个 .dspy、browserfields 非空 D 代码质量：py_compile 全量、禁硬编码库名、sqlor 白名单、三处注册同步 E RBAC：load_path.PATHS 与 init.RBAC_PATHS 一致、显式枚举无通配符 F 契约运行：内存假 sqlor 跑通 蓝图CRUD/树/fork/版本delta/模板实例化/离线兜底/发布/锁 G 租户 fail-closed：tenant_id 缺失/空/通配符/越权一律拒绝 """ import datetime import io import json import os import py_compile import re import sys import traceback HERE = os.path.dirname(os.path.abspath(__file__)) MODULE_ROOT = os.path.dirname(HERE) PKG_DIR = os.path.join(MODULE_ROOT, "pbl_blueprint") MODEL_DIR = os.path.join(PKG_DIR, "models") JSON_DIR = os.path.join(PKG_DIR, "json") if MODULE_ROOT not in sys.path: sys.path.insert(0, MODULE_ROOT) RESULTS = [] def check(group, name, ok, msg=""): RESULTS.append({"group": group, "name": name, "ok": bool(ok), "msg": msg or ""}) return bool(ok) EXPECT_TABLES = [ "pbl_blueprint", "pbl_blueprint_node", "pbl_blueprint_edge", "pbl_blueprint_version", "pbl_blueprint_version_delta", "pbl_blueprint_template", "pbl_blueprint_publish", "pbl_blueprint_offline", "pbl_blueprint_fork", "pbl_blueprint_lock", "pbl_blueprint_audit", ] MONEY_FIELDS = ("budget_amount", "score_amount", "price_amount", "cost_amount", "weight") # ------------------------------------------------------------------ A 目录结构 def check_structure(): g = "A.目录结构" must_files = [ os.path.join(PKG_DIR, "__init__.py"), os.path.join(PKG_DIR, "init.py"), os.path.join(PKG_DIR, "errors.py"), os.path.join(PKG_DIR, "tenant.py"), os.path.join(PKG_DIR, "db.py"), os.path.join(PKG_DIR, "crud.py"), os.path.join(PKG_DIR, "audit.py"), os.path.join(PKG_DIR, "service.py"), os.path.join(MODULE_ROOT, "pyproject.toml"), os.path.join(MODULE_ROOT, "README.md"), os.path.join(HERE, "load_path.py"), os.path.join(MODULE_ROOT, "skill", "SKILL.md"), ] for f in must_files: rel = os.path.relpath(f, MODULE_ROOT) check(g, "存在 %s" % rel, os.path.isfile(f), "" if os.path.isfile(f) else "文件缺失") check(g, "包目录名=模块名 pbl_blueprint/", os.path.isdir(PKG_DIR)) check(g, "models/ 目录存在", os.path.isdir(MODEL_DIR)) check(g, "json/ 目录存在", os.path.isdir(JSON_DIR)) bad = [x for x in ("module.json", "conf", "rp.json", "app.py", "Dockerfile") if os.path.exists(os.path.join(MODULE_ROOT, x))] check(g, "无违禁替代文件(module.json/conf/rp.json/app.py/Dockerfile)", not bad, str(bad)) # ------------------------------------------------------------------ B 表定义 def check_models(): g = "B.表定义" files = sorted([f for f in os.listdir(MODEL_DIR) if f.endswith(".json")]) \ if os.path.isdir(MODEL_DIR) else [] names = [f[:-5] for f in files] check(g, "models 表数量=11", len(names) == 11, "实际 %d: %s" % (len(names), names)) missing = [t for t in EXPECT_TABLES if t not in names] extra = [t for t in names if t not in EXPECT_TABLES] check(g, "11 表清单与预期一致", not missing and not extra, "缺失=%s 多余=%s" % (missing, extra)) for fn in files: tbl = fn[:-5] path = os.path.join(MODEL_DIR, fn) try: with io.open(path, encoding="utf-8") as f: m = json.load(f) except Exception as e: check(g, "%s JSON 可解析" % tbl, False, str(e)) continue seg = [k for k in ("summary", "fields", "indexes", "codes") if k in m] check(g, "%s 四段式齐备(summary/fields/indexes/codes)" % tbl, len(seg) == 4, "缺 %s" % [k for k in ("summary", "fields", "indexes", "codes") if k not in m]) check(g, "%s summary 非空" % tbl, bool(str(m.get("summary") or "").strip())) check(g, '%s primary==["id"]' % tbl, m.get("primary") == ["id"], str(m.get("primary"))) flds = m.get("fields") or {} order = list(flds.keys()) idf = flds.get("id") or {} check(g, "%s id 为 str(32) notnull" % tbl, idf.get("type") == "str" and int(idf.get("size") or 0) == 32 and idf.get("notnull"), str(idf)) check(g, "%s tenant_id 为首业务字段(id 之后第 1 个)" % tbl, len(order) >= 2 and order[0] == "id" and order[1] == "tenant_id", str(order[:3])) tf = flds.get("tenant_id") or {} check(g, "%s tenant_id str(32) notnull" % tbl, tf.get("type") == "str" and int(tf.get("size") or 0) == 32 and tf.get("notnull"), str(tf)) for name, spec in flds.items(): if name in MONEY_FIELDS and spec.get("type") == "double": sz = spec.get("size") good = isinstance(sz, list) and len(sz) == 2 and sz == [18, 2] check(g, "%s.%s 金额 double(18,2)" % (tbl, name), good, str(sz)) check(g, "%s.%s 有 summary" % (tbl, name), bool(str(spec.get("summary") or "").strip())) idxs = m.get("indexes") or {} check(g, "%s indexes 非空" % tbl, len(idxs) > 0, "0 个索引") for iname, ispec in idxs.items(): fs = ispec.get("fields") or [] check(g, "%s.%s 索引 tenant_id 打头" % (tbl, iname), bool(fs) and fs[0] == "tenant_id", str(fs)) check(g, "%s.%s 有 unique 标记" % (tbl, iname), "unique" in ispec) check(g, "%s.%s 有 summary" % (tbl, iname), bool(str(ispec.get("summary") or "").strip())) for c in fs: check(g, "%s.%s 索引列 %s 已定义" % (tbl, iname, c), c in flds) codes = m.get("codes") or {} for cname, cspec in codes.items(): items = cspec.get("items") or {} check(g, "%s.codes.%s 有 items 且非空" % (tbl, cname), len(items) > 0) check(g, "%s.codes.%s 有 summary" % (tbl, cname), bool(str(cspec.get("summary") or "").strip())) if cname in flds: check(g, "%s.codes.%s 对应字段存在" % (tbl, cname), cname in flds) # ------------------------------------------------------------------ C CRUD 定义 def check_crud_json(): g = "C.CRUD定义" files = sorted([f for f in os.listdir(JSON_DIR) if f.endswith(".json")]) \ if os.path.isdir(JSON_DIR) else [] names = [f[:-5] for f in files] check(g, "json/ CRUD 定义数量=11", len(names) == 11, "实际 %d" % len(names)) missing = [t for t in EXPECT_TABLES if t not in names] check(g, "json/ 覆盖全部 11 表", not missing, "缺失=%s" % missing) for fn in files: tbl = fn[:-5] path = os.path.join(JSON_DIR, fn) try: with io.open(path, encoding="utf-8") as f: d = json.load(f) except Exception as e: check(g, "%s CRUD JSON 可解析" % tbl, False, str(e)) continue keys = set(d.keys()) check(g, "%s 根键含 tblname+params" % tbl, "tblname" in keys and "params" in keys, str(sorted(keys))) check(g, "%s 无自创 table/list 根键" % tbl, "table" not in keys and "list" not in keys, str(sorted(keys))) check(g, "%s tblname 与文件名一致" % tbl, d.get("tblname") == tbl, str(d.get("tblname"))) ed = d.get("editable") ok_ed = isinstance(ed, list) and len(ed) == 3 and all( isinstance(x, str) and x.startswith("api/") and x.endswith(".dspy") for x in ed) check(g, "%s editable 为 3 个 api/*.dspy" % tbl, ok_ed, str(ed)) bf = d.get("browserfields") check(g, "%s browserfields 非空列表" % tbl, isinstance(bf, list) and len(bf) > 0, str(bf)) params = d.get("params") or {} check(g, "%s params 非空" % tbl, len(params) > 0, "0 字段") porder = list(params.keys()) check(g, "%s params tenant_id 在 id 之后首位" % tbl, len(porder) >= 2 and porder[0] == "id" and porder[1] == "tenant_id", str(porder[:3])) check(g, "%s params.tenant_id editable=false" % tbl, (params.get("tenant_id") or {}).get("editable") is False, str(params.get("tenant_id"))) check(g, "%s params.id editable=false" % tbl, (params.get("id") or {}).get("editable") is False, str(params.get("id"))) # 与 models 字段一致性 mpath = os.path.join(MODEL_DIR, "%s.json" % tbl) if os.path.isfile(mpath): with io.open(mpath, encoding="utf-8") as f: mm = json.load(f) mfields = set((mm.get("fields") or {}).keys()) diff1 = [k for k in params if k not in mfields] diff2 = [k for k in mfields if k not in params] check(g, "%s params 与 models.fields 字段集一致" % tbl, not diff1 and not diff2, "仅CRUD=%s 仅models=%s" % (diff1, diff2)) for bf_name in (bf or []): check(g, "%s browserfields.%s 在 params 中" % (tbl, bf_name), bf_name in params) # ------------------------------------------------------------------ D 代码质量 def check_code_quality(): g = "D.代码质量" pys = [] for root, dirs, fs in os.walk(MODULE_ROOT): dirs[:] = [d for d in dirs if d not in (".git", "__pycache__", ".venv")] for f in fs: if f.endswith(".py"): pys.append(os.path.join(root, f)) check(g, "Python 文件数>0", len(pys) > 0, str(len(pys))) fails = [] for p in pys: try: py_compile.compile(p, doraise=True, cfile=os.devnull) except Exception as e: fails.append("%s: %s" % (os.path.relpath(p, MODULE_ROOT), e)) check(g, "py_compile 全量编译通过(%d 文件)" % len(pys), not fails, "; ".join(fails)) hard = [] sqlor_bad = [] allowed_sqlor = {"C", "U", "D", "R", "I", "sqlExe"} pat_hard = re.compile(r"(^|[^A-Za-z0-9_])(DBNAME\s*=|DB_NAME\s*=|dbname\s*=\s*['\"][^'\"]+['\"])") pat_sor = re.compile(r"\bsor\.([A-Za-z_][A-Za-z0-9_]*)\s*\(") pat_sor2 = re.compile(r"\b_sor\w*\.([A-Za-z_][A-Za-z0-9_]*)\s*\(") for p in pys: rel = os.path.relpath(p, MODULE_ROOT) if rel.startswith("scripts") or rel.startswith("tests"): continue txt = io.open(p, encoding="utf-8").read() for i, line in enumerate(txt.splitlines(), 1): st = line.strip() if st.startswith("#"): continue if pat_hard.search(line): hard.append("%s:%d %s" % (rel, i, st[:80])) for m in list(pat_sor.finditer(line)) + list(pat_sor2.finditer(line)): if m.group(1) not in allowed_sqlor: sqlor_bad.append("%s:%d sor.%s" % (rel, i, m.group(1))) check(g, "无硬编码库名(DBNAME=/DB_NAME=/dbname='…')", not hard, "; ".join(hard[:5])) check(g, "sqlor 仅用白名单 C/U/D/R/I/sqlExe", not sqlor_bad, "; ".join(sqlor_bad[:5])) # 三处注册同步 try: import pbl_blueprint as pkg from pbl_blueprint import init as init_mod contract = list(getattr(pkg, "CONTRACT_FUNCTIONS", ())) reg = list(getattr(init_mod, "REGISTER_FUNCTIONS", ())) fns = pkg.get_contract_functions() check(g, "__init__.CONTRACT_FUNCTIONS 非空", len(contract) > 0, str(len(contract))) check(g, "init.REGISTER_FUNCTIONS 与 CONTRACT_FUNCTIONS 一致", sorted(contract) == sorted(reg), "仅__init__=%s 仅init=%s" % ([x for x in contract if x not in reg], [x for x in reg if x not in contract])) undef = [n for n in contract if not callable(fns.get(n))] check(g, "契约函数均已定义且可调用", not undef, str(undef)) first_bad = [] import inspect for n in contract: fn = fns.get(n) if not callable(fn): continue try: args = list(inspect.getargspec(fn).args) if hasattr(inspect, "getargspec") \ else list(inspect.signature(fn).parameters.keys()) except Exception: args = [] if not args or args[0] != "tenant_id": first_bad.append("%s(%s)" % (n, args[:1])) check(g, "契约函数首参一律 tenant_id", not first_bad, "; ".join(first_bad[:6])) tables = list(getattr(init_mod, "REGISTER_TABLES", ())) check(g, "init.REGISTER_TABLES=11 且与 models 落盘一致", sorted(tables) == sorted(EXPECT_TABLES) and sorted(tables) == sorted([f[:-5] for f in os.listdir(MODEL_DIR) if f.endswith(".json")]), str(tables)) check(g, "pkg.TABLES 与 init.REGISTER_TABLES 一致", sorted(getattr(pkg, "TABLES", ())) == sorted(tables)) except Exception as e: check(g, "三处注册同步可校验", False, "%s\n%s" % (e, traceback.format_exc()[-500:])) # ------------------------------------------------------------------ E RBAC def check_rbac(): g = "E.RBAC路径" try: sys.path.insert(0, HERE) import load_path as lp from pbl_blueprint import init as init_mod errs = lp.validate() check(g, "load_path.PATHS 合法(无通配符/无重复/.dspy)", not errs, "; ".join(errs[:5])) wild = [p for p in lp.PATHS if "%" in p or "*" in p or "?" in p] check(g, "load_path 无 %/*/? 通配符", not wild, str(wild)) same = list(lp.PATHS) == list(init_mod.RBAC_PATHS) check(g, "load_path.PATHS 与 init.RBAC_PATHS 完全一致(含顺序)", same, "仅load_path=%s 仅init=%s" % ([x for x in lp.PATHS if x not in init_mod.RBAC_PATHS], [x for x in init_mod.RBAC_PATHS if x not in lp.PATHS])) wild2 = [p for p in init_mod.RBAC_PATHS if "%" in p or "*" in p] check(g, "init.RBAC_PATHS 无通配符", not wild2, str(wild2)) check(g, "RBAC 路径数>0", len(lp.PATHS) > 0, str(len(lp.PATHS))) cov = set() for t in EXPECT_TABLES: for p in lp.PATHS: if p.startswith("/api/%s/" % t): cov.add(t) check(g, "11 表均有 RBAC 路径覆盖", len(cov) == 11, "未覆盖=%s" % sorted(set(EXPECT_TABLES) - cov)) except Exception as e: check(g, "RBAC 校验可执行", False, "%s\n%s" % (e, traceback.format_exc()[-400:])) # ------------------------------------------------------------------ F 假 sqlor class FakeSor(object): """内存假 sqlor：只实现白名单 C/U/D/R/I/sqlExe，用于契约级跑通验证。""" def __init__(self, dbname="fake_db"): self.dbname = dbname self.tables = {} self.calls = [] def _t(self, tbl): return self.tables.setdefault(tbl, []) def C(self, tbl, data): self.calls.append(("C", tbl)) rows = self._t(tbl) row = dict(data) pk = row.get("id") for r in rows: if pk is not None and r.get("id") == pk: raise Exception("Duplicate entry for PRIMARY") rows.append(row) return 1 def U(self, tbl, data, where, args=None): self.calls.append(("U", tbl)) n = 0 for r in self._t(tbl): if _match(r, where, args): r.update(data) n += 1 return n def D(self, tbl, where, args=None): self.calls.append(("D", tbl)) rows = self._t(tbl) keep = [r for r in rows if not _match(r, where, args)] n = len(rows) - len(keep) self.tables[tbl] = keep return n def R(self, tbl, where, args=None, fields="*", order="", limit=0, offset=0): self.calls.append(("R", tbl)) out = [dict(r) for r in self._t(tbl) if _match(r, where, args)] if order: for part in reversed([x.strip() for x in order.split(",") if x.strip()]): desc = part.endswith(" desc") col = part.replace(" desc", "").replace(" asc", "").strip() out.sort(key=lambda r: (r.get(col) is None, r.get(col)), reverse=desc) if offset: out = out[offset:] if limit: out = out[:limit] if fields and fields != "*": cols = [c.strip() for c in fields.split(",") if c.strip()] out = [{c: r.get(c) for c in cols} for r in out] return out def I(self, tbl, where="", args=None): self.calls.append(("I", tbl)) return len([r for r in self._t(tbl) if _match(r, where, args)]) def sqlExe(self, sql, args=None): self.calls.append(("sqlExe", sql.split()[0].lower())) s = " ".join(sql.split()).lower() if s.startswith("update pbl_blueprint_template set use_count=use_count+1"): a = list(args or []) n = 0 for r in self._t("pbl_blueprint_template"): if len(a) >= 3 and r.get("tenant_id") == a[1] and r.get("id") == a[2]: r["use_count"] = int(r.get("use_count") or 0) + 1 r["update_time"] = a[0] n += 1 return n if s.startswith("update pbl_blueprint_lock set status='expired'"): a = list(args or []) now = a[0] if a else "" tid = a[1] if len(a) > 1 else "" n = 0 for r in self._t("pbl_blueprint_lock"): if r.get("tenant_id") == tid and r.get("status") == "holding" \ and (r.get("expire_time") or "") < now: r["status"] = "expired" n += 1 return n if s.startswith("create table"): m = re.search(r"create table if not exists ([a-z_0-9]+)", s) if m: self._t(m.group(1)) return 0 return 0 def _match(row, where, args): """极简 WHERE 求值：支持 tenant_id=? / and / or / in (...) / like / 比较 / is null。""" where = str(where or "").strip() if not where: return True args = list(args or []) pos = [0] def take(): v = args[pos[0]] if pos[0] < len(args) else None pos[0] += 1 return v def eval_or(s): parts = _split_top(s, " or ") return any(eval_and(p) for p in parts) def eval_and(s): parts = _split_top(s, " and ") return all(eval_atom(p) for p in parts) def eval_atom(s): s = s.strip() while s.startswith("(") and _balanced(s): s = s[1:-1].strip() low = s.lower() m = re.match(r"^([a-z_0-9]+)\s+in\s*\((.*)\)$", low) if m: col = m.group(1) n = m.group(2).count("?") + 1 vals = [take() for _ in range(n)] return row.get(col) in vals if low.endswith(" is null"): col = low[:-8].strip() return row.get(col) is None if low.endswith(" is not null"): col = low[:-12].strip() return row.get(col) is not None m = re.match(r"^([a-z_0-9]+)\s*(>=|<=|!=|<>|=|>|<)\s*(\?.*)$", low) if m: col, op, rest = m.group(1), m.group(2), m.group(3) val = take() if rest.strip() != "?": return True rv = row.get(col) try: if op == "=": return str(rv) == str(val) if op in ("!=", "<>"): return str(rv) != str(val) if isinstance(rv, (int, float)) and isinstance(val, (int, float)): return {" >": 0}.get(" ") is None and _cmp(rv, op, val) return _cmp(str(rv or ""), op, str(val or "")) except Exception: return False m = re.match(r"^([a-z_0-9]+)\s+like\s+\?$", low) if m: col = m.group(1) pat = str(take() or "") rx = "^" + re.escape(pat).replace("%", ".*").replace("_", ".") + "$" return re.match(rx, str(row.get(col) or "")) is not None # 未识别片段：消耗其占位符后放行 for _ in range(s.count("?")): take() return True return eval_or(where) def _cmp(a, op, b): if op == ">": return a > b if op == ">=": return a >= b if op == "<": return a < b if op == "<=": return a <= b return False def _split_top(s, sep): out, depth, cur, i = [], 0, "", 0 low = s.lower() while i < len(s): if s[i] == "(": depth += 1 elif s[i] == ")": depth -= 1 if depth == 0 and low.startswith(sep, i): out.append(cur) cur = "" i += len(sep) continue cur += s[i] i += 1 out.append(cur) return [x for x in out if x.strip()] def _balanced(s): if not s.startswith("("): return False d = 0 for i, ch in enumerate(s): if ch == "(": d += 1 elif ch == ")": d -= 1 if d == 0: return i == len(s) - 1 return False def _install_fake_sor(): sor = FakeSor() from pbl_blueprint import db as _db _db.clear_cache() _db.get_sor = lambda dbname=None: sor _db.get_dbname = lambda: sor.dbname return sor # ------------------------------------------------------------------ F 契约跑通 def check_runtime(): g = "F.契约运行" try: sor = _install_fake_sor() import pbl_blueprint as bp from pbl_blueprint.errors import PblBlueprintError T = "t_a" T2 = "t_b" r = bp.create_blueprint(T, "海洋生态PBL", subject="科学", grade="五年级", op_user="u1", sor=sor) check(g, "create_blueprint 成功", r["success"] and r["data"]["tenant_id"] == T, str(r)[:120]) bid = r["data"]["id"] check(g, "create_blueprint 生成 str32 id", len(bid) == 32, bid) check(g, "create_blueprint 默认 status=draft/quality=q0", r["data"]["status"] == "draft" and r["data"]["quality_status"] == "q0") check(g, "create_blueprint 自动编码 PBL-BP-", str(r["data"]["code"]).startswith("PBL-BP-"), r["data"]["code"]) got = bp.get_blueprint(T, bid, sor=sor) check(g, "get_blueprint 命中", got["success"] and got["data"]["id"] == bid) try: bp.get_blueprint(T2, bid, sor=sor) check(g, "跨租户 get 被拒(fail-closed)", False, "未抛错") except PblBlueprintError as e: check(g, "跨租户 get 被拒(fail-closed)", e.errcode == bp.ERR_NOT_FOUND, e.errcode) upd = bp.update_blueprint(T, bid, {"summary": "更新简介", "duration_hours": 12}, op_user="u1", sor=sor) check(g, "update_blueprint 成功", upd["success"] and upd["data"]["affected"] >= 1, str(upd)[:120]) try: bp.update_blueprint(T, bid, {"tenant_id": T2}, sor=sor) check(g, "update 禁改 tenant_id", False, "未抛错") except PblBlueprintError as e: check(g, "update 禁改 tenant_id", e.errcode == bp.ERR_PARAM_INVALID, e.errcode) lst = bp.list_blueprints(T, {"subject": "科学"}, sor=sor) check(g, "list_blueprints 条件命中", lst["data"]["total"] >= 1, str(lst["data"]["total"])) lst2 = bp.list_blueprints(T2, {}, sor=sor) check(g, "list_blueprints 租户隔离(他租户 0 条)", lst2["data"]["total"] == 0, str(lst2["data"]["total"])) # 节点：7 类各建一个 node_ids = {} for nt in bp.NODE_TYPES: rr = bp.create_node(T, bid, nt, "%s节点" % nt, spec_json={"k": nt}, weight=1.5, op_user="u1", sor=sor) okk = rr["success"] and rr["data"]["node_type"] == nt node_ids[nt] = rr["data"]["id"] if okk else "" check(g, "create_node[%s] 成功" % nt, okk, str(rr)[:120]) check(g, "7 类子对象全部可建", len([v for v in node_ids.values() if v]) == 7, str(len(bp.NODE_TYPES))) sub = bp.create_node(T, bid, "task", "子任务", parent_id=node_ids["task"], sor=sor) check(g, "create_node 带 parent_id 成功", sub["success"], str(sub)[:120]) try: bp.create_node(T, bid, "badtype", "x", sor=sor) check(g, "非法 node_type 被拒", False, "未抛错") except PblBlueprintError as e: check(g, "非法 node_type 被拒", e.errcode == bp.ERR_PARAM_INVALID, e.errcode) try: bp.create_node(T, bid, "task", "嵌套task", parent_id=node_ids["task"], sor=sor) check(g, "task 嵌套 task 被拒", False, "未抛错") except PblBlueprintError as e: check(g, "task 嵌套 task 被拒", e.errcode == bp.ERR_STATE_INVALID, e.errcode) tree = bp.get_blueprint_tree(T, bid, sor=sor) check(g, "get_blueprint_tree node_count=8", tree["data"]["node_count"] == 8, str(tree["data"]["node_count"])) check(g, "get_blueprint_tree roots=7(1 个 task 子节点挂在父下)", len(tree["data"]["roots"]) == 7, str(len(tree["data"]["roots"]))) # 边 e1 = bp.create_edge(T, bid, node_ids["role"], node_ids["task"], "unlocks", sor=sor) check(g, "create_edge 成功", e1["success"], str(e1)[:120]) try: bp.create_edge(T, bid, node_ids["role"], node_ids["task"], "unlocks", sor=sor) check(g, "重复边被拒", False, "未抛错") except PblBlueprintError as e: check(g, "重复边被拒", e.errcode == bp.ERR_DUPLICATE, e.errcode) try: bp.create_edge(T, bid, node_ids["task"], node_ids["role"], "depends", sor=sor) check(g, "成环边被拒", False, "未抛错") except PblBlueprintError as e: check(g, "成环边被拒", e.errcode == bp.ERR_STATE_INVALID, e.errcode) try: bp.create_edge(T, bid, node_ids["role"], node_ids["role"], "depends", sor=sor) check(g, "自环边被拒", False, "未抛错") except PblBlueprintError as e: check(g, "自环边被拒", e.errcode == bp.ERR_PARAM_INVALID, e.errcode) # 版本 + delta v1 = bp.save_version(T, bid, "draft", "初版", op_user="u1", sor=sor) check(g, "save_version v1 成功", v1["success"] and v1["data"]["version"]["version_no"] == 1, str(v1)[:150]) bp.update_node(T, node_ids["rule"], {"name": "规则改名"}, op_user="u1", sor=sor) bp.create_node(T, bid, "resource", "新增资源", sor=sor) v2 = bp.save_version(T, bid, "minor", "改规则+加资源", op_user="u1", sor=sor) check(g, "save_version v2 成功", v2["success"] and v2["data"]["version"]["version_no"] == 2) check(g, "v2 change_delta 检出变更(>=2)", v2["data"]["delta_count"] >= 2, str(v2["data"]["delta_count"])) d = bp.get_change_delta(T, bid, 1, 2, sor=sor) check(g, "get_change_delta 返回差异行", d["success"] and d["data"]["total"] >= 2, str(d["data"]["total"])) ops = set(x["op_type"] for x in d["data"]["rows"]) check(g, "change_delta 含 add/update", "add" in ops and "update" in ops, str(ops)) vs = bp.list_versions(T, bid, sor=sor) check(g, "list_versions 返回 2 个版本", vs["data"]["total"] == 2, str(vs["data"]["total"])) bp2 = bp.get_blueprint(T, bid, sor=sor) check(g, "current_version 回写为 2", int(bp2["data"]["current_version"]) == 2, str(bp2["data"]["current_version"])) # fork fk = bp.fork_blueprint(T, bid, "副本蓝图", "copy", op_user="u2", sor=sor) check(g, "fork_blueprint 成功", fk["success"], str(fk)[:150]) check(g, "fork 复制节点数=9", fk["data"]["node_count"] == 9, str(fk["data"]["node_count"])) check(g, "fork 复制边数=1", fk["data"]["edge_count"] == 1, str(fk["data"]["edge_count"])) newid = fk["data"]["blueprint"]["id"] check(g, "fork 新蓝图 source=fork 且 source_id 指向源", fk["data"]["blueprint"]["source"] == "fork" and fk["data"]["blueprint"]["source_id"] == bid) fkl = bp.list_forks(T, bid, "children", sor=sor) check(g, "list_forks(children) 命中血缘", fkl["data"]["total"] >= 1, str(fkl["data"]["total"])) fkt = bp.get_blueprint_tree(T, newid, sor=sor) check(g, "fork 后新蓝图树节点数=9", fkt["data"]["node_count"] == 9, str(fkt["data"]["node_count"])) # 模板 + 实例化 tpl_payload = { "blueprint": {"subject": "数学", "grade": "三年级", "summary": "模板"}, "nodes": [ {"id": "n1", "code": "T1", "name": "角色A", "node_type": "role", "spec_json": {"desc": "a"}}, {"id": "n2", "code": "T2", "name": "任务A", "node_type": "task", "parent_id": "n1"}, {"id": "n3", "code": "T3", "name": "产出A", "node_type": "artifact"}, ], "edges": [{"from": "n1", "to": "n2", "edge_type": "unlocks"}], } ct = bp.create_template(T, "数学模板", category="stem", payload=tpl_payload, price_amount=99.5, op_user="u1", sor=sor) check(g, "create_template 成功", ct["success"], str(ct)[:150]) check(g, "create_template 统计 node_count=3/edge_count=1", ct["data"]["node_count"] == 3 and ct["data"]["edge_count"] == 1, "%s/%s" % (ct["data"]["node_count"], ct["data"]["edge_count"])) tid_ = ct["data"]["id"] inst = bp.instantiate_template(T, tid_, "实例化蓝图", owner_id="u9", sor=sor) check(g, "instantiate_template 成功", inst["success"], str(inst)[:150]) check(g, "实例化节点数=3", inst["data"]["node_count"] == 3, str(inst["data"]["node_count"])) check(g, "实例化边数=1", inst["data"]["edge_count"] == 1, str(inst["data"]["edge_count"])) check(g, "实例化蓝图 source=template", inst["data"]["blueprint"]["source"] == "template") itree = bp.get_blueprint_tree(T, inst["data"]["blueprint"]["id"], sor=sor) check(g, "实例化后树 roots=2(n2 挂 n1 下)", len(itree["data"]["roots"]) == 2, str(len(itree["data"]["roots"]))) tl = bp.list_templates(T, {}, sor=sor) check(g, "list_templates 命中", tl["data"]["total"] >= 1, str(tl["data"]["total"])) try: bp.instantiate_template(T2, tid_, "越权实例化", sor=sor) check(g, "他租户实例化模板被拒", False, "未抛错") except PblBlueprintError as e: check(g, "他租户实例化模板被拒", e.errcode == bp.ERR_NOT_FOUND, e.errcode) # 离线兜底 exp = bp.export_offline(T, bid, "海洋离线包", op_user="u1", sor=sor) check(g, "export_offline 成功", exp["success"], str(exp)[:150]) check(g, "export_offline 生成 sha256 校验和(64位)", len(exp["data"]["checksum"]) == 64, exp["data"]["checksum"][:16]) check(g, "export_offline file_size>0", int(exp["data"]["file_size"]) > 0, str(exp["data"]["file_size"])) off_row = [r for r in sor.tables["pbl_blueprint_offline"] if r["id"] == exp["data"]["id"]][0] imp = bp.import_offline(T, off_row["pkg_json"], "离线导入蓝图", op_user="u1", sor=sor) check(g, "import_offline 成功", imp["success"], str(imp)[:150]) check(g, "import_offline 节点数>0", imp["data"]["node_count"] > 0, str(imp["data"]["node_count"])) pay = bp.instantiate_payload(T, tpl_payload, "payload兜底实例化", sor=sor) check(g, "instantiate_payload 离线兜底成功", pay["success"], str(pay)[:150]) try: bp.import_offline(T, {"schema_version": "9.9", "nodes": [1]}, sor=sor) check(g, "schema_version 不兼容被拒", False, "未抛错") except PblBlueprintError as e: check(g, "schema_version 不兼容被拒", e.errcode == bp.ERR_PARAM_INVALID, e.errcode) try: bad_pkg = dict(json.loads(off_row["pkg_json"])) bad_pkg["checksum"] = "0" * 64 bp.import_offline(T, bad_pkg, sor=sor) check(g, "校验和不匹配被拒", False, "未抛错") except PblBlueprintError as e: check(g, "校验和不匹配被拒", e.errcode == bp.ERR_PARAM_INVALID, e.errcode) # 发布 pub = bp.publish_blueprint(T, bid, 0, "class", "cls_1", class_id="cls_1", op_user="u1", sor=sor) check(g, "publish_blueprint 成功", pub["success"], str(pub)[:150]) check(g, "发布后蓝图 status=published", bp.get_blueprint(T, bid, sor=sor)["data"]["status"] == "published") pub2 = bp.publish_blueprint(T, bid, 0, "class", "cls_1", class_id="cls_1", op_user="u1", sor=sor) check(g, "重复发布幂等(idempotent=True)", pub2["success"] and pub2["data"]["idempotent"] is True, str(pub2)[:120]) rv = bp.revoke_publish(T, pub["data"]["publish"]["id"], op_user="u1", sor=sor) check(g, "revoke_publish 成功", rv["success"], str(rv)[:120]) # 锁 lk = bp.lock_blueprint(T, bid, "u1", holder_name="张三", ttl_seconds=600, sor=sor) check(g, "lock_blueprint 成功", lk["success"] and lk["data"]["rev"] == 1, str(lk)[:120]) lk2 = bp.lock_blueprint(T, bid, "u1", ttl_seconds=600, sor=sor) check(g, "同人续锁 rev 递增", lk2["success"] and lk2["data"]["rev"] == 2, str(lk2["data"].get("rev"))) try: bp.lock_blueprint(T, bid, "u2", sor=sor) check(g, "他人抢锁被拒(ERR_LOCKED)", False, "未抛错") except PblBlueprintError as e: check(g, "他人抢锁被拒(ERR_LOCKED)", e.errcode == bp.ERR_LOCKED, e.errcode) try: bp.unlock_blueprint(T, bid, "u2", sor=sor) check(g, "非持锁人解锁被拒", False, "未抛错") except PblBlueprintError as e: check(g, "非持锁人解锁被拒", e.errcode == bp.ERR_FORBIDDEN, e.errcode) ul = bp.unlock_blueprint(T, bid, "u1", sor=sor) check(g, "持锁人解锁成功", ul["success"] and ul["data"]["released"] is True, str(ul)[:120]) cl = bp.clean_expired_locks(T, sor=sor) check(g, "clean_expired_locks 可执行", cl["success"], str(cl)[:120]) # 统计 + 删除 st = bp.blueprint_stats(T, bid, sor=sor) check(g, "blueprint_stats 返回 7 类节点计数", len(st["data"]["node_by_type"]) == 7, str(st["data"]["node_by_type"])) check(g, "blueprint_stats node_total>=9", st["data"]["node_total"] >= 9, str(st["data"]["node_total"])) dl = bp.delete_node(T, node_ids["resource"], op_user="u1", sor=sor) check(g, "delete_node 成功", dl["success"], str(dl)[:120]) dbp = bp.delete_blueprint(T, newid, op_user="u1", cascade=True, sor=sor) check(g, "delete_blueprint 级联成功", dbp["success"] and dbp["data"]["cascade"] > 0, str(dbp["data"])) try: bp.get_blueprint(T, newid, sor=sor) check(g, "删除后 get 不可见(逻辑删除)", False, "未抛错") except PblBlueprintError as e: check(g, "删除后 get 不可见(逻辑删除)", e.errcode == bp.ERR_NOT_FOUND, e.errcode) # 审计 append-only au = bp.list_audit(T, bid, sor=sor) check(g, "审计记录已写入(>0)", au["data"]["total"] > 0, str(au["data"]["total"])) try: bp.crud_update("pbl_blueprint_audit", T, au["data"]["rows"][0]["id"], {"action": "x"}, sor=sor) check(g, "审计表禁改(append-only)", False, "未抛错") except PblBlueprintError as e: check(g, "审计表禁改(append-only)", e.errcode == bp.ERR_FORBIDDEN, e.errcode) try: bp.crud_delete("pbl_blueprint_audit", T, au["data"]["rows"][0]["id"], sor=sor) check(g, "审计表禁删(append-only)", False, "未抛错") except PblBlueprintError as e: check(g, "审计表禁删(append-only)", e.errcode == bp.ERR_FORBIDDEN, e.errcode) # 所有写操作 tenant_id 打头 no_tenant = [r for tbl, rows in sor.tables.items() for r in rows if "tenant_id" in r and not r.get("tenant_id")] check(g, "落库行 tenant_id 全部非空", not no_tenant, str(len(no_tenant))) check(g, "11 张表均有数据落库", len(sor.tables) >= 11, str(sorted(sor.tables.keys()))) api_used = sorted(set(c for c, _ in sor.calls)) check(g, "运行时仅调用 sqlor 白名单 API", set(api_used) <= {"C", "U", "D", "R", "I", "sqlExe"}, str(api_used)) except Exception as e: check(g, "契约运行无异常", False, "%s\n%s" % (e, traceback.format_exc()[-1200:])) # ------------------------------------------------------------------ G 租户 fail-closed def check_tenant(): g = "G.租户fail-closed" try: sor = _install_fake_sor() import pbl_blueprint as bp from pbl_blueprint.errors import PblBlueprintError bad_inputs = [None, "", " ", "*", "tenant%", "a" * 33, "t'or'1=1"] for v in bad_inputs: try: bp.create_blueprint(v, "x", sor=sor) check(g, "非法 tenant_id 被拒: %r" % (v,), False, "未抛错") except PblBlueprintError as e: check(g, "非法 tenant_id 被拒: %r" % (v,), e.errcode == bp.ERR_TENANT_MISSING, e.errcode) try: bp.require_tenant("t_a", "t_b") check(g, "上下文租户不一致被拒", False, "未抛错") except PblBlueprintError as e: check(g, "上下文租户不一致被拒", e.errcode == bp.ERR_FORBIDDEN, e.errcode) w, a = bp.tenant_where("t_a", "id=?", ["x"]) check(g, "tenant_where 以 tenant_id=? 开头", str(w).startswith("tenant_id=?"), str(w)) check(g, "tenant_where 参数 tenant_id 打头", a[0] == "t_a", str(a)) inj = bp.inject_tenant("t_a", {"name": "n", "tenant_id": "hack"}) check(g, "inject_tenant 覆盖外部 tenant_id 且置首位", list(inj.keys())[0] == "tenant_id" and inj["tenant_id"] == "t_a", str(inj)) # 契约函数首参 tenant_id（运行时再验一次） import inspect bad = [] for n in bp.CONTRACT_FUNCTIONS: fn = getattr(bp, n, None) if not callable(fn): bad.append(n) continue try: params = list(inspect.signature(fn).parameters.keys()) except Exception: params = [] if not params or params[0] != "tenant_id": bad.append(n) check(g, "全部契约函数首参为 tenant_id", not bad, str(bad)) except Exception as e: check(g, "租户校验可执行", False, "%s\n%s" % (e, traceback.format_exc()[-600:])) # ------------------------------------------------------------------ H DDL def check_ddl(): g = "H.DDL生成" try: from pbl_blueprint import init as init_mod txt = init_mod.ddl_script() check(g, "ddl_script 生成非空", len(txt) > 500, str(len(txt))) for t in EXPECT_TABLES: check(g, "DDL 含 create table %s" % t, ("create table if not exists `%s`" % t) in txt) check(g, "DDL 无库名前缀硬编码", "`.`" not in txt and "use " not in txt.lower().split("\n")[0]) check(g, "金额列 DDL 为 decimal(18,2)", "decimal(18,2)" in txt) check(g, "主键均为 id", txt.count("primary key (`id`)") == len(EXPECT_TABLES), str(txt.count("primary key (`id`)"))) except Exception as e: check(g, "DDL 生成可执行", False, "%s\n%s" % (e, traceback.format_exc()[-400:])) def main(argv=None): argv = list(sys.argv[1:] if argv is None else argv) report = "" if "--report" in argv: i = argv.index("--report") if i + 1 < len(argv): report = argv[i + 1] t0 = datetime.datetime.now() check_structure() check_models() check_crud_json() check_code_quality() check_rbac() check_runtime() check_tenant() check_ddl() t1 = datetime.datetime.now() total = len(RESULTS) passed = len([r for r in RESULTS if r["ok"]]) failed = [r for r in RESULTS if not r["ok"]] lines = [] lines.append("=" * 78) lines.append("pbl_blueprint M1a 自查报告") lines.append("模块路径 : %s" % MODULE_ROOT) lines.append("执行时间 : %s ~ %s (%.2fs)" % (t0.strftime("%F %T"), t1.strftime("%F %T"), (t1 - t0).total_seconds())) lines.append("Python : %s" % sys.version.split()[0]) lines.append("=" * 78) groups = [] for r in RESULTS: if r["group"] not in groups: groups.append(r["group"]) for gp in groups: rs = [r for r in RESULTS if r["group"] == gp] ps = len([r for r in rs if r["ok"]]) lines.append("") lines.append("[%s] %d/%d 通过" % (gp, ps, len(rs))) for r in rs: flag = "PASS" if r["ok"] else "FAIL" lines.append(" %-4s %s%s" % (flag, r["name"], (" | " + r["msg"]) if (r["msg"] and not r["ok"]) else "")) lines.append("") lines.append("-" * 78) lines.append("合计: %d 项, 通过 %d, 失败 %d" % (total, passed, len(failed))) if failed: lines.append("失败清单:") for r in failed: lines.append(" * [%s] %s | %s" % (r["group"], r["name"], r["msg"])) lines.append("结论: %s" % ("ALL PASS" if not failed else "HAS FAILURES")) lines.append("-" * 78) txt = "\n".join(lines) print(txt) if report: d = os.path.dirname(os.path.abspath(report)) if d and not os.path.isdir(d): os.makedirs(d) with io.open(report, "w", encoding="utf-8") as f: f.write(txt + "\n") print("报告已写入: %s" % report) return 0 if not failed else 1 if __name__ == "__main__": sys.exit(main())