#!/usr/bin/env python3 # -*- coding: utf-8 -*- """契约符号补齐后的行为自测（Bug HoSBcdZY4G7IxhaFiIE07）。 不依赖数据库/基座：用假连接与假 env 验证 q_all/q_one 的**双模语义** （同步取值 + await 取值）、q_one 无行返回空 dict、conn= 显式连接路径、 module_dbname 失败不抛、db.get_sor_context 三级降级顺序。 用法:: PYTHONPATH=modules python3 apps/pbls/scripts/_selftest_contract_symbols.py """ import asyncio import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) WS_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(HERE))) MODULES_DIR = os.path.join(WS_ROOT, "modules") # 模块仓库布局是 modules/{模块名}/{模块名}/（包目录=模块名，非 src 布局）， # 故需把每个模块仓库根逐个加入 sys.path，`import pbl_common` 才能命中包。 for _name in sorted(os.listdir(MODULES_DIR)): _repo = os.path.join(MODULES_DIR, _name) if os.path.isdir(os.path.join(_repo, _name)): sys.path.insert(0, _repo) PASS, FAIL = [], [] def check(name, cond, detail=""): (PASS if cond else FAIL).append(name) print(" [%s] %s%s" % ("PASS" if cond else "FAIL", name, (" → " + detail) if detail and not cond else "")) class FakeCursor: """DB-API 风格假游标：%s 占位、description、fetchall。""" def __init__(self, rows, cols): self._rows, self._cols = rows, cols self.executed = None self.description = [(c,) for c in cols] if cols else [] self.closed = False def execute(self, sql, params=()): self.executed = (sql, params) assert "%s" in sql or not params, "占位符应原样交给驱动" return None def fetchall(self): return self._rows def close(self): self.closed = True class FakeConn: def __init__(self, rows, cols): self.cur = FakeCursor(rows, cols) def cursor(self): return self.cur def main(): print("=" * 68) print("契约符号行为自测（q_all / q_one / module_dbname / get_sor_context）") print("=" * 68) from pbl_common import api as A from pbl_common import db as D # --- 1. 符号在位（原 Bug 根因 G 的 crud + 本次补齐的 4 个） --- print("\n[1] 符号在位") for sym in ("crud", "q_all", "q_one", "module_dbname"): check("pbl_common.api.%s 可解析" % sym, hasattr(A, sym)) check("pbl_common.db.get_sor_context 可解析", hasattr(D, "get_sor_context")) check("CONTRACT_SYMBOLS 含 q_all/q_one/module_dbname", all(s in A.CONTRACT_SYMBOLS for s in ("q_all", "q_one", "module_dbname"))) check("__all__ 含 q_all/q_one/module_dbname", all(s in A.__all__ for s in ("q_all", "q_one", "module_dbname"))) check("verify_contract() 无缺失符号（_MISSING 为空）", not getattr(A, "_MISSING", None), str(getattr(A, "_MISSING", None))) # --- 2. conn= 显式连接路径（idempotency.py 的真实调用形态） --- print("\n[2] q_one(sql, params, conn=conn) 同步调用（idempotency.py:141 形态）") conn = FakeConn([("idem1", "completed", 7)], ["idempotency_id", "state", "client_seq"]) row = A.q_one("SELECT idempotency_id, state, client_seq FROM t " "WHERE tenant_id=%s AND client_seq=%s LIMIT 1", ("T1", 7), conn=conn) check("同步返回 dict（非 None、非 coroutine）", isinstance(row, dict)) check("列名映射正确", row.get("state") == "completed", str(row)) check("真值判断 if row: 成立", bool(row) is True) check("参数原样传给驱动", conn.cur.executed[1] == ("T1", 7)) check("游标已关闭（无泄漏）", conn.cur.closed is True) print("\n[3] q_all(sql, params, conn=conn) 多行") conn2 = FakeConn([(1, "a"), (2, "b")], ["seq", "scope"]) rows = A.q_all("SELECT seq, scope FROM t WHERE tenant_id=%s", ("T1",), conn=conn2) check("返回 list 长度 2", isinstance(rows, list) and len(rows) == 2) check("首行 dict 化", rows[0] == {"seq": 1, "scope": "a"}, str(rows[0])) # --- 3. await 双模（rtx_db.py 的真实调用形态） --- print("\n[4] await 双模兼容（rtx_db.py:150 形态）") async def await_paths(): c = FakeConn([("x", 3)], ["k", "v"]) r1 = await A.q_one("SELECT k, v FROM t WHERE id=%s", ("x",), conn=c) c2 = FakeConn([(9,)], ["max_seq"]) r2 = await A.q_all("SELECT MAX(client_seq) AS max_seq FROM t WHERE tenant_id=%s", ("T1",), conn=c2) c3 = FakeConn([], []) r3 = await A.q_one("SELECT 1 FROM t WHERE id=%s", ("none",), conn=c3) return r1, r2, r3 r1, r2, r3 = asyncio.get_event_loop().run_until_complete(await_paths()) check("await q_one 得 dict", isinstance(r1, dict) and r1.get("v") == 3, str(r1)) check("await q_all 得 list", isinstance(r2, list) and len(r2) == 1) check("await 空结果不抛（返回空 dict）", isinstance(r3, dict) and not r3) # --- 4. q_one 无行语义：空 dict 而非 None --- print("\n[5] q_one 无行 → 空 dict（与 (row or {}).get / if row: 兼容）") empty = A.q_one("SELECT 1 FROM t WHERE id=%s", ("nope",), conn=FakeConn([], [])) check("falsy（if existing: 走 else 分支）", not empty) check("(row or {}).get('x') 不炸", (empty or {}).get("x") is None) check("int((row or {}).get('max_seq') or 0) == 0", int((empty or {}).get("max_seq") or 0) == 0) # --- 5. 异常不抛断链（fail-soft：返回空容器） --- print("\n[6] 连接异常时 q_all/q_one 返回空容器而非抛异常") class BoomConn: def cursor(self): raise RuntimeError("connection lost") try: rr = A.q_all("SELECT 1", (), conn=BoomConn()) check("q_all 异常降级为空 list", isinstance(rr, list) and len(rr) == 0) ro = A.q_one("SELECT 1", (), conn=BoomConn()) check("q_one 异常降级为空 dict", isinstance(ro, dict) and not ro) except Exception as exc: check("q_all/q_one 不应抛异常", False, repr(exc)) # --- 6. module_dbname：失败不抛、不返回硬编码库名 --- print("\n[7] module_dbname(module) 失败不抛") v = A.module_dbname("pbl_runtime_ext", default=None) check("返回 str 或 None（不抛异常）", v is None or isinstance(v, str), str(v)) check("未解析时返回 default（不硬编码库名）", v is None or v == A.get_module_dbname("pbl_runtime_ext")) check("default 参数生效", A.module_dbname("__no_such_module__", default="D") in ("D",) or isinstance(A.module_dbname("__no_such_module__", default="D"), str)) # --- 7. db.get_sor_context 三级降级 --- print("\n[8] db.get_sor_context(env, module) 降级顺序") class FakeEnv: def __init__(self, injected=False): self._injected = injected if injected: self.get_sor_context = lambda e, m: ("INJECTED", m) check("env 注入 helper 时优先复用", D.get_sor_context(FakeEnv(True), "pbl_compiler") == ("INJECTED", "pbl_compiler")) try: D.get_sor_context(FakeEnv(False), None) check("无注入且无库名时应 fail-closed 抛 RuntimeError", False) except RuntimeError: check("无注入且无库名时 fail-closed 抛 RuntimeError", True) except Exception as exc: check("无注入且无库名时 fail-closed 抛 RuntimeError", False, repr(exc)) # --- 8. 下游模块导入闭包（原 Bug 的致命症状：包导入即 ImportError） --- print("\n[9] 下游模块导入闭包（pbl_runtime_ext / pbl_assessment / pbl_scense_ext）") for mod in ("pbl_runtime_ext", "pbl_runtime_ext.idempotency", "pbl_runtime_ext.rtx_db", "pbl_assessment", "pbl_scense_ext", "pbl_compiler.exporter"): try: __import__(mod) check("import %s 成功" % mod, True) except ImportError as exc: # 仅当失败原因是 pbl_common 契约符号缺失时才算本 Bug 复发 bad = "pbl_common" in str(exc) and any( s in str(exc) for s in ("q_all", "q_one", "module_dbname", "get_sor_context", "crud")) check("import %s 无 pbl_common 契约符号缺失" % mod, not bad, str(exc)) if not bad: print(" (基座依赖缺失导致的 ImportError，非本 Bug 范畴: %s)" % exc) except Exception as exc: check("import %s 无 pbl_common 契约符号缺失" % mod, True) print(" (非 ImportError: %r)" % (exc,)) print("\n" + "=" * 68) print("自测结果: PASS=%d FAIL=%d" % (len(PASS), len(FAIL))) if FAIL: print("失败项: %s" % ", ".join(FAIL)) print("=" * 68) return 0 if not FAIL else 1 if __name__ == "__main__": sys.exit(main())