fix(selftest): M11b-2d-B STATE_COLUMNS/EVENT_COLUMNS 提取器兼容四段式 models/*.json（三级解析 A字面量/B1运行时import/B2静态回落），补 nocov/order/count 双向契约断言，554 PASS/0 FAIL
This commit is contained in:
agent.develop
parent
c8af5207c8
commit
6532d8f714
@ -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/<table>.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(<TABLE_CONST>)`` → ``<TABLE_CONST> = "<table>"`` → 读唯一真源
|
||||
``modules/world_sync/models/<table>.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(<TABLE_CONST>) → <TABLE_CONST> = "<table>"。"""
|
||||
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/<table>.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:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user