dzxj_dtu/backend/scripts/test_alarm_flow.py

"""同步映射 + 非法插接告警端到端测试

场景（PANEL_dtu_001_2, 地址=2）:
  1. 订阅 event / alarm / patchpanel status / response 主题
  2. 清空注入 & 现有 port_state
  3. SYNC_PORT_MAPPING: port 1 期望 AABBCCDDEEFF, port 5 期望 112233445566
  4. 注入正确卡（port 1 = AABBCCDDEEFF, port 5 = 112233445566）-> 预期 CONNECT, 无告警
  5. 注入错误卡（port 1 = DEADBEEF0001）-> 预期 MOVE + ILLEGAL_CONNECT
  6. 注入空（port 5 拔出）-> 预期 DISCONNECT + ILLEGAL_DISCONNECT
  7. 恢复正确卡（port 1 恢复 AABBCCDDEEFF）-> 预期 MOVE，告警清零

轮询间隔约 5s，脚本每步等待 8s 收 2 轮轮询结果。
"""
import json
import time
import uuid
import requests
import paho.mqtt.client as mqtt

DTU_HOST = "192.168.199.149"
DTU_PORT = 5001
BROKER = "xt.wenhq.top"
BPORT = 8581
BUSER, BPASS = "admin", "admin"

PREFIX = "线架系统"
CUSTOMER = "default_customer"
DTU_ID = "dtu_001"
PANEL_ID = "PANEL_dtu_001_2"
DEVICE_ADDR = 2

control_topic = f"{PREFIX}/{CUSTOMER}/dtu/{DTU_ID}/control"

events, alarms, responses, statuses = [], [], [], []


def ts():
    return time.strftime("%H:%M:%S")


def on_connect(client, userdata, flags, rc):
    client.subscribe(f"{PREFIX}/{CUSTOMER}/#", qos=1)
    print(f"[{ts()}] 已订阅 {PREFIX}/{CUSTOMER}/#")


def on_message(client, userdata, msg):
    try:
        env = json.loads(msg.payload.decode("utf-8"))
        pl = env.get("payload", {})
    except Exception:
        return

    if msg.topic.endswith("/event"):
        events.append(pl)
        print(f"[{ts()}] 🔔 EVENT panel={pl.get('panel_id')} port={pl.get('port_id')} "
              f"type={pl.get('event_type')} uid={pl.get('jumper_uid')} prev={pl.get('previous_jumper_uid')}")
    elif msg.topic.endswith("/alarm"):
        alarms.append(pl)
        print(f"[{ts()}] ⚠️  ALARM panel={pl.get('panel_id')} port={pl.get('port_id')} "
              f"type={pl.get('alarm_type')} sev={pl.get('severity')} "
              f"exp={pl.get('expected_jumper_uid')} act={pl.get('actual_jumper_uid')}")
    elif msg.topic.endswith("/response"):
        responses.append(pl)
        print(f"[{ts()}] ✅ RESP cmd={pl.get('command')} success={pl.get('success')} err={pl.get('error_code')}")
    elif "/patchpanel/" in msg.topic and msg.topic.endswith("/status"):
        if pl.get("panel_id") == PANEL_ID:
            connected = [p for p in pl.get("ports", []) if p.get("jumper_uid")]
            illegal = [p for p in pl.get("ports", []) if p.get("status") == "ILLEGAL"]
            statuses.append(pl)
            if connected or illegal:
                summary = ", ".join(f"p{p['port_id']}:{p['status']}={p.get('jumper_uid','-')}"
                                    for p in (connected + illegal))
                print(f"[{ts()}] 📊 面板状态: {summary}")


def send_ctrl(client, command, target, params=None):
    env = {
        "msg_id": f"ctrl_{uuid.uuid4().hex[:8]}",
        "timestamp": int(time.time() * 1000),
        "dtu_id": DTU_ID, "type": "CONTROL",
        "payload": {"command": command, "target": target, "params": params or {}},
    }
    print(f"\n[{ts()}] ▶ 下发 {command} target={target} params={params}")
    client.publish(control_topic, json.dumps(env), qos=1)
    return env["msg_id"]


def inject(cards_map):
    """cards_map: {port_id: hex12|None}, 全部注入到 DEVICE_ADDR"""
    body = {"cards": {str(DEVICE_ADDR): {str(k): v for k, v in cards_map.items()}}}
    r = requests.post(f"http://{DTU_HOST}:{DTU_PORT}/api/test/inject_cards",
                      json=body, timeout=5)
    print(f"[{ts()}] 💉 注入 addr={DEVICE_ADDR} {cards_map} -> {r.status_code}")


def clear_inject():
    r = requests.post(f"http://{DTU_HOST}:{DTU_PORT}/api/test/inject_cards",
                      json={"cards": None}, timeout=5)
    print(f"[{ts()}] 🧹 清空注入 -> {r.status_code}")


def wait(seconds, label=""):
    print(f"[{ts()}] ⏳ 等 {seconds}s {label}")
    time.sleep(seconds)


def main():
    client = mqtt.Client(client_id="alarm_test")
    client.username_pw_set(BUSER, BPASS)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, BPORT, 60)
    client.loop_start()
    time.sleep(2)

    UID_A = "aabbccddeeff"
    UID_B = "112233445566"
    UID_WRONG = "deadbeef0001"

    # -------- 步骤 1: 清零 & 同步期望映射 --------
    print("\n===== 步骤 1: 同步期望映射（port1=A, port5=B）=====")
    events.clear(); alarms.clear(); responses.clear()
    send_ctrl(client, "SYNC_ALL_MAPPING", "all", {"mappings": [
        {"panel_id": PANEL_ID, "port_id": 1, "jumper_uid": UID_A.upper()},
        {"panel_id": PANEL_ID, "port_id": 5, "jumper_uid": UID_B.upper()},
    ]})
    wait(3, "等 SYNC 生效")
    # 清空所有假数据（无卡状态）
    inject({i: None for i in range(1, 25)})
    wait(6, "等一轮无卡轮询稳定 baseline")

    # -------- 步骤 2: 插入正确卡 --------
    print("\n===== 步骤 2: 插入正确卡（port1=A, port5=B）预期 CONNECT，无 ILLEGAL =====")
    events.clear(); alarms.clear()
    inject({1: UID_A, 5: UID_B, **{i: None for i in range(2, 25) if i != 5}})
    wait(8, "等 CONNECT 事件")
    connect_events = [e for e in events if e.get("event_type") == "CONNECT"]
    connect_ports = sorted(e.get("port_id") for e in connect_events)
    illegal_now = [a for a in alarms if a.get("alarm_type", "").startswith("ILLEGAL_")]
    print(f"[结果] CONNECT ports={connect_ports}, ILLEGAL 告警={len(illegal_now)}")
    assert 1 in connect_ports and 5 in connect_ports, "预期 port 1/5 CONNECT"
    assert len(illegal_now) == 0, f"不应有 ILLEGAL 告警: {illegal_now}"
    print("✅ 通过：正确卡不触发 ILLEGAL")

    # -------- 步骤 3: 换错卡 --------
    print("\n===== 步骤 3: port1 换成错卡（DEADBEEF0001）预期 MOVE + ILLEGAL_CONNECT =====")
    events.clear(); alarms.clear()
    inject({1: UID_WRONG, 5: UID_B, **{i: None for i in range(2, 25) if i != 5}})
    wait(8, "等 MOVE + ILLEGAL_CONNECT")
    move_events = [e for e in events if e.get("event_type") == "MOVE" and e.get("port_id") == 1]
    illegal_conn = [a for a in alarms if a.get("alarm_type") == "ILLEGAL_CONNECT" and a.get("port_id") == 1]
    print(f"[结果] port1 MOVE={len(move_events)}, ILLEGAL_CONNECT={len(illegal_conn)}")
    assert move_events, "预期 port1 MOVE 事件"
    assert illegal_conn, "预期 port1 ILLEGAL_CONNECT 告警"
    a0 = illegal_conn[0]
    assert a0.get("expected_jumper_uid", "").lower() == UID_A, \
        f"expected_jumper_uid 错: {a0.get('expected_jumper_uid')}"
    assert a0.get("actual_jumper_uid", "").lower() == UID_WRONG, \
        f"actual_jumper_uid 错: {a0.get('actual_jumper_uid')}"
    print(f"✅ 通过：ILLEGAL_CONNECT expected={a0.get('expected_jumper_uid')} actual={a0.get('actual_jumper_uid')} sev={a0.get('severity')}")

    # -------- 步骤 4: port5 拔出 --------
    print("\n===== 步骤 4: port5 拔出（保持 port1 错卡）预期 DISCONNECT + ILLEGAL_DISCONNECT =====")
    events.clear(); alarms.clear()
    inject({1: UID_WRONG, **{i: None for i in range(2, 25)}})
    wait(8, "等 DISCONNECT + ILLEGAL_DISCONNECT")
    disc_events = [e for e in events if e.get("event_type") == "DISCONNECT" and e.get("port_id") == 5]
    illegal_disc = [a for a in alarms if a.get("alarm_type") == "ILLEGAL_DISCONNECT" and a.get("port_id") == 5]
    print(f"[结果] port5 DISCONNECT={len(disc_events)}, ILLEGAL_DISCONNECT={len(illegal_disc)}")
    assert disc_events, "预期 port5 DISCONNECT 事件"
    assert illegal_disc, "预期 port5 ILLEGAL_DISCONNECT 告警"
    print(f"✅ 通过：ILLEGAL_DISCONNECT expected={illegal_disc[0].get('expected_jumper_uid')}")

    # -------- 步骤 5: 累计告警 3 次升级 CRITICAL --------
    print("\n===== 步骤 5: 累计告警观察 severity 升级（>=3 变 CRITICAL）=====")
    events.clear(); alarms.clear()
    inject({1: UID_WRONG, **{i: None for i in range(2, 25)}})  # 保持错卡+port5 拔出
    wait(18, "等 3 轮轮询累计告警")
    sev1 = [a.get("severity") for a in alarms if a.get("port_id") == 1]
    sev5 = [a.get("severity") for a in alarms if a.get("port_id") == 5]
    print(f"[结果] port1 severity 序列={sev1}, port5={sev5}")
    if any(s == "CRITICAL" for s in sev1 + sev5):
        print("✅ 通过：severity 已升级到 CRITICAL")
    else:
        print("ℹ 告警仍为 WARNING（未达 3 次阈值）")

    # -------- 步骤 6: 恢复正确卡 --------
    print("\n===== 步骤 6: 恢复正确卡 预期 MOVE + 告警清零 =====")
    events.clear(); alarms.clear()
    inject({1: UID_A, 5: UID_B, **{i: None for i in range(2, 25) if i != 5}})
    wait(8, "等 MOVE 恢复")
    move1 = [e for e in events if e.get("event_type") == "MOVE" and e.get("port_id") == 1]
    conn5 = [e for e in events if e.get("event_type") == "CONNECT" and e.get("port_id") == 5]
    illegal_still = [a for a in alarms if a.get("alarm_type", "").startswith("ILLEGAL_")]
    print(f"[结果] port1 MOVE={len(move1)}, port5 CONNECT={len(conn5)}, ILLEGAL={len(illegal_still)}")
    assert move1 or conn5, "预期 port1 恢复"
    print("✅ 通过：port 恢复到期望 UID，告警停止累计")

    # -------- 收尾: 清空注入 --------
    print("\n===== 收尾：清空注入 =====")
    clear_inject()

    print(f"\n===== 汇总：events={len(events)}, alarms={len(alarms)}, responses={len(responses)}, statuses={len(statuses)} =====")
    client.loop_stop()
    client.disconnect()


if __name__ == "__main__":
    try:
        main()
        print("\n🎉 所有步骤 PASS")
    except AssertionError as e:
        print(f"\n❌ 断言失败: {e}")
        raise
    except Exception as e:
        print(f"\n❌ 异常: {e}")
        raise