xianyan/docs/toolsapi/scripts/test_signaling_server.py

import paramiko
import json
import time
import hashlib
import sys

HOST = '123.207.67.197'
PORT = 22
USER = 'root'
PASS = '520Kiss123'
WS_PORT = 3001

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(HOST, PORT, USER, PASS)

print('=== 接口测试: 信令服务器 v11.0.0 ===\n')

test_results = []

def run_test(name, command):
    print(f'\n--- 测试: {name} ---')
    stdin, stdout, stderr = ssh.exec_command(command)
    out = stdout.read().decode().strip()
    err = stderr.read().decode().strip()
    print(f'输出: {out[:500]}')
    if err:
        print(f'错误: {err[:300]}')
    success = 'error' not in out.lower() and 'fail' not in out.lower() and not err
    test_results.append((name, success))
    return out, err

print('1. 检查服务运行状态')
out, err = run_test('PM2 Status', 'pm2 describe signaling | grep status')
if 'online' in out:
    print('   ✅ 信令服务在线')
else:
    print('   ❌ 信令服务离线')

print('\n2. 检查WebSocket端口监听')
out, err = run_test('Port Check', f'ss -tlnp | grep {WS_PORT}')
if str(WS_PORT) in out:
    print(f'   ✅ 端口 {WS_PORT} 正在监听')
else:
    print(f'   ❌ 端口 {WS_PORT} 未监听')

print('\n3. 测试WebSocket连接 + register')
node_test = f'''
const WebSocket = require("ws");
const ws = new WebSocket("ws://127.0.0.1:{WS_PORT}");
let results = [];

ws.on("open", () => {{
    ws.send(JSON.stringify({{
        type: "register",
        payload: {{ userId: "test_user_001", alias: "TestDevice", platform: "test" }}
    }}));
}});

ws.on("message", (data) => {{
    const msg = JSON.parse(data);
    results.push(msg);
    if (msg.type === "registered") {{
        console.log("REGISTER OK: " + JSON.stringify(msg));
        ws.send(JSON.stringify({{
            type: "discoverMyDevices",
            payload: {{ userId: "test_user_001" }}
        }}));
    }}
    if (msg.type === "myDevicesResponse") {{
        console.log("DISCOVER OK: " + JSON.stringify(msg));
        ws.send(JSON.stringify({{
            type: "pair-request",
            to: "nonexistent_peer",
            fingerprint: "test_fp_123"
        }}));
    }}
    if (msg.type === "pair-response") {{
        console.log("PAIR-REQUEST ROUTED: " + JSON.stringify(msg));
        ws.send(JSON.stringify({{
            type: "heartbeat"
        }}));
        console.log("HEARTBEAT OK");
        ws.close();
    }}
    if (msg.type === "pong") {{
        console.log("PONG OK");
    }}
}});

ws.on("error", (err) => {{
    console.log("WS ERROR: " + err.message);
}});

setTimeout(() => {{
    console.log("TIMEOUT - closing");
    ws.close();
    process.exit(0);
}}, 5000);
'''

out, err = run_test('WebSocket + Register + Discover + Pair', f'cd /www/wwwroot/tools.wktyl.com/signaling && node -e \'{node_test.replace(chr(10), " ").replace(chr(13), "")}\'')

print('\n4. 测试通用消息转发 (delivery-ack)')
node_test_relay = f'''
const WebSocket = require("ws");
const ws1 = new WebSocket("ws://127.0.0.1:{WS_PORT}");
const ws2 = new WebSocket("ws://127.0.0.1:{WS_PORT}");
let peer1Id, peer2Id;

ws1.on("open", () => {{
    ws1.send(JSON.stringify({{ type: "register", payload: {{ userId: "test_relay_user" }} }}));
}});

ws1.on("message", (data) => {{
    const msg = JSON.parse(data);
    if (msg.type === "registered") peer1Id = msg.id;
}});

ws2.on("open", () => {{
    ws2.send(JSON.stringify({{ type: "register", payload: {{ userId: "test_relay_user" }} }}));
}});

ws2.on("message", (data) => {{
    const msg = JSON.parse(data);
    if (msg.type === "registered") {{
        peer2Id = msg.id;
        setTimeout(() => {{
            ws1.send(JSON.stringify({{
                type: "delivery-ack",
                to: peer2Id,
                messageId: "test_msg_001",
                status: "delivered"
            }}));
        }}, 500);
    }}
    if (msg.type === "delivery-ack") {{
        console.log("DELIVERY-ACK FORWARDED: " + JSON.stringify(msg));
        ws1.close();
        ws2.close();
        process.exit(0);
    }}
}});

setTimeout(() => {{
    console.log("TIMEOUT");
    ws1.close();
    ws2.close();
    process.exit(0);
}}, 8000);
'''

out, err = run_test('Delivery-Ack Forward', f'cd /www/wwwroot/tools.wktyl.com/signaling && node -e \'{node_test_relay.replace(chr(10), " ").replace(chr(13), "")}\'')

print('\n5. 测试 chunk-ack / resume-request 通用转发')
node_test_chunk = f'''
const WebSocket = require("ws");
const ws1 = new WebSocket("ws://127.0.0.1:{WS_PORT}");
const ws2 = new WebSocket("ws://127.0.0.1:{WS_PORT}");
let peer1Id, peer2Id;

ws1.on("open", () => {{
    ws1.send(JSON.stringify({{ type: "register", payload: {{ userId: "test_chunk_user" }} }}));
}});

ws1.on("message", (data) => {{
    const msg = JSON.parse(data);
    if (msg.type === "registered") peer1Id = msg.id;
}});

ws2.on("open", () => {{
    ws2.send(JSON.stringify({{ type: "register", payload: {{ userId: "test_chunk_user" }} }}));
}});

ws2.on("message", (data) => {{
    const msg = JSON.parse(data);
    if (msg.type === "registered") {{
        peer2Id = msg.id;
        setTimeout(() => {{
            ws1.send(JSON.stringify({{
                type: "chunk-ack",
                to: peer2Id,
                taskId: "test_task_001",
                chunkIndex: 5
            }}));
        }}, 500);
    }}
    if (msg.type === "chunk-ack") {{
        console.log("CHUNK-ACK FORWARDED: " + JSON.stringify(msg));
        ws1.send(JSON.stringify({{
            type: "resume-request",
            to: peer2Id,
            taskId: "test_task_001",
            missingChunks: [3, 7, 12]
        }}));
    }}
    if (msg.type === "resume-request") {{
        console.log("RESUME-REQUEST FORWARDED: " + JSON.stringify(msg));
        ws1.close();
        ws2.close();
        process.exit(0);
    }}
}});

setTimeout(() => {{
    console.log("TIMEOUT");
    ws1.close();
    ws2.close();
    process.exit(0);
}}, 8000);
'''

out, err = run_test('Chunk-Ack + Resume-Request Forward', f'cd /www/wwwroot/tools.wktyl.com/signaling && node -e \'{node_test_chunk.replace(chr(10), " ").replace(chr(13), "")}\'')

print('\n\n========================================')
print('测试结果汇总:')
print('========================================')
for name, success in test_results:
    status = '✅ PASS' if success else '❌ FAIL'
    print(f'  {status} - {name}')

ssh.close()