charging-cabinet/test/tcp_latency_probe.py

"""测量端到端真实延迟：发一条 auth_str，看应答多久到达（最长等 75 秒）。

退出码（全仓统一口径）：0 = 三步都有应答；1 = 有步骤无应答/断言失败；
2 = 连接或环境问题（DB 连不上、TCP 建连失败）。
⛔ 以前它只打印 ✅/❌ 却永远 exit 0 —— 无应答也算"通过"，属假绿。
"""
import hashlib
import json
import os
import socket
import sys
import time
import pymysql

# 3001 是 device-server 的 TCP 业务端口（设备经公网/ZeroTier 直连），与数据库通路无关，保持不变
HOST, PORT = "10.8.0.252", 3001
IMEI = "860000000000001"

# 复位夹具，保证能重新申领
# 默认指向 SSH 隧道（tools/dev-tunnel.ps1），不再使用已移除的 ZeroTier/nginx 转发
try:
    conn = pymysql.connect(host=os.environ.get("PMS_MYSQL_HOST", "127.0.0.1"),
                           port=int(os.environ.get("PMS_MYSQL_PORT", "13306")),
                           user="root", password="Hbhyg731024@",
                           database="pms_dev", connect_timeout=8, autocommit=True)
except pymysql.MySQLError as e:
    print(f"✗ MySQL 连接失败：{e}", file=sys.stderr)
    print("连接被拒：请先运行 `pwsh tools/dev-tunnel.ps1` 建立 SSH 隧道", file=sys.stderr)
    sys.exit(2)
cur = conn.cursor()
cur.execute("UPDATE cabinets SET auth_str=NULL, auth_str_issued=0 WHERE imei=%s", (IMEI,))
cur.execute("DELETE FROM charge_records WHERE cabinet_id=1")
cur.execute("DELETE FROM charge_orders WHERE cabinet_id=1")
cur.execute("DELETE FROM device_logs WHERE cabinet_id=1")
cur.execute("DELETE FROM batteries")
cur.execute("UPDATE compartments SET status=0")
conn.close()
print("  夹具已复位")

try:
    s = socket.create_connection((HOST, PORT), timeout=5)
except OSError as e:
    print(f"✗ TCP 建连失败 {HOST}:{PORT}：{e}", file=sys.stderr)
    print("（该端口是 device-server 业务端口；测试环境为 :3001）", file=sys.stderr)
    sys.exit(2)
s.settimeout(1.0)
buf = b""
fails = []          # 收集失败项，决定退出码（非 0 = 失败）


def send(obj):
    line = json.dumps(obj, separators=(",", ":")) + "\n"
    s.sendall(line.encode())
    return time.time()


def wait_reply(t0, timeout=75):
    """等到收到一行报文，返回 (耗时, 报文) 或 (None, None)。"""
    global buf
    while time.time() - t0 < timeout:
        try:
            chunk = s.recv(65536)
        except socket.timeout:
            continue
        if not chunk:
            return None, None
        buf += chunk
        if b"\n" in buf:
            line, buf = buf.split(b"\n", 1)
            return time.time() - t0, json.loads(line.decode())
    return None, None


print("\n=== ① auth_str 延迟实测 ===")
t0 = send({"act": "auth_str", "msg_id": 1, "dev_id": IMEI})
dt, resp = wait_reply(t0)
if resp:
    print(f"  ✅ 应答耗时 {dt:.1f}s : {json.dumps(resp, ensure_ascii=False)}")
    auth_str = resp.get("auth_str")
    if not auth_str:
        fails.append("① auth_str 有应答但未带 auth_str")
else:
    print(f"  ❌ {dt if dt else '>75'}s 内无应答")
    auth_str = None
    fails.append(f"① auth_str 无应答（等待 {dt if dt else '>75'}s）")

if auth_str:
    print("\n=== ② login 延迟实测 ===")
    ts = int(time.time())
    sign = hashlib.sha256(f"{IMEI}{auth_str}{ts}".encode()).hexdigest()[56:64]
    t0 = send({"act": "login", "msg_id": 2, "dev_id": IMEI, "sign": sign, "timestamp": ts})
    dt, resp = wait_reply(t0)
    print(f"  {'✅' if resp else '❌'} 应答耗时 {dt:.1f}s : {json.dumps(resp, ensure_ascii=False) if resp else '无'}")
    if not resp:
        fails.append(f"② login 无应答（等待 {dt if dt else '>75'}s）")

    print("\n=== ③ status_post 延迟实测 ===")
    t0 = send({"act": "status_post", "msg_id": 3, "rssi": 85,
               "cabinet": {"status1": "00", "status2": "00", "temp": 29800, "meter": 125500},
               "boards": [{"board": 1, "channels": ["008200000000000000", "008200000000000000"]},
                          {"board": 2, "channels": ["008200000000000000", "008200000000000000"]}]})
    dt, resp = wait_reply(t0)
    print(f"  {'✅' if resp else '❌'} 应答耗时 {dt:.1f}s : {json.dumps(resp, ensure_ascii=False) if resp else '无'}")
    if not resp:
        fails.append(f"③ status_post 无应答（等待 {dt if dt else '>75'}s）")

s.close()

# ⛔ 退出码必须反映结果（口径：0=通过，1=断言失败，2=环境/连接问题）
if fails:
    print("\n未通过项（stderr）：", file=sys.stderr)
    for f in fails:
        print(f"  ❌ {f}", file=sys.stderr)
    sys.exit(1)
print("\n✅ 三步均有应答")
sys.exit(0)