From 6532d8f71465d84da472364dc561096f7f8fcf76 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 17:19:05 +0800 Subject: [PATCH] =?UTF-8?q?fix(selftest):=20M11b-2d-B=20STATE=5FCOLUMNS/EV?= =?UTF-8?q?ENT=5FCOLUMNS=20=E6=8F=90=E5=8F=96=E5=99=A8=E5=85=BC=E5=AE=B9?= =?UTF-8?q?=E5=9B=9B=E6=AE=B5=E5=BC=8F=20models/*.json=EF=BC=88=E4=B8=89?= =?UTF-8?q?=E7=BA=A7=E8=A7=A3=E6=9E=90=20A=E5=AD=97=E9=9D=A2=E9=87=8F/B1?= =?UTF-8?q?=E8=BF=90=E8=A1=8C=E6=97=B6import/B2=E9=9D=99=E6=80=81=E5=9B=9E?= =?UTF-8?q?=E8=90=BD=EF=BC=89=EF=BC=8C=E8=A1=A5=20nocov/order/count=20?= =?UTF-8?q?=E5=8F=8C=E5=90=91=E5=A5=91=E7=BA=A6=E6=96=AD=E8=A8=80=EF=BC=8C?= =?UTF-8?q?554=20PASS/0=20FAIL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/m11b2_selftest.py | 167 +++++++++++++++++++++++++++++++++++--- 1 file changed, 156 insertions(+), 11 deletions(-) diff --git a/scripts/m11b2_selftest.py b/scripts/m11b2_selftest.py index fa63d12..70643f5 100644 --- a/scripts/m11b2_selftest.py +++ b/scripts/m11b2_selftest.py @@ -15,6 +15,42 @@ 任一断言不符即 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/.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()`` → `` = "
"`` → 读唯一真源 + ``modules/world_sync/models/
.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 @@ -55,8 +91,14 @@ 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: @@ -77,6 +119,87 @@ def load(path: str): 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() → = "
"。""" + 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/
.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) @@ -169,18 +292,34 @@ def assert_spec(rel: str) -> None: 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}")) + + # --- 交叉核对：代码里的列名全部在册（兼容形态 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: - check(False, f"[cross-check-sql] {rel} 未能解析 {COLS_IN_CODE[tname]}") + 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 [] @@ -234,6 +373,12 @@ def main() -> int: 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: