WebSocket Tester
无需浏览器即可测试 WebSocket 端点。连接、发送消息、验证响应、测试重连、测量延迟、并发压测——适用于聊天、实时更新、游戏服务器、IoT、流式 API。
使用场景：
“test websocket”“websocket not connecting”“debug real-time”“ws connection issues”“websocket load test”“test socket.io”“chat server testing”或构建/调试实时功能时。
Commands
- connect — 测试 WebSocket 连接
# 使用 wscat（Node.js）
npx wscat -c "wss://$HOST/ws" 2>&1 & WSCAT_PID=$!
sleep 3
kill $WSCAT_PID 2>/dev/null
# 或使用 websocat（Rust）
echo "test" | websocat "wss://$HOST/ws" 2>&1
# 或使用 Python
python3 -c "
import asyncio, websockets, time, json
async def test_connection():
url = 'wss://$HOST/ws'
start = time.time()
try:
async with websockets.connect(url, close_timeout=5) as ws:
latency = (time.time() - start) 1000
print(f'✅ Connected in {latency:.0f}ms')
print(f'Protocol: {ws.subprotocol or \"none\"}')
print(f'Headers: {dict(ws.response_headers)}')
pong = await ws.ping()
await pong
print('✅ Ping/Pong working')
try:
msg = await asyncio.wait_for(ws.recv(), timeout=3)
print(f'Server sent: {msg[:200]}')
except asyncio.TimeoutError:
print('No initial message from server (normal for some protocols)')
except websockets.exceptions.InvalidStatusCode as e:
print(f'❌ Connection rejected: HTTP {e.status_code}')
except ConnectionRefusedError:
print(f'❌ Connection refused — server not listening on WebSocket endpoint')
except Exception as e:
print(f'❌ Error: {e}')
asyncio.run(test_connection())
" 2>&1
import asyncio, websockets, json, time
async def test_messages():
async with websockets.connect('wss://$HOST/ws') as ws:
# Test 1: 收发
test_msg = json.dumps({'type': 'ping', 'data': 'hello'})
await ws.send(test_msg)
print(f'→ Sent: {test_msg}')
response = await asyncio.wait_for(ws.recv(), timeout=5)
print(f'← Received: {response[:200]}')
# 验证响应格式
try:
data = json.loads(response)
assert 'type' in data, 'Missing type field'
print('✅ Response is valid JSON with type field')
except json.JSONDecodeError:
print('⚠️ Response is not JSON — might be binary or plain text')
except AssertionError as e:
print(f'❌ Schema validation: {e}')
# Test 2: 测量往返延迟
latencies = []
for i in range(10):
start = time.time()
await ws.send(json.dumps({'type': 'ping', 'seq': i}))
await asyncio.wait_for(ws.recv(), timeout=5)
latencies.append((time.time() - start) 1000)
avg = sum(latencies) / len(latencies)
p95 = sorted(latencies)[int(len(latencies) * 0.95)]
print(f'\\nLatency (10 round-trips): avg={avg:.1f}ms, p95={p95:.1f}ms')
asyncio.run(test_messages())
async def test_reconnection():
"""模拟断线并验证客户端重连"""
# Test 1: 服务器主动关闭
async with websockets.connect('wss://$HOST/ws') as ws:
print('Connected. Sending close frame...')
await ws.close(1001, 'Testing reconnection')
print(f'Close code: {ws.close_code}, reason: {ws.close_reason}')
# Test 2: 重连并验证状态
async with websockets.connect('wss://$HOST/ws') as ws:
print('✅ Reconnected successfully')
await ws.send(json.dumps({'type': 'get_state'}))
state = await asyncio.wait_for(ws.recv(), timeout=5)
print(f'State after reconnect: {state[:200]}')
# Test 3: 带 auth token 连接
async with websockets.connect(
'wss://$HOST/ws',
extra_headers={'Authorization': f'Bearer {TOKEN}'}
) as ws:
print('✅ Authenticated connection')
import asyncio, websockets, time, statistics
async def load_test(url, num_connections, duration_sec, messages_per_sec):
"""WebSocket 端点压测"""
results = {
'connected': 0,
'failed': 0,
'messages_sent': 0,
'messages_received': 0,
'latencies': [],
'errors': [],
}
async def client(client_id):
try:
async with websockets.connect(url, close_timeout=5) as ws:
results['connected'] += 1
end_time = time.time() + duration_sec
while time.time() < end_time:
by