248 lines
13 KiB
Python
248 lines
13 KiB
Python
#!/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())
|