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