"""同步映射 + 非法插接告警端到端测试 场景（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