Testing WebSocket Connections — Real-Time Communication Testing Guide
In this tutorial, you'll learn about Testing Websocket Connections. We cover key concepts, practical examples, and best practices.
WebSocket testing is different from HTTP testing because connections are persistent, bidirectional, and stateful — you must verify message ordering, reconnection behavior, concurrent connection handling, and protocol compliance rather than simple request-response pairs. In this guide, you will learn how to test WebSocket connections with Python and JavaScript, validate real-time message delivery, simulate network interruptions, and build automated test suites for real-time applications. Doda Browser's live collaboration feature uses WebSockets for real-time document editing, and every release includes a full WebSocket test suite.
Learning Path
flowchart LR A[API Testing] --> B[HTTP Fundamentals] B --> C[Testing WebSockets
You are here] C --> D[Real-Time Monitoring] D --> E[Performance Testing] style C fill:#f90,color:#fff
Testing WebSocket Connections with Python
Use the websockets library to test connection establishment and message exchange:
import asyncio, websockets, json
async def test_connection():
async with websockets.connect("ws://localhost:8080/ws") as ws:
await ws.send(json.dumps({"type": "ping"}))
response = await ws.recv()
data = json.loads(response)
assert data["type"] == "pong"
print("Connection and ping-pong test passed")
asyncio.run(test_connection())
Expected output:
Connection and ping-pong test passed
Testing Message Broadcasting
Verify that messages sent by one client are received by others:
import asyncio, websockets, json
async def test_broadcast():
async with websockets.connect("ws://localhost:8080/ws") as sender:
async with websockets.connect("ws://localhost:8080/ws") as receiver:
msg = json.dumps({"type": "message", "text": "Hello everyone"})
await sender.send(msg)
received = json.loads(await receiver.recv())
assert received["type"] == "message"
assert received["text"] == "Hello everyone"
print("Broadcast test passed")
asyncio.run(test_broadcast())
Testing Reconnection
Simulate network drops and verify automatic reconnection:
import asyncio, websockets, json
async def test_reconnection():
uri = "ws://localhost:8080/ws"
async with websockets.connect(uri) as ws:
await ws.send(json.dumps({"type": "ping"}))
first = await ws.recv()
print(f"First message received: {first}")
async with websockets.connect(uri) as ws:
await ws.send(json.dumps({"type": "ping"}))
second = await ws.recv()
print(f"After reconnect: {second}")
assert first == second, "Reconnection should work seamlessly"
asyncio.run(test_reconnection())
Testing Concurrent Connections
Verify the server handles multiple simultaneous connections:
import asyncio, websockets, json
async def test_concurrent_connections():
uri = "ws://localhost:8080/ws"
async def client(client_id):
async with websockets.connect(uri) as ws:
await ws.send(json.dumps({"type": "ping", "id": client_id}))
response = await ws.recv()
data = json.loads(response)
assert data["type"] == "pong"
return client_id
results = await asyncio.gather(*[client(i) for i in range(10)])
assert len(results) == 10
print(f"All {len(results)} concurrent connections succeeded")
asyncio.run(test_concurrent_connections())
Expected output:
All 10 concurrent connections succeeded
Testing WebSocket with JavaScript
const WebSocket = require('ws');
function testWebSocketConnection() {
return new Promise((resolve, reject) => {
const ws = new WebSocket('ws://localhost:8080/ws');
ws.on('open', () => {
ws.send(JSON.stringify({ type: 'ping' }));
});
ws.on('message', (data) => {
const response = JSON.parse(data.toString());
if (response.type === 'pong') {
console.log('JavaScript WebSocket test passed');
ws.close();
resolve();
}
});
ws.on('error', reject);
setTimeout(() => reject(new Error('Timeout')), 5000);
});
}
testWebSocketConnection();
Testing Invalid Messages
Verify the server rejects malformed messages gracefully:
import asyncio, websockets, json
async def test_invalid_messages():
async with websockets.connect("ws://localhost:8080/ws") as ws:
await ws.send("not valid json")
response = await ws.recv()
data = json.loads(response)
assert data["type"] == "error"
assert "invalid" in data["message"].lower()
print("Invalid message test passed")
async with websockets.connect("ws://localhost:8080/ws") as ws:
await ws.send(json.dumps({"type": "unknown_type"}))
response = await ws.recv()
data = json.loads(response)
assert data["type"] == "error"
print("Unknown type test passed")
asyncio.run(test_invalid_messages())
Practice Questions
1. Why is WebSocket testing different from HTTP API testing?
WebSockets are persistent, bidirectional, and stateful. You must test message ordering, reconnection, concurrent connections, and real-time delivery.
2. How do you test message broadcasting in WebSockets?
Open multiple connections, send a message from one client, and verify all other connected clients receive it.
3. What is the reconnection test and why is it important?
It tests that a client can reconnect after a network drop and continue receiving messages without data loss or duplicate messages.
4. How do you simulate network interruptions in WebSocket tests?
Close the client connection programmatically and verify the client reconnects and resumes normal operation. More advanced tests use tools like toxiproxy to simulate latency and packet loss.
Challenge: Build a WebSocket chat server and test suite. Test: connection, message broadcasting to all users, private messaging between two users, reconnection with message replay, 100 concurrent users, and invalid message rejection.
FAQ
What's Next
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro