"""端点层验收：造临时数据 → 调接口 → 断言 200 且能返回行 → 清理临时数据 用法： python .tmp-schema-align/endpoint_verify.py dev # 测试环境（pms_dev + test.anzhizhichong.com） python .tmp-schema-align/endpoint_verify.py prod # 生产环境（pms + app.anzhizhichong.com） python .tmp-schema-align/endpoint_verify.py dev clean # 只清理 """ import base64, hashlib, hmac, json, ssl, sys, time, urllib.request, urllib.error, pymysql ENVS = { "dev": dict( db="pms_dev", base="https://test.anzhizhichong.com/api/pms", secret="32e8708743c42fe764aff1d40737a63803a46815a9501f4bcb50c41fe3d27108", admin_phone="19900000001", admin_role=2, ), "prod": dict( db="pms", base="https://app.anzhizhichong.com/api/pms", secret="787789e8eb1f3a10e81596f8219a20aa3c34102acacba2a5ed14bbb1d16ec833", admin_phone="13912391234", admin_role=2, # 生产库**不新建账号**：直接复用既有的总管理员账号（id=2）签 token， # 只造 charge_orders / device_logs 两条临时业务行，验完立刻删除。 existing_admin_id=2, ), } TAG = "VERIFY-SCHEMA-TMP" def b64(b): return base64.urlsafe_b64encode(b).rstrip(b"=").decode() def sign(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 call(base, path, token): 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=25, context=ctx) as r: return r.status, r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "replace") def main(): env = ENVS[sys.argv[1]] only_clean = len(sys.argv) > 2 and sys.argv[2] == "clean" c = pymysql.connect(host="10.8.0.252", user="root", password="Hbhyg731024@", database=env["db"], autocommit=True) cur = c.cursor() # ---- 清理上一轮遗留 ---- creates_admin = "existing_admin_id" not in env cur.execute("DELETE FROM device_logs WHERE content LIKE %s", (f"%{TAG}%",)) cur.execute("DELETE FROM charge_orders WHERE battery_id = %s", (TAG,)) if creates_admin: # 只删自己造的临时账号；复用既有账号时**绝不能**碰 users 表 cur.execute("DELETE FROM users WHERE phone = %s", (env["admin_phone"],)) if only_clean: print("已清理临时数据（device_logs/charge_orders/users）") return # ---- 取一个真实拓扑：柜子 + 仓体（满足外键）---- 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" ) row = cur.fetchone() if not row: sys.exit("库里没有 compartments，无法造订单（外键需要）") compartment_id, cabinet_id = row # ---- 造临时超管（JWT 只带 sub，角色与权限从库里读）---- if creates_admin: cur.execute( "INSERT INTO users (phone, name, password, role, organization_id, status) VALUES (%s, %s, 'x', %s, NULL, 1)", (env["admin_phone"], TAG, env["admin_role"]), ) admin_id = cur.lastrowid else: admin_id = env["existing_admin_id"] # ---- 造 2 条 charge_orders（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 # ---- 造 2 条 device_logs ---- cur.execute( "INSERT INTO device_logs (cabinet_id, log_type, content) VALUES (%s, 1, %s)", (cabinet_id, json.dumps({"tag": TAG, "n": 1})), ) log_id = cur.lastrowid print(f"环境={sys.argv[1]} 库={env['db']} 临时数据: users.id={admin_id} " f"charge_orders.id={order_id} device_logs.id={log_id} (cabinet_id={cabinet_id})") # ---- 自签 token 调接口 ---- token = sign(env["secret"], admin_id, env["admin_phone"]) results = {} for path in ("/charge-orders?page=1&page_size=20", "/device-logs?page=1&page_size=20", "/charge-records?page=1&page_size=5"): status, body = call(env["base"], path, token) results[path] = (status, body) has_tmp = TAG in body print(f"\n--- GET {path} -> HTTP {status} (临时行出现={has_tmp}) ---") print(body[:600]) # ---- 断言 ---- ok_orders = results["/charge-orders?page=1&page_size=20"][0] == 200 and TAG in results["/charge-orders?page=1&page_size=20"][1] ok_logs = results["/device-logs?page=1&page_size=20"][0] == 200 and TAG in results["/device-logs?page=1&page_size=20"][1] ok_records = results["/charge-records?page=1&page_size=5"][0] == 200 print("\n==== 断言 ====") print(" /api/charge-orders 200 且返回临时行:", "PASS" if ok_orders else "FAIL") print(" /api/device-logs 200 且返回临时行:", "PASS" if ok_logs else "FAIL") print(" /api/charge-records 200 :", "PASS" if ok_records else "FAIL") print("\n==== 清理临时数据 ====") cur.execute("DELETE FROM device_logs WHERE content LIKE %s", (f"%{TAG}%",)) print(" device_logs 删除行数:", cur.rowcount) cur.execute("DELETE FROM charge_orders WHERE battery_id = %s", (TAG,)) print(" charge_orders 删除行数:", cur.rowcount) if creates_admin: cur.execute("DELETE FROM users WHERE phone = %s", (env["admin_phone"],)) print(" users 删除行数:", cur.rowcount) else: print(" users: 未新建账号（复用既有超管 id=%s），无需删除" % admin_id) cur.execute("SELECT COUNT(*) FROM charge_orders WHERE battery_id=%s", (TAG,)) print(" 残留 charge_orders:", cur.fetchone()[0]) cur.execute("SELECT COUNT(*) FROM device_logs WHERE content LIKE %s", (f"%{TAG}%",)) print(" 残留 device_logs:", cur.fetchone()[0]) c.close() sys.exit(0 if (ok_orders and ok_logs and ok_records) else 1) main()