#!/usr/bin/env python3 # -*- coding: utf-8 -*- """m11b-1a 运行时缺陷验证 harness（响应 QC #1 % 转义 / QC #6 verify_db 判定 / QC #5 落点）。 被验对象：scripts/pbl_runtime_event_ddl.py（唯一被改文件）。 两层： A. 离线层（无依赖，必跑） A1 maintenance_sql() 渲染不再抛 ValueError（QC #1 崩溃点） A2 渲染正文里 DATE_FORMAT 参数为**单 %** 形式（'%Y-%m-01' / '%Y%m'），且无 '%%Y' 残留 A3 append_only_sql()/full_script() 既有转义语义未改坏：'%(dbname)s'/'%(app_user)s' 被 replace 掉、host 通配符输出为单 % （'pbls'@'%'），无 '@'%%'' 残留 A4 DDL 未被改动：create_table_sql() 的 sha256 == 修复前基线（验收标准 4） A5 源码无 `EXPLAIN PARTITIONS ` 废弃语法残留、无「查询结果未使用」的空转 for 循环 A6 verify_db 判定改用 information_schema.PARTITIONS(按 ORDINAL 排序) + EXPLAIN FORMAT=JSON + PARTITION(p) 反查（静态断言三要素齐备） A7 CLI 冒烟：--verify / --print / --print-maintain / --emit 全部 rc=0 B. 实库层（仅当 M11B1A_LOCAL_DB=1，由 tests/run_m11b1a_localdb.sh 拉起一次性 MariaDB 后运行；**不会**连项目 test/prod 库，避免 DROP PARTITION 误伤真数据） B1 用 mariadb CLI(DELIMITER) 装载 full_script()，核对表/分区/过程/触发器齐备 B2 --verify-db 全项 PASS（分区数/兜底/跨月落不同分区/分区裁剪/UPDATE·DELETE 被拒/探针清理） B3 CALL pbl_runtime_event_ensure_partitions(9) 真能 REORGANIZE 出新月分区（A2 的单 % SQL 在 服务端可执行，不只是渲染正确） B4 CALL pbl_runtime_event_prune_partitions(2) 真能 DROP 旧月分区且保留 pmax 用法： python3 tests/test_m11b1a_ddl_runtime.py # 只跑 A 层 bash tests/run_m11b1a_localdb.sh # A + B 层（自带一次性库） 退出码：0 = 无 FAIL；1 = 有 FAIL；2 = B 层被跳过且 A 层通过时仍返回 0（B 缺失会显式打印 SKIP 行）。 """ import hashlib import importlib.util import os import re import subprocess import sys HERE = os.path.dirname(os.path.abspath(__file__)) MODULE_DIR = os.path.dirname(HERE) # modules/pbl_runtime_ext SCRIPT = os.path.join(MODULE_DIR, "scripts", "pbl_runtime_event_ddl.py") WORKSPACE = os.path.dirname(os.path.dirname(MODULE_DIR)) # 机构工作空间根 ENV_JSON = os.path.join(WORKSPACE, "projects", "pbls", "env", "test.json") # 修复前（git 049e09a）`--print` 输出的 sha256 —— DDL 未改动的机械证据（验收标准 4） DDL_BASELINE_SHA256 = "30a83c8a9d729db987f82de1a5bc30229dc4a7b1cc9881c7a5a45673b1a62619" RESULTS = [] def chk(name, passed, detail=""): RESULTS.append((name, bool(passed), detail)) print(" %-4s %-46s %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", SCRIPT) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod # ------------------------------------------------------------------ A. 离线层 def layer_a(): print("[A] 离线层（% 转义 + 判定逻辑静态核验）") src = open(SCRIPT, encoding="utf-8").read() # A1/A2 —— QC #1 崩溃点 try: maint = subprocess.run([sys.executable, SCRIPT, "--print-maintain"], capture_output=True, text=True) rc_maint = maint.returncode body = maint.stdout except Exception as exc: # noqa: BLE001 rc_maint, body = 99, "" chk("A1 maintenance_sql 渲染不抛异常", False, str(exc)) chk("A1 maintenance_sql 渲染不抛异常(QC#1)", rc_maint == 0, "rc=%s, %d bytes（修复前 rc=1 ValueError: unsupported format character 'Y'）" % (rc_maint, len(body))) for token in ("pbl_runtime_event_ensure_partitions", "pbl_runtime_event_prune_partitions", "ev_pbl_runtime_event_partition_maintain"): chk("A2 维护段含完整对象 %s" % token, token in body, "%d 处" % body.count(token)) chk("A2 DATE_FORMAT 参数为单 %% 形式('%Y-%m-01')", "DATE_FORMAT(CURDATE(), '%Y-%m-01')" in body and "DATE_FORMAT(month_start, '%Y%m')" in body, "命中 %d 处单 %% 形式" % (body.count("'%Y-%m-01'") + body.count("'%Y%m'"))) chk("A2 无 '%%Y' 双写转义残留", "%%Y" not in body and "%%'" not in body, "grep -c '%%Y' = %d" % body.count("%%Y")) # A3 —— append_only 既有转义语义 full = subprocess.run([sys.executable, SCRIPT, "--print-maintain"], capture_output=True, text=True) assert full.returncode == 0 script_txt = subprocess.run([sys.executable, "-c", "import sys;sys.path.insert(0,%r);" "import importlib.util as u;" "s=u.spec_from_file_location('m',%r);m=u.module_from_spec(s);s.loader.exec_module(m);" "print(m.full_script(),end='')" % (MODULE_DIR, SCRIPT)], capture_output=True, text=True) chk("A3 full_script() rc=0", script_txt.returncode == 0, "rc=%s" % script_txt.returncode) fs = script_txt.stdout chk("A3 %(dbname)s/%(app_user)s 已被替换", "%(dbname)s" not in fs and "%(app_user)s" not in fs, "残留 %d 处" % (fs.count("%(dbname)s") + fs.count("%(app_user)s"))) chk("A3 REVOKE 主机通配符为单 %%（'pbls'@'%'）", "FROM 'pbls'@'%'" in fs and "@'%%'" not in fs, "命中 'pbls'@'%%' = %d，'@%%' = %d" % (fs.count("FROM 'pbls'@'%'"), fs.count("@'%%'"))) chk("A3 break-glass ctx 原样输出", "append_only_admin" in fs, "%d 处" % fs.count("append_only_admin")) # A4 —— DDL 逐字未动 gen = load_gen() digest = hashlib.sha256((gen.create_table_sql() + "\n").encode("utf-8")).hexdigest() chk("A4 create_table_sql sha256 == 修复前基线", digest == DDL_BASELINE_SHA256, digest[:16] + "…") # A5/A6 —— 判定逻辑（QC #6） chk("A5 源码无 'EXPLAIN PARTITIONS ' 废弃语法", "EXPLAIN PARTITIONS " not in src, "grep 命中 %d" % src.count("EXPLAIN PARTITIONS ")) chk("A5 无未使用查询结果的空转 for 循环", not re.search(r"for\s+k\s+in\s*\(", src) and "row[3]" not in src, "row[3] 命中 %d，for k in 命中 %d" % (src.count("row[3]"), len(re.findall(r"for\s+k\s+in\s*\(", src)))) vsrc = src[src.index("def verify_db("):] chk("A6 分区清单来自 information_schema.PARTITIONS+ORDINAL", "information_schema.PARTITIONS" in vsrc and "PARTITION_ORDINAL_POSITION" in vsrc, "OK") chk("A6 落分区判定用 PARTITION(p) 反查 + EXPLAIN FORMAT=JSON", "_row_in_partition" in src and "PARTITION (`%s`)" in src and "EXPLAIN FORMAT=JSON" in src and "_locate_row_partition" in vsrc, "OK") chk("A6 探针清理 break-glass 逻辑保留", "SET @pbl_guard_ctx=" in vsrc and "GUARD_ADMIN_CTX" in vsrc and "SET @pbl_guard_ctx=NULL" in vsrc, "guard_ctx 设置 %d 处 / 复位 %d 处" % (vsrc.count("SET @pbl_guard_ctx="), vsrc.count("SET @pbl_guard_ctx=NULL"))) # A7 —— CLI 冒烟 for args in (["--verify"], ["--print"], ["--print-guard"], ["--print-maintain"]): r = subprocess.run([sys.executable, SCRIPT] + args, capture_output=True, text=True) tail = (r.stdout.strip().splitlines() or [""])[-1][:40] chk("A7 CLI %s rc=0" % " ".join(args), r.returncode == 0, "末行: %s" % tail) tmp_emit = os.path.join(HERE, "_emit_check") r = subprocess.run([sys.executable, SCRIPT, "--emit", tmp_emit, "--dbname", "pbls"], capture_output=True, text=True) files = sorted(os.listdir(tmp_emit)) if os.path.isdir(tmp_emit) else [] chk("A7 CLI --emit rc=0", r.returncode == 0 and len(files) >= 3, "%d 文件: %s" % (len(files), ",".join(files))) # 生成的维护 SQL 同样必须是单 % 形式（emit 路径回归） m3 = os.path.join(tmp_emit, "20260919_m11b_03_pbl_runtime_event_partition_maintenance.sql") if os.path.exists(m3): t = open(m3, encoding="utf-8").read() chk("A7 --emit 维护段 DATE_FORMAT 单 %", "DATE_FORMAT(CURDATE(), '%Y-%m-01')" in t and "%%Y" not in t, "%d bytes" % len(t)) subprocess.run(["rm", "-rf", tmp_emit]) return gen # ------------------------------------------------------------------ B. 实库层 def layer_b(gen): print("\n[B] 实库层（一次性本地 MariaDB，不触碰项目 test/prod 库）") if os.environ.get("M11B1A_LOCAL_DB") != "1": print(" SKIP B 层：未设 M11B1A_LOCAL_DB=1（用 bash tests/run_m11b1a_localdb.sh 跑全量）") return True import json import pymysql conf = json.load(open(ENV_JSON, encoding="utf-8"))["db"] db = dict(conf) db["host"] = os.environ.get("M11B1A_DB_HOST", "127.0.0.1") db["port"] = int(os.environ.get("M11B1A_DB_PORT", "13306")) os.environ["PBL_DDL_DB_HOST"] = db["host"] os.environ["PBL_DDL_DB_PORT"] = str(db["port"]) conn0 = pymysql.connect(host=db["host"], port=db["port"], user=db["user"], password=db["password"], database=db["dbname"], charset="utf8mb4", autocommit=True) cur0 = conn0.cursor() cur0.execute("SELECT COUNT(*) FROM information_schema.PARTITIONS " "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s " "AND PARTITION_NAME IS NOT NULL", (gen.TABLE,)) n0 = cur0.fetchone()[0] cur0.execute("SELECT COUNT(*) FROM information_schema.ROUTINES " "WHERE ROUTINE_SCHEMA=DATABASE() AND ROUTINE_NAME LIKE 'pbl_runtime_event%%'") n_proc = cur0.fetchone()[0] cur0.execute("SELECT COUNT(*) FROM information_schema.TRIGGERS " "WHERE EVENT_OBJECT_TABLE=%s", (gen.TABLE,)) n_trg = cur0.fetchone()[0] cur0.close(); conn0.close() chk("B1 CLI(mariadb+DELIMITER) 装载 full_script() 后对象齐备", n0 >= gen.MONTHS_PRECREATE + 1 and n_proc >= 2 and n_trg >= 2, "分区 %d 个 / 维护过程 %d 个 / 触发器 %d 个" % (n0, n_proc, n_trg)) rc2 = gen.verify_db() chk("B2 verify_db() rc=0 全项 PASS(QC#6)", rc2 == 0, "verify-db rc=%s" % rc2) conn = pymysql.connect(host=db["host"], port=db["port"], user=db["user"], password=db["password"], database=db["dbname"], charset="utf8mb4", autocommit=True) cur = conn.cursor() def parts(): cur.execute("SELECT PARTITION_NAME FROM information_schema.PARTITIONS " "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s " "AND PARTITION_NAME IS NOT NULL ORDER BY PARTITION_ORDINAL_POSITION", (gen.TABLE,)) return [r[0] for r in cur.fetchall()] before = parts() cur.execute("CALL pbl_runtime_event_ensure_partitions(9)") after = parts() new = [p for p in after if p not in before] chk("B3 CALL ensure_partitions(9) 补建出新月分区", len(new) >= 3 and after.count("pmax") == 1, "新增 %s（分区数 %d→%d）" % (",".join(new), len(before), len(after))) cur.execute("CALL pbl_runtime_event_prune_partitions(2)") pruned = parts() dropped = [p for p in after if p not in pruned] chk("B4 CALL prune_partitions(2) DROP 旧月分区且保留 pmax", len(dropped) >= 1 and "pmax" in pruned, "DROP %s（剩余 %d 个）" % (",".join(dropped), len(pruned))) # 分区裁剪 + 落分区判定的服务端可执行性（EXPLAIN FORMAT=JSON 解析） planned = gen._explain_partitions(cur, "SELECT id FROM `%s` WHERE created_at>='2026-01-01'" " AND created_at<'2026-02-01'" % gen.TABLE) chk("B5 _explain_partitions 解析出分区清单", bool(planned) and len(planned) < len(pruned), "命中 %s（表内共 %d 分区）" % (",".join(planned), len(pruned))) cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" " VALUES ('t_harness',%%s,'probe',0,%%s,%%s)" % gen.TABLE, ("h_a", 1, "2026-01-05 00:00:00")) located = gen._locate_row_partition(cur, "h_a", "2026-01-05 00:00:00", pruned) chk("B6 _locate_row_partition 定位到 information_schema 清单内的分区", located in pruned, "h_a -> %s（清单 %s）" % (located, ",".join(pruned))) cur.execute("SET @pbl_guard_ctx='%s'" % gen.GUARD_ADMIN_CTX) cur.execute("DELETE FROM `%s` WHERE tenant_id='t_harness'" % gen.TABLE) cur.execute("SET @pbl_guard_ctx=NULL") cur.close(); conn.close() return True def main(): print("harness: %s" % SCRIPT) gen = layer_a() ok_b = layer_b(gen) fails = [n for n, p, _ in RESULTS if not p] print("\n[汇总] 共 %d 项，PASS %d，FAIL %d%s" % (len(RESULTS), len(RESULTS) - len(fails), len(fails), "" if ok_b else "（B 层 SKIP）")) if fails: print("FAIL 明细: %s" % " | ".join(fails)) print("HARNESS FAIL") return 1 print("HARNESS PASS") return 0 if __name__ == "__main__": sys.exit(main())