#!/usr/bin/env python3 """ 数据中心 → 信号管理页面 (out) 添加信号功能测试 (v3) 修复: 终端命令用 "data" 字段而非 "cmd" """ import socket, base64, os, struct, json, time, sys IP = '198.120.0.100'; PORT = 8000 def ws_handshake(sock): key = base64.b64encode(os.urandom(16)).decode() req = (f"GET /ws HTTP/1.1\r\nHost: {IP}:{PORT}\r\nUpgrade: websocket\r\n" f"Connection: Upgrade\r\nSec-WebSocket-Key: {key}\r\n" f"Sec-WebSocket-Version: 13\r\n\r\n") sock.sendall(req.encode()) resp = b"" while b"\r\n\r\n" not in resp: c = sock.recv(4096) if not c: break resp += c if b"101" not in resp.split(b"\r\n")[0]: raise Exception(f"Handshake failed: {resp[:200]}") return resp def ws_send(sock, msg): if isinstance(msg, dict): msg = json.dumps(msg, ensure_ascii=False) payload = msg.encode() L = len(payload) mask = os.urandom(4) masked = bytes(payload[i] ^ mask[i % 4] for i in range(L)) frame = bytearray([0x81]) if L < 126: frame.append(0x80 | L) elif L < 65536: frame.append(0x80 | 126) frame.extend(struct.pack('>H', L)) else: frame.append(0x80 | 127) frame.extend(struct.pack('>Q', L)) frame.extend(mask) frame.extend(masked) sock.sendall(bytes(frame)) def ws_recv_frames(sock, timeout=2.0): """接收所有可用帧，解析后返回消息列表""" sock.settimeout(timeout) all_data = b"" try: while True: chunk = sock.recv(65536) if not chunk: break all_data += chunk except socket.timeout: pass msgs = [] offset = 0 while offset + 2 <= len(all_data): b0, b1 = all_data[offset], all_data[offset + 1] opcode = b0 & 0x0F length = b1 & 0x7F offset += 2 if length == 126: if offset + 2 > len(all_data): break length = struct.unpack('>H', all_data[offset:offset+2])[0] offset += 2 elif length == 127: if offset + 8 > len(all_data): break length = struct.unpack('>Q', all_data[offset:offset+8])[0] offset += 8 if offset + length > len(all_data): break payload = all_data[offset:offset+length] offset += length if opcode in (0x01, 0x02): try: msgs.append(json.loads(payload.decode('utf-8'))) except: msgs.append(payload.decode('utf-8', errors='replace')) elif opcode == 0x08: pass return msgs def main(): print("=" * 70) print(" RTU 数据中心 → 信号管理页(out) 添加信号功能测试") print(f" 目标: {IP}:{PORT}") print("=" * 70) sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(10) sock.connect((IP, PORT)) print(f"[✓] TCP 连接成功") ws_handshake(sock) print("[✓] WebSocket 握手成功") # ---- 步骤 1: 注册 dc_data 信号 ---- # 注意: 终端命令格式为 {"type":"cmd","data":"datacenter out"} print("\n" + "=" * 70) print(" 步骤 1: 通过 datacenter 命令注册所有类型信号") print("=" * 70) all_dc_data = {} for st in ['out', 'in', 'yk', 'ao', 'param']: print(f"\n [→] datacenter {st}") ws_send(sock, {"type": "cmd", "data": f"datacenter {st}"}) time.sleep(1.5) msgs = ws_recv_frames(sock, timeout=3.0) print(f" [i] 收到 {len(msgs)} 条消息") for msg in msgs: if isinstance(msg, dict): if msg.get('type') == 'dc_data': all_dc_data[st] = msg cnt = len(msg.get(st, [])) print(f" [✓] dc_data: {st} = {cnt} 个信号") for item in msg.get(st, [])[:5]: print(f" saddr={item.get('saddr','?')} desc={item.get('desc','?')} ctrl={item.get('ctrl_type','?')}") if cnt > 5: print(f" ... 还有 {cnt-5} 个") elif isinstance(msg, str) and len(msg) < 500: print(f" [cmd输出] {msg.strip()[:200]}") # ---- 步骤 2: 选择 out 信号 ---- print("\n" + "=" * 70) print(" 步骤 2: 选择 out 信号并添加到配置页") print("=" * 70) dc_out = all_dc_data.get('out', {}) out_signals = dc_out.get('out', []) if not out_signals: print(" [!] 没有可用的 out 信号") print(" 原因: datacenter 命令未注册任何 out 信号到数据中心") print(" 可能需要在配置文件中添加信号定义") sock.close() return test_signals = out_signals[:3] test_saddrs = {s.get('saddr') for s in test_signals} print(f" 从 {len(out_signals)} 个信号中选择 {len(test_signals)} 个:") for i, sig in enumerate(test_signals): print(f" [{i+1}] saddr={sig.get('saddr')} desc={sig.get('desc','?')}") # 排空之前的消息 ws_recv_frames(sock, timeout=0.5) # ---- 步骤 3: 执行 curd:add ---- print("\n" + "=" * 70) print(" 步骤 3: 执行 curd:add (模拟前端「添加到配置页」)") print("=" * 70) for i, sig in enumerate(test_signals): saddr = sig.get('saddr') print(f"\n [{i+1}] curd:add saddr={saddr}") ws_send(sock, { "saddr": saddr, "signal_type": "out", "curd": "add", "setting_zone": "0", "signal_data": "" }) time.sleep(2.0) # 2秒间隔模拟真人操作 # ---- 步骤 4: 等待 ws_task 推送 out 数据 ---- print("\n" + "=" * 70) print(" 步骤 4: 等待 ws_task 周期性推送 out 数据") print(" (add 操作不会立即推送，需要等 ws_task 检测到 has_change)") print("=" * 70) out_received = False out_data = [] start = time.time() while time.time() - start < 15.0: msgs = ws_recv_frames(sock, timeout=3.0) for msg in msgs: if isinstance(msg, dict): out_arr = msg.get('out') if isinstance(out_arr, list) and len(out_arr) > 0: out_data = out_arr out_received = True break if out_received: break elapsed = time.time() - start if elapsed > 2: print(f" [i] 等待中... ({elapsed:.0f}s)") # ---- 步骤 5: 验证结果 ---- print("\n" + "=" * 70) print(" 步骤 5: 验证结果") print("=" * 70) if out_received: print(f" [✓] 收到 out 数据推送: {len(out_data)} 个注册信号") registered_saddrs = {o.get('saddr') for o in out_data} for o in out_data: marker = " ← 新添加" if o.get('saddr') in test_saddrs else "" print(f" saddr={o.get('saddr','?')} val={o.get('val','?')} desc={o.get('desc','?')}{marker}") all_found = test_saddrs.issubset(registered_saddrs) missing = test_saddrs - registered_saddrs if all_found: print(f"\n ✅ 所有 {len(test_saddrs)} 个信号成功注册到 out!") else: print(f"\n ❌ 缺少 {len(missing)} 个信号: {missing}") else: print(" ❌ 未收到 out 数据推送") print(" 说明: add_signal 后服务端没有通过 ws_task 推送变化") print(" 可能原因: has_change 为 false (新信号初始值可能与 last_val 相同)") # ---- 步骤 6: 清理 ---- print("\n" + "=" * 70) print(" 步骤 6: 清理 — 删除测试添加的信号") print("=" * 70) for sig in test_signals: saddr = sig.get('saddr') print(f" [→] del saddr={saddr}") ws_send(sock, { "saddr": saddr, "signal_type": "out", "curd": "del", "setting_zone": "0", "signal_data": "" }) time.sleep(0.3) # del 后立即推送 msgs = ws_recv_frames(sock, timeout=2.0) for msg in msgs: if isinstance(msg, dict) and msg.get('out') is not None: print(f" [i] 删除后 out 信号数: {len(msg['out'])}") # 发送关闭帧 try: sock.sendall(b'\x88\x80' + os.urandom(4)) except: pass sock.close() print("\n[✓] 测试完成") if __name__ == '__main__': main()