From 30b92eafda5f9d392a33eee80022d3e01a3830d6 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Sun, 20 Sep 2026 10:11:04 +0800 Subject: [PATCH] =?UTF-8?q?[M11b-1c-B]=20=E7=94=9F=E6=88=90=E5=99=A8?= =?UTF-8?q?=E5=81=A5=E5=A3=AE=E6=80=A7=E4=BF=AE=E5=A4=8D=E5=85=A5=E5=BA=93?= =?UTF-8?q?=EF=BC=9A=E8=A1=A5=E4=B8=81=E5=BA=94=E7=94=A8=E5=99=A8=20+=20?= =?UTF-8?q?=E8=BF=9E=E5=BA=93=E4=BB=BF=E7=9C=9F=20harness=20=E8=BF=81?= =?UTF-8?q?=E5=85=A5=20scripts/=EF=BC=88%=20=E8=BD=AC=E4=B9=89=20P1~P10=20?= =?UTF-8?q?=E4=B8=8E=20PBL=5FDDL=5FDB=5FHOST/PORT=20=E6=B2=99=E7=AE=B1?= =?UTF-8?q?=E5=BA=93=E5=BC=80=E5=85=B3=E5=8F=96=E8=AF=81=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=EF=BC=8C=E4=B8=9A=E5=8A=A1=E8=84=9A=E6=9C=AC=E4=B8=8E=20json?= =?UTF-8?q?=20=E5=A5=91=E7=BA=A6=E9=9B=B6=E6=94=B9=E5=8A=A8=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/m11b1cb_patch_generator.py | 357 +++++++++++++++++++++++++++ scripts/m11b1cb_verify_sandbox_db.py | 290 ++++++++++++++++++++++ 2 files changed, 647 insertions(+) create mode 100644 scripts/m11b1cb_patch_generator.py create mode 100644 scripts/m11b1cb_verify_sandbox_db.py diff --git a/scripts/m11b1cb_patch_generator.py b/scripts/m11b1cb_patch_generator.py new file mode 100644 index 0000000..19dd4ad --- /dev/null +++ b/scripts/m11b1cb_patch_generator.py @@ -0,0 +1,357 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M11b-1c-B T-1 补丁应用器：对同目录 scripts/pbl_runtime_event_ddl.py +做**精确串替换**（每处断言命中且仅命中 1 次），修复 QC #1 的 % 转义缺陷并重引入 +PBL_DDL_DB_HOST/PBL_DDL_DB_PORT 沙箱库开关（凭据仍只来自 env/*.json）。 + +不改 models/*.json、不改任何其他 .py。 + +用法： + python3 modules/pbl_runtime_ext/scripts/m11b1cb_patch_generator.py + # 应用到模块仓库工作树（默认目标=同目录 pbl_runtime_event_ddl.py） + python3 .../m11b1cb_patch_generator.py --check + # 只试跑，报告每处命中次数，不落盘 + python3 .../m11b1cb_patch_generator.py --target /tmp/base.py + # 重放到指定文件副本（取证用） + +幂等（QC #2 要求）：重复运行不会产出重复函数定义——检测到修复已全部在位则 SKIP 并返回 0； +检测到「只应用了一半」的部分状态则 FAIL 并返回 1（要求先还原到基线再重放）。 + +落点（QC #11 整改）：本脚本从工作空间根 tools/ 迁入模块仓库 +modules/pbl_runtime_ext/scripts/（与它服务的 pbl_runtime_event_ddl.py、以及已入库的 +m11b1cb_verify_sandbox_db.py 同仓同目录），纳入 git 版本控制，可被部署/复现阶段 git pull 取到。 +工作空间根 tools/m11b1cb_patch_generator.py 为迁移前的旧副本，自本次提交起**作废**， +唯一权威版本是本文件；旧副本仅作历史留痕，不再维护（清理动作见 dev-notes 第 5 节）。 + +取证方式（QC #8 整改：基线引用改为固定 commit + blob，不再用易漂移的 HEAD~1）： + 基线 = modules/pbl_runtime_ext 仓库 commit 4a7c7b9，其 scripts/pbl_runtime_event_ddl.py + 的 blob = 7b9c991（由 M11b-1c-A 还原任务确立）。本补丁所有 old 串以该基线为准。 + + git -C modules/pbl_runtime_ext show 4a7c7b9:scripts/pbl_runtime_event_ddl.py > /tmp/base.py + python3 modules/pbl_runtime_ext/scripts/m11b1cb_patch_generator.py --target /tmp/base.py + diff /tmp/base.py modules/pbl_runtime_ext/scripts/pbl_runtime_event_ddl.py + + ⚠ 预期：上面最后一条 diff **非空属正常，不代表补丁有问题**。仓库 HEAD 在本补丁（T-1） + 之外还叠加了 M11b-1c 其它子任务的改动——具体三处： + (a) _with_delimiter 改为按行扫描重写； + (b) _norm_part 额外多一个 .strip("'")（单引号剥离）； + (c) 离线自检 need_map 新增段。 + 这三处不在本任务授权范围内，故「REPL 里的 new 串与仓库最终代码逐字一致」这一 stronger + 断言在当前仓库历史下**不成立**，本文件不再声称它成立（旧版 docstring 的该声明已被 QC 实测证伪）。 + + 本补丁可机械核验的断言收敛为以下三条（都不依赖 HEAD 的其它改动，可复现）： + (1) 基线重放：对 commit 4a7c7b9 的副本执行，10/10 处 REPL 各命中且仅命中 1 次，rc=0； + (2) 幂等：对已应用结果（即仓库 HEAD 版本）重跑 → SKIP，rc=0； + (3) 部分应用：缺任一 APPLIED_MARKERS 标记 → FAIL，rc=1（QC #8 记录的 + `git show 7f0f560:...`（=HEAD~1）重放即触发此分支，rc=1，属守卫按设计生效）。 +""" +import argparse +import io +import os +import sys + +# 迁移后（QC #11）：默认目标与本脚本同目录，不再依赖工作空间根的相对层级 +DEFAULT_TARGET = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "pbl_runtime_event_ddl.py") + +# 修复已在位 / 部分在位 的判定标记（幂等守卫用） +APPLIED_MARKERS = ( + "def _norm_part(name):", + "def _render(sql, args=None, placeholders=None):", + '"PBL_DDL_DB_HOST"', + "part_norm = set(_norm_part(n) for n in part_names)", +) + +# ---------------------------------------------------------------- 替换清单 +REPL = [] + +# --- P1: _db_conf 重新实现「仅覆盖 host/port」的沙箱库开关（基线 @@ -531/@@ -539 区段） --- +REPL.append(( +'''def _db_conf(): + """读项目唯一事实源 env/test.json（禁止硬编码连接串）。""" + for env_name in ("test", "prod"): + path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name) + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + conf = json.load(f) + db = conf.get("db") or {} + if db.get("host"): + return env_name, db + return None, None +''', +'''def _db_conf(): + """读项目唯一事实源 env/test.json（禁止硬编码连接串）。 + + 沙箱库开关（harness，QC #1 配套的验证隔离能力）：环境变量 PBL_DDL_DB_HOST / + PBL_DDL_DB_PORT **只覆盖 host/port**，供 CI / 验证 harness 把连库分支指向一次性沙箱库 + ——破坏性判定（DROP PARTITION、探针 INSERT、guard break-glass）不能拿共享 test/prod 库做实验。 + · user / password / dbname 一律仍只来自 env/*.json：本函数不接受任何环境变量覆盖， + 也不内置默认主机、默认端口、默认账号（禁止硬编码）； + · 两个变量都不设置时，返回值与引入该开关之前逐字一致（零行为变化）； + · 覆盖生效时 env_name 追加 "+harness" 后缀，日志与 --verify-db 输出可一眼看出连的是沙箱库。 + """ + for env_name in ("test", "prod"): + path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name) + if os.path.exists(path): + with open(path, encoding="utf-8") as f: + conf = json.load(f) + db = conf.get("db") or {} + if db.get("host"): + host = (os.environ.get("PBL_DDL_DB_HOST") or "").strip() + port = (os.environ.get("PBL_DDL_DB_PORT") or "").strip() + if host: + db = dict(db, host=host) + env_name = "%s+harness" % env_name + if port: + if not port.isdigit(): + raise RuntimeError("PBL_DDL_DB_PORT 必须是数字端口号: %r" % port) + db = dict(db, port=int(port)) + if "+harness" not in env_name: + env_name = "%s+harness" % env_name + return env_name, db + return None, None +''')) + +# --- P2: _explain_partitions docstring 去掉 row[3] 字样（A5 静态断言）+ 入参转义自检 --- +REPL.append(( +''' 不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL + 版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。 + 取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。 + """ + try: + cur.execute("EXPLAIN FORMAT=JSON " + sql, tuple(params or ())) + row = cur.fetchone() +''', +''' 不再用旧的 PARTITIONS 扩展语法（EXPLAIN 后跟 PARTITIONS 关键字那种）：那种写法把分区列固定在 + 结果集第 4 列（下标 3），列序随 MariaDB/MySQL 版本变化，且该扩展语法在新版 MySQL 已废弃。 + JSON 计划里的 partitions 字段两版一致。 + 取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。 + """ + # QC #1(转义)：传进来的 sql 必须已完成标识符渲染，只剩 DBAPI 占位符 %s；残留 '%%' 说明 + # 调用方转义写错，直接抛出让它响，别丢给驱动报 "unsupported format character"。 + if "%%" in sql: + raise RuntimeError("EXPLAIN SQL 仍残留 '%%'，转义写法有误: " + sql[:200]) + args = tuple(params) if params else None + if args is not None and sql.count("%s") != len(args): + raise RuntimeError("EXPLAIN SQL 占位符 %d 个与参数 %d 个不符: %s" + % (sql.count("%s"), len(args), sql[:200])) + try: + cur.execute("EXPLAIN FORMAT=JSON " + sql, args) + row = cur.fetchone() +''')) + +# --- P3: _explain_partitions 结果分区名统一口径 --- +REPL.append(( +''' if key == "partitions" and isinstance(val, list): + for p in val: + if isinstance(p, str) and p and p not in found: + found.append(p) +''', +''' if key == "partitions" and isinstance(val, list): + for p in val: + name = _norm_part(p) + if name and name not in found: + found.append(name) +''')) + +# --- P4: 新增 _norm_part / _render 两个私有工具（插在 _explain_partitions 之前） --- +REPL.append(( +'''def _explain_partitions(cur, sql, params=None): +''', +'''def _norm_part(name): + """分区名统一口径（QC #1 附带缺陷：两处来源比对方式不一致）。 + + information_schema.PARTITIONS 的 PARTITION_NAME 与 EXPLAIN FORMAT=JSON 计划里的 + partitions 元素，跨引擎/版本可能带反引号、双引号、空白或大小写差异。两边都过这个 + 归一化再比对，避免「分区确实存在但判定说不属于清单」的假失败。 + """ + if not name: + return "" + text = name.decode("utf-8", "replace") if isinstance(name, (bytes, bytearray)) else str(name) + return text.strip().strip("`").strip('"').lower() + + +def _render(sql, args=None, placeholders=None): + """SQL 的 % 转义统一出口：可选渲染标识符 + fail-fast 自检（QC #1 的根因处置）。 + + 约定：模板中由 Python 侧填入的标识符（表名/分区名）写 %s；要留给 DBAPI 绑参的 + 百分号占位符一律写 %%s，标识符渲染后即变回 %s。 + · args 不为 None：先执行 sql % args 完成标识符渲染； + · args 为 None：视为调用方已完成渲染（模板就地 % 过），本函数只做转义自检。 + 自检两条，任一不满足立即抛 RuntimeError（而不是丢给驱动报 + "unsupported format character"，那会在 verify_db 里被 except 吞成一条模糊失败）： + · 结果仍残留 '%%' → 转义写错； + · '%s' 个数与 placeholders 不符 → 漏写/多写绑参占位符。 + 渲染单独成行、括号显式定界，不再依赖「相邻字面量先拼接、再整体 %」的隐式优先级。 + """ + if args is not None: + sql = sql % tuple(args) + if "%%" in sql: + raise RuntimeError("SQL 渲染后仍残留 '%%'，转义写法有误: " + sql[:200]) + if placeholders is not None and sql.count("%s") != placeholders: + raise RuntimeError("SQL 渲染后 DBAPI 占位符应为 %d 个，实际 %d 个: %s" + % (placeholders, sql.count("%s"), sql[:200])) + return sql + + +def _explain_partitions(cur, sql, params=None): +''')) + +# --- P5: _row_in_partition 转义重构（基线 @@ -641 区段） --- +REPL.append(( +''' try: + cur.execute("SELECT id FROM `%s` PARTITION (`%s`)" + " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, part), (uid,)) + except Exception: # noqa: BLE001 + return False +''', +''' sql = _render("SELECT id FROM `%s` PARTITION (`%s`)" + " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, _norm_part(part)), + placeholders=1) + try: + cur.execute(sql, (uid,)) + except Exception: # noqa: BLE001 + return False +''')) + +# --- P6: _locate_row_partition 复用 _row_in_partition + 转义/口径统一（基线 @@ -660 区段） --- +REPL.append(( +''' for part in part_names: + if _row_in_partition(cur, part, uid): + return part + planned = _explain_partitions(cur, + "SELECT id FROM `%s` WHERE tenant_id='t_probe'" + " AND event_uid=%%s AND created_at=%%s" % TABLE, + (uid, ts)) + if len(planned) == 1: + return planned[0] + return ",".join(planned) or "?" +''', +''' # 1) 逐个 PARTITION(p) 物理反查（复用 _row_in_partition，SQL 只在其内拼装一次， + # 避免同一份转义逻辑在两处各写一遍、改一处漏一处） + for part in part_names: + if _row_in_partition(cur, part, uid): + return _norm_part(part) + # 2) 反查不中才退回 EXPLAIN FORMAT=JSON；标识符渲染与转义自检统一走 _render + probe_sql = _render("SELECT id FROM `%s` WHERE tenant_id='t_probe'" + " AND event_uid=%%s AND created_at=%%s" % TABLE, + placeholders=2) + planned = _explain_partitions(cur, probe_sql, (uid, ts)) + if len(planned) == 1: + return planned[0] + return ",".join(planned) or "?" +''')) + +# --- P7: verify_db 探针 INSERT 转义自检（基线 @@ -694 区段） --- +REPL.append(( +''' try: + for i, (uid, ts) in enumerate(probes): + cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" + " VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE, (uid, i + 1, ts)) + except Exception as exc: # noqa: BLE001 +''', +''' ins_sql = _render("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" + " VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE, + placeholders=3) + try: + for i, (uid, ts) in enumerate(probes): + cur.execute(ins_sql, (uid, i + 1, ts)) + except Exception as exc: # noqa: BLE001 +''')) + +# --- P8: verify_db 分区清单归一化口径（供 P9/P10 比对） --- +REPL.append(( +''' results.append(("MAXVALUE 兜底", PARTITION_MAXVALUE in part_names, ",".join(part_names))) +''', +''' results.append(("MAXVALUE 兜底", PARTITION_MAXVALUE in part_names, ",".join(part_names))) + # 分区名比对统一走归一化口径（information_schema 与 EXPLAIN JSON 计划两来源一致） + part_norm = set(_norm_part(n) for n in part_names) +''')) + +# --- P9: 跨月落不同分区判定用统一口径 --- +REPL.append(( +''' located = [_locate_row_partition(cur, uid, ts, part_names) for uid, ts in probes] + distinct = len(set(located)) == 2 + in_list = all(p in part_names for p in located) +''', +''' located = [_locate_row_partition(cur, uid, ts, part_names) for uid, ts in probes] + distinct = len(set(located)) == 2 + in_list = all(_norm_part(x) in part_norm + for p in located for x in str(p).split(",")) +''')) + +# --- P10: 分区裁剪判定：SQL 渲染单独成行 + 统一口径（基线 @@ -710 区段） --- +REPL.append(( +''' # 旧写法（PARTITIONS 扩展语法 + 取 row[3]）：列序跨版本不稳定，新版 MySQL 已废弃该语法。 + pruned = _explain_partitions(cur, + "SELECT id FROM `%s` WHERE created_at>='2026-01-01'" + " AND created_at<'2026-02-01'" % TABLE) + results.append(("分区裁剪", + bool(pruned) and all(p in part_names for p in pruned) + and len(pruned) < max(1, len(part_names)), +''', +''' # 旧写法（PARTITIONS 扩展语法 + 取结果集下标 3 那一列）：列序跨版本不稳定，新版 MySQL 已废弃。 + prune_sql = _render("SELECT id FROM `%s` WHERE created_at>='2026-01-01'" + " AND created_at<'2026-02-01'" % TABLE, + placeholders=0) + pruned = _explain_partitions(cur, prune_sql) + results.append(("分区裁剪", + bool(pruned) and all(p in part_norm for p in pruned) + and len(pruned) < max(1, len(part_names)), +''')) + + +def _already_applied(src): + """幂等守卫：返回 (全已在位, 在位标记列表, 缺失标记列表)。""" + hit = [m for m in APPLIED_MARKERS if m in src] + return len(hit) == len(APPLIED_MARKERS), hit, [m for m in APPLIED_MARKERS if m not in src] + + +def main(): + ap = argparse.ArgumentParser(description="M11b-1c-B T-1 生成器补丁应用器") + ap.add_argument("--target", default=DEFAULT_TARGET, help="待打补丁的脚本路径") + ap.add_argument("--check", action="store_true", help="只试跑并报告命中次数，不落盘") + opts = ap.parse_args() + + with io.open(opts.target, encoding="utf-8") as f: + src = f.read() + orig = src + + # ---- 幂等守卫（QC #2）：修复已在位就 SKIP，绝不重复插入函数定义 ---- + done, hit, missing = _already_applied(src) + if done: + print("SKIP 补丁已全部在位（%d/%d 标记命中），无需重复应用: %s" + % (len(hit), len(APPLIED_MARKERS), opts.target)) + return 0 + if hit: + print("FAIL 目标处于「部分应用」状态（%d 个标记已在位，%d 个缺失），" + "请先还原到基线 4a7c7b9(blob 7b9c991) 再重放。缺失: %s" + % (len(hit), len(missing), " / ".join(m[:40] for m in missing))) + return 1 + + # ---- 逐处替换：断言命中且仅命中 1 次，任一处不符立即中止且**不落盘** ---- + for idx, (old, new) in enumerate(REPL, 1): + n = src.count(old) + if n != 1: + print("FAIL P%d 命中 %d 次（要求恰好 1 次），中止且不写盘: %s" + % (idx, n, opts.target)) + return 1 + src = src.replace(old, new, 1) + print("OK P%d 命中 1 次" % idx) + + print("HIT %d/%d 处替换全部命中且唯一" % (len(REPL), len(REPL))) + + if opts.check: + print("CHECK --check 模式：只试跑，未写盘") + return 0 + if src == orig: + print("FAIL 替换后内容与原文件一致，疑似未生效，拒绝写盘") + return 1 + with io.open(opts.target, "w", encoding="utf-8") as f: + f.write(src) + print("WRITE 已写盘: %s" % opts.target) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/m11b1cb_verify_sandbox_db.py b/scripts/m11b1cb_verify_sandbox_db.py new file mode 100644 index 0000000..4dc16e3 --- /dev/null +++ b/scripts/m11b1cb_verify_sandbox_db.py @@ -0,0 +1,290 @@ +#!/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())