charging-cabinet/test/reset_fixture.py

"""夹具复位（一次性脚本，不属于交付物）：让 e2e 从干净状态开始。

1. 清掉测试柜的 Redis 实时键（`device:860000000000001*`）
2. 清掉安全码下发标记（Redis `auth_str:issued:{imei}` + DB `cabinets.auth_str_issued = 0`），
   使 ① `auth_str` 能重新下发 8 位安全码（否则平台回 `auth_str already issued`）
3. 结束该柜未结束的充电订单（`active_flag` 是生成列，由 `end_time` 派生，不能直接写），
   避免 ⑤ `on` 撞上 `active_flag` 唯一键
"""
import os
import sys

import pymysql
import redis

IMEI = '860000000000001'
CABINET_ID = 1

# 默认指向 SSH 隧道（tools/dev-tunnel.ps1），不再使用已移除的 ZeroTier/nginx 转发
REDIS_HOST = os.environ.get('PMS_REDIS_HOST', '127.0.0.1')
REDIS_PORT = int(os.environ.get('PMS_REDIS_PORT', '16379'))
MYSQL_HOST = os.environ.get('PMS_MYSQL_HOST', '127.0.0.1')
MYSQL_PORT = int(os.environ.get('PMS_MYSQL_PORT', '13306'))

# 连不上时给可读提示并 exit 2（隧道没起 ≠ 复位成功，绝不能静默继续）
HINT = ('连接被拒：请先运行 `pwsh tools/dev-tunnel.ps1` 建立 SSH 隧道'
        '（或设置 PMS_MYSQL_HOST/PMS_MYSQL_PORT / PMS_REDIS_HOST/PMS_REDIS_PORT 指向其它可达地址）')

r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT,
                password='Hbhyg731024@', db=6, decode_responses=True)
try:
    keys = r.keys(f'device:{IMEI}*') + [f'auth_str:issued:{IMEI}']
except redis.RedisError as e:
    print(f'✗ Redis 连接失败 {REDIS_HOST}:{REDIS_PORT}：{e}', file=sys.stderr)
    print(HINT, file=sys.stderr)
    sys.exit(2)
print('Redis 待删除 key 数:', len(keys))
if keys:
    print('Redis deleted:', r.delete(*keys))

try:
    c = pymysql.connect(host=MYSQL_HOST, port=MYSQL_PORT, user='root', password='Hbhyg731024@',
                        database='pms_dev', autocommit=True)
except pymysql.MySQLError as e:
    print(f'✗ MySQL 连接失败 {MYSQL_HOST}:{MYSQL_PORT}：{e}', file=sys.stderr)
    print(HINT, file=sys.stderr)
    sys.exit(2)
cur = c.cursor()
cur.execute("UPDATE cabinets SET auth_str_issued = 0 WHERE imei = %s", (IMEI,))
print('auth_str_issued 置 0 行数:', cur.rowcount)
cur.execute(
    "UPDATE charge_orders SET end_time = NOW(), status = 2 "
    "WHERE cabinet_id = %s AND end_time IS NULL",
    (CABINET_ID,),
)
print('结束未完成订单行数:', cur.rowcount)
print('剩余 device 键:', r.keys(f'device:{IMEI}*'))