#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M11b-1c-B 验收仿真（harness）：不依赖真实数据库，用「pymysql 风格 % 绑定」的假游标把 同目录 scripts/pbl_runtime_event_ddl.py 的连库路径（_db_conf / _connect / verify_db / _explain_partitions / _row_in_partition / _locate_row_partition）整条跑一遍。 覆盖验收点： V1 PBL_DDL_DB_HOST / PBL_DDL_DB_PORT 开关生效，且 user/password/dbname 仍只来自 env/*.json V2 两个变量都不设时行为与基线逐字一致（env_name 无 +harness、host/port 原值） V3 开关只覆盖 host/port：PBL_DDL_DB_USER/PASSWORD/DB 之类环境变量一律无效（不读） V4 verify_db 全链路：任何一条发往驱动前的 SQL 都不残留 '%%'，且 %s 个数与参数个数严格相符 （这正是 pymysql 会抛 "unsupported format character" 的 QC #1 崩溃点） V5 分区名口径统一：EXPLAIN JSON 里带反引号/大写/空白的分区名，仍能与 information_schema 清单匹配（旧实现用 `in part_names` 直比会假失败） V6 _render / _explain_partitions 的 fail-fast 自检能抓到写错的转义（防回归） V7 T-2：tests/test_m11b1a_ddl_runtime.py 的连库分支（163-164 行设开关 → gen.verify_db()） 确实经 _db_conf 读到沙箱 host/port 退出码 0 = 全 PASS。 【落点（QC #11 整改）】本脚本与 m11b1cb_patch_generator.py 一并迁入模块仓库 modules/pbl_runtime_ext/scripts/，纳入 git 版本控制。工作空间根 tools/ 下的同名旧副本自本次 提交起作废。路径解析改为「向上寻找含 projects/pbls/env/test.json 的机构工作空间根」， 因此在模块仓库内、在工作空间根下调用都能正确定位（不再依赖脚本所在层级的固定上跳数）。 """ import importlib.util import json import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) # 模块仓库根（scripts/ 的上一级）：tests/ 相对它定位 MODULE_DIR = os.path.dirname(HERE) def _find_workspace(start): """向上找机构工作空间根（判据：存在 projects/pbls/env/test.json）。""" d = os.path.abspath(start) for _ in range(8): if os.path.exists(os.path.join(d, "projects", "pbls", "env", "test.json")): return d parent = os.path.dirname(d) if parent == d: break d = parent # 兜底：按旧约定（scripts → module → modules → 工作空间根）上跳三级 return os.path.dirname(os.path.dirname(MODULE_DIR)) WS = _find_workspace(MODULE_DIR) SCRIPT = os.path.join(HERE, "pbl_runtime_event_ddl.py") ENV_JSON = os.path.join(WS, "projects", "pbls", "env", "test.json") RESULTS = [] def chk(name, passed, detail=""): RESULTS.append((name, bool(passed))) print(" %-4s %-52s %s" % ("PASS" if passed else "FAIL", name, detail)) return bool(passed) def load_gen(): spec = importlib.util.spec_from_file_location("pbl_runtime_event_ddl_sim", SCRIPT) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod # ------------------------------------------------------------ 假 DBAPI 层 class BindError(Exception): """模拟 pymysql/MySQLdb 在参数绑定阶段对 % 的处理失败。""" class FakeCursor: """严格复现 MySQLdb 系驱动行为：有 args 时执行 sql % args， 未转义的 % 后面跟非 s 字符 → ValueError: unsupported format character。""" def __init__(self, conn): self.conn = conn self._rows = [] @staticmethod def _lit(fragment): """取 PARTITION (`p202609`) 括号内的标识符并归一化（与 _norm_part 同口径）。""" text = fragment.split(")", 1)[0] return text.strip().strip("`").strip("'").strip('"').lower() def execute(self, sql, args=None): self.conn.sent.append(sql) if args is not None: try: rendered = sql % tuple(args) except (ValueError, TypeError) as exc: raise BindError("%s | sql=%r | args=%r" % (exc, sql, args)) if "%s" in rendered or "%%" in rendered: raise BindError("绑定后仍残留占位符: %r" % rendered) elif "%%" in sql: raise BindError("无参语句残留 '%%': %r" % sql) up = sql.upper() # 1) 探针 INSERT：按 created_at 记录该 uid 真实落入的分区（RANGE 语义） if up.startswith("INSERT INTO"): uid, _seq, ts = args self.conn.uid_part[uid] = "p202609" if str(ts) < "2026-10-01" else "pmax" self._rows = [] return True # 2) PARTITION(p) 物理反查：只有落点对得上才返回行 if "PARTITION (" in up: part = self._lit(sql.split("PARTITION (", 1)[1]) uid = args[0] self._rows = [(7,)] if self.conn.uid_part.get(uid) == part else [] return True # 3) guard break-glass 上下文 if up.startswith("SET @PBL_GUARD_CTX"): self.conn.guard_off = "NULL" not in sql.upper() self._rows = [] return True # 4) UPDATE / DELETE：护栏生效时被触发器拒绝。 # 错误文本按真实 MariaDB 触发器 SIGNAL SQLSTATE '45000' + # MESSAGE_TEXT='PBL_APPEND_ONLY_DENIED*' 的驱动回显构造（1644/45000 双含）， # 与被测代码的判定口径（APPEND_ONLY / 45000 / 4500）一致，不迁就实现放水。 if up.startswith("UPDATE") or up.startswith("DELETE FROM"): if self.conn.guard_off: self._rows = [] return True raise RuntimeError("(1644) Got error 1644 from handler, SQLSTATE (45000): " "PBL_APPEND_ONLY_DENIED") # 5) 元数据与 EXPLAIN if up.startswith("EXPLAIN FORMAT=JSON"): self._rows = [(json.dumps(self.conn.plan),)] return True if "PARTITION_NAME, PARTITION_DESCRIPTION" in sql: self._rows = [(n, "20270301") for n in self.conn.parts] return True if "COUNT(*) FROM INFORMATION_SCHEMA.PARTITIONS" in up: self._rows = [(len(self.conn.parts),)] return True self._rows = [] return True def fetchone(self): return self._rows[0] if self._rows else None def fetchall(self): return list(self._rows) def close(self): pass class FakeConn: def __init__(self): self.sent = [] self.parts = ["p202609", "p202610", "p202611", "p202612", "p202701", "p202702", "pmax"] # 故意让 EXPLAIN JSON 计划里的分区名带反引号+大写 → 考验统一口径（V5）： # 旧实现 `p in part_names` 直比会假失败，归一化后应命中清单内的 p202609 self.plan = {"query_block": {"table": {"partitions": ["`P202609`"]}}} self.guard_off = False self.uid_part = {} def cursor(self): return FakeCursor(self) def close(self): pass def main(): print("harness: %s" % SCRIPT) print("ws : %s" % WS) print("env : %s" % ENV_JSON) base = json.load(open(ENV_JSON, encoding="utf-8"))["db"] # ------------------------------------------------ V1/V2/V3 沙箱库开关 for k in ("PBL_DDL_DB_HOST", "PBL_DDL_DB_PORT"): os.environ.pop(k, None) os.environ["PBL_DDL_DB_USER"] = "hacker_user" os.environ["PBL_DDL_DB_NAME"] = "hacker_db" os.environ["PBL_DDL_DB_PASSWORD"] = "hacker_pw" gen = load_gen() env_name, db = gen._db_conf() chk("V2 未设开关时 env_name/db 与基线一致", env_name == "test" and db["host"] == base["host"] and int(db.get("port") or 3306) == int(base.get("port") or 3306), "env=%s host=%s port=%s" % (env_name, db.get("host"), db.get("port"))) os.environ["PBL_DDL_DB_HOST"] = "127.0.0.1" os.environ["PBL_DDL_DB_PORT"] = "13306" env_name, db = gen._db_conf() chk("V1 开关覆盖 host/port 生效", db["host"] == "127.0.0.1" and db["port"] == 13306, "host=%s port=%r" % (db["host"], db["port"])) chk("V1 env_name 标记 +harness（可一眼看出连沙箱库）", env_name.endswith("+harness"), env_name) chk("V1 凭据仍只来自 env/*.json：dbname 未被覆盖", db.get("dbname") == base.get("dbname"), "dbname=%s" % db.get("dbname")) chk("V3 user/password 未被任何环境变量污染", db.get("user") == base.get("user") and db.get("password") == base.get("password"), "user=%s（与 env/test.json 同）" % db.get("user")) # 只设 PORT 也要标记 harness del os.environ["PBL_DDL_DB_HOST"] env_name, db = gen._db_conf() chk("V1 仅设 PORT 时同样生效并标记 harness", db["port"] == 13306 and env_name.endswith("+harness") and db["host"] == base["host"], "env=%s port=%s" % (env_name, db["port"])) os.environ["PBL_DDL_DB_HOST"] = "127.0.0.1" # 非数字端口必须 fail-fast，不静默回落 os.environ["PBL_DDL_DB_PORT"] = "abc" try: gen._db_conf() chk("V1 非数字 PBL_DDL_DB_PORT fail-fast", False, "未抛错") except RuntimeError as exc: chk("V1 非数字 PBL_DDL_DB_PORT fail-fast", "PBL_DDL_DB_PORT" in str(exc), str(exc)[:60]) os.environ["PBL_DDL_DB_PORT"] = "13306" # ------------------------------------------------ V4/V5 verify_db 全链路 conn = FakeConn() gen._connect = lambda: (conn, "test+harness", db) rc = gen.verify_db() chk("V4 verify_db() 走通假驱动 rc=0", rc == 0, "rc=%s，发出 %d 条 SQL" % (rc, len(conn.sent))) bad = [s for s in conn.sent if "%%" in s] chk("V4 无一条 SQL 残留 '%%'（QC#1 崩溃点）", not bad, "残留 %d 条" % len(bad)) ins = [s for s in conn.sent if s.upper().startswith("INSERT INTO")] chk("V4 探针 INSERT 渲染后恰 3 个 %s", ins and all(s.count("%s") == 3 for s in ins), "%d 条 / %s" % (len(ins), ins[0][:64] + "…" if ins else "-")) rowp = [s for s in conn.sent if "PARTITION (" in s] chk("V4 PARTITION(p) 反查语句渲染正确", rowp and all(s.count("%s") == 1 and "%%" not in s for s in rowp), "%d 条，示例: %s" % (len(rowp), rowp[0][:70] + "…" if rowp else "-")) exp = [s for s in conn.sent if s.upper().startswith("EXPLAIN FORMAT=JSON")] chk("V5 EXPLAIN 走 JSON 计划（无 PARTITIONS 扩展语法）", exp and all("EXPLAIN PARTITIONS " not in s for s in exp), "%d 条" % len(exp)) chk("V5 反引号/大写分区名仍与清单匹配（口径统一）", gen._norm_part("`P202601`") == "p202601" and gen._norm_part(" pmax ") == "pmax", "_norm_part OK") # 跨月探针必须真的落到两个不同分区（旧 harness 解析 PARTITION(p) 取串有误， # 会把两条都判到同一分区 → 假失败；此处按括号内标识符取串，与实现口径一致） located = [conn.uid_part[k] for k in sorted(conn.uid_part)] chk("V5 跨月探针落两个不同分区（物理落点仿真）", len(set(located)) == 2, "落点: %s" % " | ".join("%s→%s" % (k[-6:], v) for k, v in sorted(conn.uid_part.items()))) # ------------------------------------------------ V6 fail-fast 自检 try: gen._render("SELECT * FROM `%s` WHERE a=%%s" % gen.TABLE, placeholders=2) chk("V6 _render 占位符数不符时抛错", False, "未抛错") except RuntimeError as exc: chk("V6 _render 占位符数不符时抛错", "DBAPI 占位符" in str(exc), str(exc)[:70]) try: gen._render("SELECT '%%Y' FROM x", placeholders=0) chk("V6 _render 残留 '%%' 时抛错", False, "未抛错") except RuntimeError as exc: chk("V6 _render 残留 '%%' 时抛错", "转义写法有误" in str(exc), str(exc)[:70]) try: gen._explain_partitions(FakeCursor(FakeConn()), "SELECT '%%Y' FROM x") chk("V6 _explain_partitions 拒绝未转义 SQL", False, "未抛错") except RuntimeError as exc: chk("V6 _explain_partitions 拒绝未转义 SQL", "残留" in str(exc), str(exc)[:70]) # ------------------------------------------------ V7 T-2 测试连库分支链路 tpath = os.path.join(MODULE_DIR, "tests", "test_m11b1a_ddl_runtime.py") tsrc = open(tpath, encoding="utf-8").read() ln = next(i + 1 for i, line in enumerate(tsrc.splitlines()) if 'os.environ["PBL_DDL_DB_HOST"]' in line) chk("V7 测试仍在 163-164 行区段设置 PBL_DDL_DB_HOST/PORT", 'os.environ["PBL_DDL_DB_HOST"]' in tsrc and 'os.environ["PBL_DDL_DB_PORT"]' in tsrc and 160 <= ln <= 168, "PBL_DDL_DB_HOST 位于第 %d 行" % ln) chk("V7 生成器侧已提供该能力（_db_conf 读同名变量）", "PBL_DDL_DB_HOST" in open(SCRIPT, encoding="utf-8").read() and "PBL_DDL_DB_PORT" in open(SCRIPT, encoding="utf-8").read(), "OK") # 复现测试的调用序列：设开关 → 生成器 _connect 取库 os.environ["PBL_DDL_DB_HOST"] = "127.0.0.1" os.environ["PBL_DDL_DB_PORT"] = "13306" g2 = load_gen() env_name, db2 = g2._db_conf() chk("V7 测试设开关后生成器连到沙箱库（T-2 恢复）", db2["host"] == "127.0.0.1" and db2["port"] == 13306, "env=%s host=%s port=%s" % (env_name, db2["host"], db2["port"])) # 取库副作用处置：连库分支只连沙箱 host/port，dbname/凭据仍取自 env/test.json， # 即测试不会因环境变量而误连/误写其它库（共享 test/prod 库零写入） chk("V7 连库分支不改变 dbname/凭据（不产生跨库写副作用）", db2.get("dbname") == base.get("dbname") and db2.get("user") == base.get("user"), "dbname=%s user=%s" % (db2.get("dbname"), db2.get("user"))) n_fail = sum(1 for _, p in RESULTS if not p) print("\n[汇总] 共 %d 项，PASS %d，FAIL %d" % (len(RESULTS), len(RESULTS) - n_fail, n_fail)) print("SIM RESULT: %s" % ("ALL-PASS" if n_fail == 0 else "HAS-FAIL")) return 0 if n_fail == 0 else 1 if __name__ == "__main__": sys.exit(main())