charging-cabinet/test/tcp_e2e_smoke.py

"""TCP 协议端到端联调（模拟柜控板）——针对测试环境 device-server。

流程（严格按 docs/protocol/柜控板固件对接文档.md v1.1）：
    auth_str → login → status_post → bat_in → on(上报) → status_post×3(积分) → off

每一步都打印平台的应答，便于人工核对协议实现。
"""
import sys
import hashlib
import json
import os
import socket
import time

HOST, PORT = "10.8.0.252", 3001
IMEI = "860000000000001"
BATTERY_ID = "SN-LEGACY-001"

# ⚠️ **平台侧积分电量的期望值推导**（与 status_post 的负载常量绑定）：
#   CHARGING = 54.00V / 8.00A ⇒ 电压 5400×10 = 54000 mV、电流 800×10 = 8000 mA
#   积分口径（software/api-server/src/workers/energy.rs）：
#       mWh = mV × mA × Δt / 3_600_000，Δt 固定 60 秒
#   本脚本第 ⑥ 步发 3 次 status_post ⇒ 期望 Wh = (3×7200 + 500) / 1000 = 22（mWh→Wh 四舍五入）
ENERGY_SAMPLES = 3
ENERGY_VOLTAGE_MV = 5400 * 10
ENERGY_CURRENT_MA = 800 * 10
ENERGY_DT_SECS = 60


def expected_energy_wh():
    """按积分公式算出本脚本应落库的订单电量（Wh）。"""
    per_sample_mwh = ENERGY_VOLTAGE_MV * ENERGY_CURRENT_MA * ENERGY_DT_SECS // 3_600_000
    total_mwh = ENERGY_SAMPLES * per_sample_mwh
    return (total_mwh + 500) // 1000


def hex_channel(batt: int, power: int, soc_pct: int, voltage_001v: int,
                current_001a: int, temp_k: int) -> str:
    """按协议 §5.3：byte0 电池状态 / byte1 电源状态 / byte2 SOC(1%) /
    byte3-4 电压(0.01V 大端) / byte5-6 电流(0.01A 大端) / byte7-8 温度(1K 大端)。"""
    return (f"{batt:02x}{power:02x}{soc_pct:02x}"
            f"{voltage_001v:04x}{current_001a:04x}{temp_k:04x}")


EMPTY = hex_channel(0x00, 0x82, 0, 0, 0, 0)          # 空仓：无电池但有电源
CHARGING = hex_channel(0x82, 0x82, 85, 5400, 800, 298)  # 充电中：54.00V / 8.00A / 298K
FULL = hex_channel(0x81, 0x02, 100, 5460, 0, 298)    # 充满

assert len(EMPTY) == 18 and len(CHARGING) == 18, "通道 hex 必须 18 字符"
print(f"  通道 hex：空仓={EMPTY}  充电中={CHARGING}  充满={FULL}")


class Cabinet:
    def __init__(self):
        self.sock = socket.create_connection((HOST, PORT), timeout=10)
        self.sock.settimeout(10)
        self.buf = b""
        self.msg_id = 0

    def send(self, payload: dict):
        line = json.dumps(payload, separators=(",", ":"), ensure_ascii=False) + "\n"
        print(f"  → {line.strip()}")
        self.sock.sendall(line.encode())

    def recv(self, timeout=8):
        """读一行平台报文（LF 分帧）。"""
        deadline = time.time() + timeout
        while b"\n" not in self.buf:
            if time.time() > deadline:
                return None
            try:
                chunk = self.sock.recv(65536)
            except socket.timeout:
                return None
            if not chunk:
                return None
            self.buf += chunk
        line, self.buf = self.buf.split(b"\n", 1)
        obj = json.loads(line.decode())
        print(f"  ← {json.dumps(obj, ensure_ascii=False)}")
        return obj

    def next_id(self):
        self.msg_id += 1
        return self.msg_id


def step(title):
    print(f"\n=== {title} ===")


def status_post(cab, channels_board1, channels_board2, msg_id):
    cab.send({
        "act": "status_post",
        "msg_id": msg_id,
        "rssi": 85,
        "cabinet": {"status1": "00", "status2": "00", "temp": 29800, "meter": 125500},
        "boards": [
            {"board": 1, "channels": channels_board1},
            {"board": 2, "channels": channels_board2},
        ],
    })
    return cab.recv()


cab = Cabinet()
ok = []

step("① auth_str（未登录可发）")
cab.send({"act": "auth_str", "msg_id": cab.next_id(), "dev_id": IMEI})
resp = cab.recv() or {}
# ⚠️ **限流要单独报**：协议 §5.1 规定同一 IMEI「5 次 / 5 分钟」。
# 连续多跑几次本脚本就会撞上 —— 此时后续每一步都会级联失败，
# 输出看起来像"登录/建单/结算全坏了"。必须与真正的功能缺陷区分开：
# 明确说清原因并以**退出码 2（环境/夹具问题）**结束，别让人去查不存在的 bug。
if resp.get("err") == "rate limit exceeded":
    print("  ⚠️ 未验证：auth_str 被频率限制（协议 §5.1：同一 IMEI 5 次/5 分钟）")
    print("     这是**重复运行本脚本**导致的，不是功能缺陷。")
    print("     处理：等 5 分钟再跑；或重启测试环境 api-server 清零进程内计数器")
    print("           （systemctl restart pms-api-test）；再跑前先 python test/reset_fixture.py")
    sys.exit(2)
auth_str = resp.get("auth_str")
ok.append(("auth_str 返回 8 位安全码", bool(auth_str) and len(auth_str) == 8))
print(f"  安全码 = {auth_str}")

step("② login（签名 = SHA256(dev_id+auth_str+timestamp)[56..64]）")
ts = int(time.time())
sign = hashlib.sha256(f"{IMEI}{auth_str}{ts}".encode()).hexdigest()[56:64]
cab.send({"act": "login", "msg_id": cab.next_id(), "dev_id": IMEI,
          "sign": sign, "timestamp": ts, "iccid": "8986012520983901841"})
resp = cab.recv() or {}
ok.append(("login suc=1", resp.get("suc") == 1))
ok.append(("login 应答带 spec（非旧 cell）", "spec" in resp and "cell" not in resp))
ok.append(("login 应答带 work_mode", "work_mode" in resp))
ok.append(("login 应答带 act（协议 §3.1）", resp.get("act") == "login"))

step("③ status_post（全空仓，应回 suc=1）")
resp = status_post(cab, [EMPTY, EMPTY], [EMPTY, EMPTY], cab.next_id()) or {}
ok.append(("status_post suc=1", resp.get("suc") == 1))

step("④ bat_in（电池插入 —— 只登记台账，不建单）")
cab.send({"act": "bat_in", "msg_id": cab.next_id(), "id": "01-1", "battery": {
    "battery_id": BATTERY_ID, "brand_code": "1", "model_code": "1", "material_code": "1",
    "cell_count": 15, "rated_voltage": 48000, "rated_capacity": 24000,
    "info": "80825515180320012a",
    "cells": "0e100e150e0e0e120e170e0b0e110e140e0f0e130e160e0d0e100e120e0e0e170e0b012a0128",
}})
resp = cab.recv() or {}
ok.append(("bat_in suc=1", resp.get("suc") == 1))

step("⑤ on（设备上报形态，不带 suc —— 建单唯一入口）")
cab.send({"act": "on", "msg_id": cab.next_id(), "id": "01-1"})
resp = cab.recv() or {}
ok.append(("on 上报 suc=1", resp.get("suc") == 1))

step("⑥ status_post ×3（充电中，验证平台侧积分电量）")
for _ in range(3):
    resp = status_post(cab, [CHARGING, EMPTY], [EMPTY, EMPTY], cab.next_id()) or {}
    ok.append(("status_post(充电中) suc=1", resp.get("suc") == 1))

step("⑦ off（充满停止 —— 结束订单并结算电量）")
cab.send({"act": "off", "msg_id": cab.next_id(), "id": "01-1",
          "battery": {"battery_id": BATTERY_ID, "info": FULL}})
resp = cab.recv() or {}
ok.append(("off suc=1", resp.get("suc") == 1))

step("⑧ 未登录发业务报文（应被丢弃、无任何应答）")
try:
    bad = Cabinet()
    bad.send({"act": "status_post", "msg_id": 1, "rssi": 50,
              "cabinet": {"status1": "00", "status2": "00", "temp": 0, "meter": 0},
              "boards": [{"board": 1, "channels": [EMPTY, EMPTY]}]})
    silent = bad.recv(timeout=4)
    ok.append(("未登录业务报文被静默丢弃", silent is None))
    bad.sock.close()
except Exception as e:  # noqa: BLE001
    ok.append(("未登录业务报文被静默丢弃", False))
    print(f"    异常: {e}")

cab.sock.close()

# ---------------------------------------------------------------------------
# ⑨ 平台侧积分电量：**端到端断言**（读测试库，而不是只看 suc=1）
# ---------------------------------------------------------------------------
unverified = []
step("⑨ 平台侧积分电量（读测试库：charge_orders.energy 应等于积分期望值）")


def fetch_settled_order(timeout_secs=15):
    """轮询测试库，返回 `(order_id, order_status, order_energy, compartment_id, record)`。

    `record` = 该仓体最新 `charge_records` 行 `(energy, max_temperature)`，无则 None。

    ⚠️ 列名以 information_schema 实测为准：**`max_temperature` 只在 `charge_records` 上**，
    `charge_orders` 没有该列（早期版本这里写错，被"未验证"机制挡下而非静默通过）。
    ⚠️ 结算在 report_worker 里是异步的（`off` 的 `suc` 可能在写回电量之前返回），
    所以这里**轮询**而不是读一次。连不上库时抛异常，由调用方判为"未验证"。
    """
    import pymysql

    host = os.environ.get("PMS_MYSQL_HOST", "127.0.0.1")
    port = int(os.environ.get("PMS_MYSQL_PORT", "13306"))
    db = os.environ.get("PMS_MYSQL_DB", "pms_dev")
    conn = pymysql.connect(host=host, port=port, user="root",
                           password="Hbhyg731024@", database=db, autocommit=True)
    try:
        cur = conn.cursor()
        deadline = time.time() + timeout_secs
        last = None
        while time.time() < deadline:
            cur.execute(
                "SELECT co.id, co.status, co.energy, co.compartment_id "
                "  FROM charge_orders co JOIN cabinets c ON c.id = co.cabinet_id "
                " WHERE c.imei = %s ORDER BY co.id DESC LIMIT 1",
                (IMEI,),
            )
            last = cur.fetchone()
            if last and last[1] != 0:  # status <> 0 ⇒ 已结算
                break
            time.sleep(0.5)
        if last is None:
            return None
        cur.execute(
            "SELECT energy, max_temperature FROM charge_records "
            " WHERE compartment_id = %s ORDER BY id DESC LIMIT 1",
            (last[3],),
        )
        return last + (cur.fetchone(),)
    finally:
        conn.close()


try:
    row = fetch_settled_order()
    if row is None:
        ok.append(("积分电量：存在订单", False))
    else:
        order_id, order_status, energy_wh, _comp_id, rec = row
        want = expected_energy_wh()
        rec_energy, rec_temp = rec if rec else (None, None)
        print(f"  订单 id={order_id} status={order_status} energy={energy_wh}Wh | "
              f"记录 energy={rec_energy}Wh 最高温={rec_temp}K×100 | 期望={want}Wh")
        ok.append(("积分电量：订单已结算", order_status != 0))
        ok.append((f"积分电量：charge_orders.energy == 积分期望值 {want}Wh（实测 {energy_wh}）",
                   energy_wh == want))
        ok.append(("积分电量：已生成 charge_records", rec is not None))
        # 同一积分值写两处，必须一致（两处不一致说明写回路径有分叉）
        ok.append((f"积分电量：charge_records.energy 与订单一致（记录 {rec_energy}）",
                   rec_energy == want))
        # 最高温由积分路径一并跟踪（CHARGING 的 temp = 298K ⇒ K×100 = 29800）
        ok.append((f"积分路径：最高温已跟踪（29800 K×100，实测 {rec_temp}）", rec_temp == 29800))
except Exception as e:  # noqa: BLE001
    print(f"  ⚠️ 未验证：连不上测试库（{e}）")
    print("     需要先起 SSH 隧道：pwsh tools/dev-tunnel.ps1"
          "（或用 PMS_MYSQL_HOST/PMS_MYSQL_PORT/PMS_MYSQL_DB 指向别处）")
    unverified.append("平台侧积分电量（连不上测试库 —— 见上方 ⚠️，通常是没有 SSH 隧道）")

print("\n========== 联调结果 ==========")
for name, passed in ok:
    print(f"  {'✅' if passed else '❌'} {name}")
print(f"\n  合计 {sum(1 for _, p in ok if p)}/{len(ok)} 通过")
# ⛔ 退出码必须反映结果：0=全部通过，1=有失败（此前本脚本永远 exit 0，属假绿）；
#    失败项额外打到 stderr，便于 CI/人只看 stderr 就能定位（口径见 AGENTS.md「验收流程」）。
failed = [n for n, p in ok if not p]
if failed:
    print("未通过项（stderr）：", file=sys.stderr)
    for n in failed:
        print(f"  ❌ {n}", file=sys.stderr)
    print(f"合计 {sum(1 for _, p in ok if p)}/{len(ok)} 通过，{len(failed)} 项失败", file=sys.stderr)
if unverified:
    # ⛔ 未验证 ≠ 通过：环境问题按仓库退出码约定走 **2**
    print("未验证项（stderr，**不是通过**）：", file=sys.stderr)
    for n in unverified:
        print(f"  ⚠️ {n}", file=sys.stderr)
sys.exit(1 if failed else (2 if unverified else 0))