pbl_runtime_ext/tests/test_m11b1a_ddl_runtime.py

#!/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])
    # A8 —— PBL-RTX-0005 根因形态回归门禁（过程体非法 RETURN + DELIMITER 包裹完整性）
    # 缺陷本体：C 段 pbl_assert_append_only() 内 `IF ... THEN RETURN; END IF;`。MariaDB 只允许
    # FUNCTION 使用 RETURN，且 RETURN 后的分号会让 mariadb CLI 在语句中间提前切分 -> 实库装载
    # ERROR 1064 (42000) at line 65，连带 `CALL pbl_runtime_event_ensure_partitions(9)` 报 1305
    # PROCEDURE does not exist。修复口径：等价改写为「正向条件 + 单语句 SIGNAL SQLSTATE 45000」，
    # 并由渲染器统一以 DELIMITER $$ ... $$ ... DELIMITER ; 包裹过程段。下面 5 条断言锁死这两点。
    guard_sql = gen.append_only_sql()
    full_sql = gen.full_script()
    ret_bad = gen._procedure_body_return_errors(guard_sql, "append_only")
    chk("A8 append_only_sql 过程体无非法 RETURN（PBL-RTX-0005 根因）",
        ret_bad == [],
        "检出 %d 条: %s" % (len(ret_bad), ret_bad[:2] or "无"))
    naked = re.findall(r"(?im)^\s*RETURN\b", full_sql)
    chk("A8 渲染产物全文无裸 RETURN 语句", len(naked) == 0,
        "grep -i '^\\s*RETURN' 命中 %d 处" % len(naked))
    chk("A8 append-only 守卫未弱化：正向条件 + 单语句 SIGNAL + 双触发器均在位",
        guard_sql.count("SIGNAL SQLSTATE '45000'") >= 3 and "IF p_table" in guard_sql
        and "BEFORE UPDATE" in guard_sql and "BEFORE DELETE" in guard_sql,
        "SIGNAL %d 处 / IF p_table %d 处" % (guard_sql.count("SIGNAL SQLSTATE '45000'"),
                                             guard_sql.count("IF p_table")))
    delim_errs = gen._delimiter_integrity_errors(full_sql, "full_script")
    chk("A8 过程段 DELIMITER 配对完整（_delimiter_integrity_errors 零错）",
        delim_errs == [] and "DELIMITER $$" in full_sql and "DELIMITER ;" in full_sql,
        "%s" % (delim_errs[:2] or "DELIMITER $$ / DELIMITER ; 配对 OK"))
    # 门禁有效性自检：把守卫还原成缺陷形态（插入一条 RETURN;）必须被同一套判定抓出来，
    # 否则说明上面的断言是「永远为真的假门禁」。
    bad_sql = guard_sql.replace("SIGNAL SQLSTATE '45000'",
                                "RETURN;\n        SIGNAL SQLSTATE '45000'", 1)
    bad_hits = gen._procedure_body_return_errors(bad_sql, "bad_sample")
    chk("A8 自检：坏样本(还原 RETURN;)被 _procedure_body_return_errors 检出",
        bool(bad_hits), "检出 %d 条: %s" % (len(bad_hits), bad_hits[:1]))

    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)))

    # ---- B4 前置：造出一个「已过保留期」的历史月分区，prune 才有真实可 DROP 的对象 ----
    # PBL-RTX-0005 说明：本任务只修 DDL 装载 ERROR 1064，**不改** prune_partitions 的保留期语义
    # （仍以 CURDATE 为锚：只 DROP 早于「当月 - p_keep_months」的整月分区，绝不碰未来预建分区与
    #  pmax，避免误删数据）。一次性沙箱库里的分区全是当月/未来月，故先用 REORGANIZE 从首个
    #  月分区里拆出一个 2020-01 的历史月分区（空分区、零数据损失），再断言它被 DROP。
    cur.execute("SELECT PARTITION_NAME, PARTITION_DESCRIPTION FROM information_schema.PARTITIONS"
                " WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s AND PARTITION_NAME<>%s"
                " ORDER BY PARTITION_ORDINAL_POSITION LIMIT 1", (gen.TABLE, "pmax"))
    first_name, first_upper = cur.fetchone()
    rc_setup = 0
    try:
        cur.execute("ALTER TABLE `%s` REORGANIZE PARTITION `%s` INTO ("
                    "PARTITION `p202001` VALUES LESS THAN (DATE '2020-02-01'), "
                    "PARTITION `%s` VALUES LESS THAN (DATE '%s'))"
                    % (gen.TABLE, first_name, first_name, str(first_upper).strip("'")))
    except Exception as exc:          # noqa: BLE001
        rc_setup = 1
        print("  [B4-setup] REORGANIZE 失败: %s" % str(exc)[:160])
    chk("B4-setup REORGANIZE 拆出过期历史月分区 p202001（保留期语义不变）",
        rc_setup == 0 and "p202001" in parts(),
        "分区清单含 p202001=%s" % ("p202001" in parts()))

    # REORGANIZE 会改写分区清单，prune 的比对基线必须在 setup 之后重新采集，
    # 否则 dropped 恒为空（PBL-RTX-0005 修好装载后才执行到本段，属 harness 场景缺陷，
    # 未改动任何断言阈值：仍要求 dropped >= 1 且保留 pmax）。
    after = parts()
    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,
                                        tenant="t_harness")
    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())