AiDOPWarehouse/ai-dop-platform/tools/_verify_notify_tpl_cfg.py

# -*- coding: utf-8 -*-
"""P4-16 扩展：通知模板 / 通知渠道配置 冒烟验证（DB 侧）。

仅做结构与种子数据可用性检查，不依赖前后端 HTTP 联调：
- T1：ApprovalFlowNotifyTemplate / ApprovalFlowNotifyConfig 两表存在，列齐。
- T2：后端启动过至少一次后，系统预置模板应至少 9 条（NewTask / Urge / FlowCompleted / Transferred / Returned / AddSign / Withdrawn / Escalated / Timeout 的全局默认）。
- T3：通知配置菜单 SysMenu Id=1310300010106 存在且已授权（RoleMenu + TenantMenu 有记录）。
- T4：渲染兜底：模拟 BizType 级覆盖缺失，应能退回全局模板（通过 Python 端模拟 SELECT）。

使用方法：
    python _verify_notify_tpl_cfg.py
若 T2 失败（模板 0 条），说明后端尚未启动过（InitTables + EnsureSystemTemplates 未执行）。
"""
from __future__ import annotations
import pymysql
import sys

CONN_ARGS = dict(
    host='123.60.180.165', port=3306,
    user='aidopremote', password='1234567890aiDOP#',
    database='aidopdev', charset='utf8mb4', autocommit=True,
)

EXPECTED_TYPES = {"NewTask", "Urge", "FlowCompleted", "Transferred", "Returned", "AddSign", "Withdrawn", "Escalated", "Timeout"}


def ok(name: str, cond: bool, detail: str = "") -> bool:
    flag = "PASS" if cond else "FAIL"
    print(f"[{flag}] {name}" + (f" - {detail}" if detail else ""))
    return cond


def main() -> int:
    conn = pymysql.connect(**CONN_ARGS)
    passed = 0
    total = 0
    try:
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            # T1 表结构
            for t, expected_cols in [
                ("ApprovalFlowNotifyTemplate", {"NotifyType", "BizType", "Title", "Content", "IsEnabled", "IsSystem"}),
                ("ApprovalFlowNotifyConfig", {"ChannelKey", "Enabled", "WebhookUrl", "Secret", "TemplateId", "Remark"}),
            ]:
                total += 1
                cur.execute(f"SHOW TABLES LIKE '{t}'")
                has_tbl = bool(cur.fetchall())
                if not has_tbl:
                    ok(f"T1 表 {t} 存在", False, "表不存在")
                    continue
                cur.execute(f"DESC `{t}`")
                cols = {r["Field"] for r in cur.fetchall()}
                miss = expected_cols - cols
                if ok(f"T1 表 {t} 结构", not miss, f"缺字段: {miss}" if miss else f"列 {len(cols)} 个"):
                    passed += 1

            # T2 默认模板种子
            total += 1
            cur.execute(
                "SELECT NotifyType, COUNT(*) AS n FROM ApprovalFlowNotifyTemplate "
                "WHERE BizType='' AND IsSystem=1 GROUP BY NotifyType"
            )
            rows = cur.fetchall()
            seed_types = {r["NotifyType"] for r in rows}
            miss = EXPECTED_TYPES - seed_types
            if ok(
                "T2 系统预置模板（全局）覆盖 9 类通知",
                not miss,
                f"缺: {miss}" if miss else f"已预置 {len(seed_types)} 类（{len(rows)} 行）",
            ):
                passed += 1

            # T3 通知配置菜单
            total += 1
            cur.execute("SELECT Id, Title, Path FROM SysMenu WHERE Id=1310300010106")
            m = cur.fetchone()
            cur.execute("SELECT COUNT(*) AS n FROM SysRoleMenu WHERE MenuId=1310300010106")
            r = cur.fetchone()["n"]
            cur.execute("SELECT COUNT(*) AS n FROM SysTenantMenu WHERE MenuId=1310300010106")
            t_cnt = cur.fetchone()["n"]
            cond = bool(m) and r > 0 and t_cnt > 0
            detail = f"menu={bool(m)} RoleMenu={r} TenantMenu={t_cnt}"
            if ok("T3 通知配置菜单已挂载 + 授权", cond, detail):
                passed += 1

            # T4 渲染兜底（BizType 级不存在时回退全局）
            total += 1
            biz = "__VERIFY_NOT_EXIST_BIZ__"
            cur.execute(
                "SELECT Id FROM ApprovalFlowNotifyTemplate WHERE NotifyType='NewTask' AND BizType=%s",
                (biz,),
            )
            biz_hit = cur.fetchone()
            cur.execute(
                "SELECT Id FROM ApprovalFlowNotifyTemplate WHERE NotifyType='NewTask' AND BizType='' AND IsEnabled=1 LIMIT 1"
            )
            global_hit = cur.fetchone()
            fallback_ok = biz_hit is None and global_hit is not None
            if ok("T4 BizType 级缺失时可回退全局", fallback_ok):
                passed += 1

        print("\n==============================")
        print(f"结果：{passed}/{total} 通过")
        return 0 if passed == total else 1
    finally:
        conn.close()


if __name__ == "__main__":
    sys.exit(main())