charging-cabinet/test/schema_alignment_e2e.py

#!/usr/bin/env python3
"""PMS 时间列解码端到端验收（会造临时数据并自动清理）

它在验收什么
------------
sqlx 的 `chrono::NaiveDateTime` **只与 SQL 类型 `DATETIME` 兼容**（`TIMESTAMP` 只与
`DateTime<Utc>` 兼容），而本项目时间列在 Rust 侧全是 `NaiveDateTime`/`Option<NaiveDateTime>`
⇒ 任何一个 `TIMESTAMP` 列都会让对应接口 **500**：
`error occurred while decoding column "start_time": mismatched types; Rust type
 Option<NaiveDateTime> (as SQL type DATETIME) is not compatible with SQL type TIMESTAMP`

已实测的两次事故：测试环境充电记录导出必败；生产 `/api/charge-orders` 一有订单就 500。
本脚本用"造 1~2 条真实行 → 调接口 → 断言 200 且能返回行 → 清理"的方式，
把"列定义对齐了吗"从**看日志**升级成**看接口**。

为什么放在 test/ 而不是 tools/
------------------------------
`tools/` 放的是开发期只读工具（`db.py`、`db_schema_diff.py`），不写任何数据；
本脚本会往库里**写临时行**、需要**活的 API 与 token**，属于集成/端到端验收用例 ——
正是 `AGENTS.md` 里 `test/`（测试用例/报告）的定位。

用法
----
  # 测试环境（默认）：JWT_SECRET 取 /data/pms_dev/api-server/.env
  python test/schema_alignment_e2e.py --env dev --jwt-secret <JWT_SECRET>

  # 生产环境：必须显式加 --yes-prod；不新建账号，复用既有超管 id 签 token
  python test/schema_alignment_e2e.py --env prod --jwt-secret <JWT_SECRET> --yes-prod

  # 只清理上一次的残留
  python test/schema_alignment_e2e.py --env dev --clean-only

退出码：0=全部通过；1=有断言失败；2=参数/连接错误
"""

import argparse
import base64
import hashlib
import hmac
import json
import os
import ssl
import sys
import time
import urllib.error
import urllib.request

try:
    import pymysql
except ImportError:  # pragma: no cover
    print("缺少依赖：pip install pymysql", file=sys.stderr)
    sys.exit(2)

# ---- 环境参数 ---------------------------------------------------------------
ENVS = {
    "dev": {
        "db": "pms_dev",
        "base": "https://test.anzhizhichong.com/api/pms",
        "temp_phone": "19900000003",
    },
    "prod": {
        "db": "pms",
        "base": "https://app.anzhizhichong.com/api/pms",
        "temp_phone": None,          # 生产不新建账号
        "admin_user_id": 2,          # 复用既有总管理员账号签 token
        "admin_phone": "13912391234",
    },
}
DEFAULT_HOST = os.environ.get("PMS_MYSQL_HOST", "127.0.0.1")
DEFAULT_PORT = int(os.environ.get("PMS_MYSQL_PORT", "13306"))
DEFAULT_USER = "root"
DEFAULT_PASSWORD = "Hbhyg731024@"

# 临时数据标记（清理时按它精确定位，绝不误删业务数据）
TAG = "VERIFY-SCHEMA-TMP"
# 必须已是 DATETIME 的列（TIMPESTAMP 会让接口 500）
DATETIME_REQUIRED = [
    ("charge_records", "start_time"),
    ("charge_records", "end_time"),
    ("charge_orders", "start_time"),
    ("charge_orders", "end_time"),
    ("charge_orders", "created_at"),
    ("device_logs", "created_at"),
    ("users", "created_at"),
    ("users", "updated_at"),
    ("download_tasks", "completed_at"),
]


def die(message, code=2):
    print(message, file=sys.stderr)
    sys.exit(code)


# ---- 鉴权：按 API 的 Claims {sub, phone, exp, iat} 自签 HS256 ----------------
def _b64(raw):
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()


def sign_jwt(secret, sub, phone):
    now = int(time.time())
    seg = _b64(json.dumps({"alg": "HS256", "typ": "JWT"}, separators=(",", ":")).encode()) + "." + \
          _b64(json.dumps({"sub": sub, "phone": phone, "exp": now + 1800, "iat": now},
                          separators=(",", ":")).encode())
    return seg + "." + _b64(hmac.new(secret.encode(), seg.encode(), hashlib.sha256).digest())


def http_get(base, path, token):
    """带 Bearer token 的 GET；返回 (状态码, 响应体文本)"""
    req = urllib.request.Request(base + path, headers={"Authorization": "Bearer " + token})
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    try:
        with urllib.request.urlopen(req, timeout=30, context=ctx) as resp:
            return resp.status, resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace")


def cleanup(cur, phone):
    """删掉本脚本造的所有临时数据（幂等，可重复执行）"""
    cur.execute("DELETE FROM download_tasks WHERE file_name LIKE %s OR error_message LIKE %s",
                (f"%{TAG}%", f"%{TAG}%"))
    n_dl = cur.rowcount
    cur.execute("DELETE FROM device_logs WHERE content LIKE %s", (f"%{TAG}%",))
    n_log = cur.rowcount
    cur.execute("DELETE FROM charge_orders WHERE battery_id = %s", (TAG,))
    n_order = cur.rowcount
    n_user = 0
    if phone:
        cur.execute("DELETE FROM users WHERE phone = %s", (phone,))
        n_user = cur.rowcount
    return n_dl, n_log, n_order, n_user


def main():
    ap = argparse.ArgumentParser(
        description="PMS 时间列解码端到端验收（造临时行 → 调接口 → 断言 → 自动清理）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="退出码：0=全部通过；1=断言失败；2=参数/连接错误",
    )
    ap.add_argument("--env", choices=["dev", "prod"], default="dev", help="目标环境（默认 dev）")
    ap.add_argument("--jwt-secret", help="该环境 api-server 的 JWT_SECRET（或设环境变量 PMS_JWT_SECRET）")
    ap.add_argument("--api-base", help="覆盖 API 基址，如 http://127.0.0.1:3000/api")
    ap.add_argument("--admin-user-id", type=int, help="reuse 模式下的超管 user id（prod 默认 2）")
    ap.add_argument("--yes-prod", action="store_true", help="确认对生产库写临时数据（prod 必填）")
    ap.add_argument("--clean-only", action="store_true", help="只清理临时数据后退出")
    ap.add_argument("--host", default=DEFAULT_HOST)
    ap.add_argument("--port", type=int, default=DEFAULT_PORT)
    ap.add_argument("--user", default=DEFAULT_USER)
    ap.add_argument("--password", default=DEFAULT_PASSWORD)
    args = ap.parse_args()

    env = ENVS[args.env]
    base = args.api_base or env["base"]
    if args.env == "prod" and not args.yes_prod:
        die("对生产环境跑验收会写临时行：确认请加 --yes-prod（脚本会在结束时自动清理）")
    secret = args.jwt_secret or os.environ.get("PMS_JWT_SECRET")
    if not secret and not args.clean_only:
        die("缺少 --jwt-secret（或环境变量 PMS_JWT_SECRET）\n"
            "  取法：ssh 到服务器读 /data/pms_dev/api-server/.env 或 /data/pms/api-server/.env")

    try:
        conn = pymysql.connect(host=args.host, port=args.port, user=args.user,
                               password=args.password, database=env["db"],
                               charset="utf8mb4", autocommit=True)
    except pymysql.MySQLError as e:
        die("数据库连接失败：%s\n"
            "连接被拒：请先运行 `pwsh tools/dev-tunnel.ps1` 建立 SSH 隧道"
            "（或设置 PMS_MYSQL_HOST/PMS_MYSQL_PORT 指向其它可达地址）" % e)
    cur = conn.cursor()
    phone = env.get("temp_phone")

    if args.clean_only:
        print("清理临时数据：download_tasks/device_logs/charge_orders/users =", cleanup(cur, phone))
        return 0

    failures = []
    try:
        # ---- 0) 前置检查：这些列必须已经是 DATETIME ----
        print("== 0) 列类型前置检查（%s）==" % env["db"])
        bad_types = []
        for table, column in DATETIME_REQUIRED:
            cur.execute(
                "SELECT DATA_TYPE FROM information_schema.COLUMNS "
                "WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s AND COLUMN_NAME=%s",
                (env["db"], table, column),
            )
            row = cur.fetchone()
            got = row[0] if row else "<列不存在>"
            flag = "OK" if got == "datetime" else "FAIL"
            if got != "datetime":
                bad_types.append("%s.%s=%s" % (table, column, got))
            print("  [%s] %-30s = %s" % (flag, "%s.%s" % (table, column), got))
        if bad_types:
            failures.append("时间列仍是 TIMESTAMP/非 DATETIME：%s" % ", ".join(bad_types))

        # ---- 1) 造临时数据（外键需要的真实拓扑：柜子 + 仓体）----
        print("\n== 1) 造临时数据 ==")
        cur.execute("DELETE FROM download_tasks WHERE file_name LIKE %s", (f"%{TAG}%",))
        cur.execute("DELETE FROM device_logs WHERE content LIKE %s", (f"%{TAG}%",))
        cur.execute("DELETE FROM charge_orders WHERE battery_id = %s", (TAG,))
        cur.execute("SELECT c.id, cb.cabinet_id FROM compartments c "
                    "JOIN cabin_boards cb ON c.cabin_board_id = cb.id ORDER BY c.id LIMIT 1")
        topo = cur.fetchone()
        if not topo:
            die("库里没有 compartments/cabin_boards，无法造订单（外键需要）", 2)
        compartment_id, cabinet_id = topo

        if phone:
            cur.execute("DELETE FROM users WHERE phone = %s", (phone,))
            cur.execute("INSERT INTO users (phone,name,password,role,organization_id,status) "
                        "VALUES (%s,%s,'x',2,NULL,1)", (phone, TAG))
            admin_id, admin_phone = cur.lastrowid, phone
            print("  临时超管 id=%s（脚本结束时删除）" % admin_id)
        else:
            admin_id = args.admin_user_id or env["admin_user_id"]
            admin_phone = env["admin_phone"]
            print("  复用既有超管 id=%s（不新建账号）" % admin_id)

        # status<>0 ⇒ 生成列 active_flag=NULL，不会撞"同仓体唯一活跃单"
        cur.execute(
            "INSERT INTO charge_orders (compartment_id, cabinet_id, battery_id, start_time, end_time, "
            "start_soc, end_soc, energy, reason, status) "
            "VALUES (%s,%s,%s,NOW(),NOW(),850,1000,22,'normal',1)",
            (compartment_id, cabinet_id, TAG))
        order_id = cur.lastrowid
        cur.execute("INSERT INTO device_logs (cabinet_id, log_type, content) VALUES (%s,1,%s)",
                    (cabinet_id, json.dumps({"tag": TAG})))
        log_id = cur.lastrowid
        print("  charge_orders.id=%s device_logs.id=%s（cabinet_id=%s）" % (order_id, log_id, cabinet_id))

        token = sign_jwt(secret, admin_id, admin_phone)

        # ---- 2) 端点断言 ----
        print("\n== 2) 接口断言 ==")
        checks = [
            ("/charge-orders?page=1&page_size=20", TAG, "订单列表（NaiveDateTime 解码）"),
            ("/device-logs?page=1&page_size=20", TAG, "设备日志列表（created_at 解码）"),
            ("/charge-records?page=1&page_size=5", None, "充电记录列表"),
        ]
        for path, needle, desc in checks:
            status, body = http_get(base, path, token)
            ok = status == 200 and (needle is None or needle in body)
            print("  [%s] GET %-38s HTTP %s  %s" % ("OK" if ok else "FAIL", path.split("?")[0], status, desc))
            if not ok:
                failures.append("GET %s -> HTTP %s（期望 200%s）：%s"
                                % (path, status, "" if needle is None else " 且含临时行", body[:300]))

        # ---- 3) 导出链路（xlsx 生成）----
        print("\n== 3) 导出链路（GET /api/charge-records/export）==")
        status, body = http_get(base, "/charge-records/export", token)
        if status != 200:
            failures.append("创建导出任务失败：HTTP %s %s" % (status, body[:200]))
        else:
            task_id = json.loads(body)["data"]["task_id"]
            cur.execute("UPDATE download_tasks SET file_name=CONCAT(IFNULL(file_name,''), %s) WHERE id=%s",
                        (TAG, task_id))  # 打标记，便于清理
            deadline = time.time() + 60
            row = None
            while time.time() < deadline:
                cur.execute("SELECT status,progress,file_path,IFNULL(error_message,'') "
                            "FROM download_tasks WHERE id=%s", (task_id,))
                row = cur.fetchone()
                if row and row[0] in (2, 3):
                    break
                time.sleep(2)
            print("  download_tasks.id=%s -> status=%s progress=%s file=%s err=%s"
                  % (task_id, row[0], row[1], row[2], row[3]))
            if not row or row[0] != 2 or row[1] != 100 or not row[2]:
                failures.append("导出任务未成功（status/progress/file_path 见上）")
            else:
                print("  ⚠️ 导出文件在 API 服务器上：%s（如需删除请到服务器 exports/ 目录）" % row[2])
    finally:
        print("\n== 4) 清理临时数据 ==")
        n_dl, n_log, n_order, n_user = cleanup(cur, phone)
        print("  download_tasks=%d device_logs=%d charge_orders=%d users=%d" % (n_dl, n_log, n_order, n_user))
        conn.close()

    print("\n== 结论 ==")
    if failures:
        for f in failures:
            print("  ❌ %s" % f)
        return 1
    print("  ✅ 全部通过：时间列已是 DATETIME，订单/日志/充电记录/导出四个链路的解码不再 500")
    return 0


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