WebSocket Testing

GET /chat HTTP/1.1
Host: example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

Client                          Server
  |--- HTTP GET /ws (Upgrade) --->|
  |<-- 101 Switching Protocols ---|
  |=== WebSocket frames =========|   (bidirectional)
  |--- Ping ---------------------->|  (heartbeat)
  |<-- Pong -----------------------|
  |--- Close (0x8, 1000) -------->|  (clean teardown)
  |<-- Close (0x8, 1000) ---------|

import { test, expect } from '@playwright/test';

test('chat message appears in UI after WebSocket delivery', async ({ page }) => {
  const wsFrames: { dir: string; payload: string }[] = [];

  // Register before navigation so no frames are missed
  page.on('websocket', (ws) => {
    ws.on('framesent', (frame) => {
      wsFrames.push({ dir: 'sent', payload: frame.payload as string });
    });
    ws.on('framereceived', (frame) => {
      wsFrames.push({ dir: 'received', payload: frame.payload as string });
    });
    ws.on('close', () => console.log('WebSocket closed:', ws.url()));
  });

  await page.goto('/chat/room/42');

  // Send a message through the UI
  await page.fill('[data-testid="message-input"]', 'Hello room');
  await page.click('[data-testid="send-button"]');

  // Assert the message appears in the conversation
  await expect(page.locator('[data-testid="message-list"]'))
    .toContainText('Hello room');

  // Assert the frame was sent on the WebSocket
  const sentFrames = wsFrames.filter((f) => f.dir === 'sent');
  expect(sentFrames.length).toBeGreaterThan(0);

  const lastSent = JSON.parse(sentFrames.at(-1)!.payload);
  expect(lastSent).toMatchObject({ type: 'message', text: 'Hello room' });
});

test('notification badge increments on each server push', async ({ page }) => {
  await page.routeWebSocket('ws://localhost:3000/notifications', (ws) => {
    ws.onMessage((msg) => {
      const parsed = JSON.parse(msg as string);
      if (parsed.type === 'subscribe') {
        ws.send(JSON.stringify({ type: 'subscribed', topic: parsed.topic }));
      }
    });

    // Push two server-initiated notifications after the page connects
    setTimeout(() => {
      ws.send(JSON.stringify({ type: 'notification', id: 1, text: 'New comment' }));
    }, 300);
    setTimeout(() => {
      ws.send(JSON.stringify({ type: 'notification', id: 2, text: 'New like' }));
    }, 600);
  });

  await page.goto('/dashboard');

  await expect(page.locator('[data-testid="notification-badge"]'))
    .toHaveText('2', { timeout: 2000 });
});

test.describe('Chat application — WebSocket functional tests', () => {
  test('two clients receive each other messages', async ({ browser }) => {
    const context1 = await browser.newContext();
    const context2 = await browser.newContext();
    const page1 = await context1.newPage();
    const page2 = await context2.newPage();

    const receivedByPage2: string[] = [];
    page2.on('websocket', (ws) => {
      ws.on('framereceived', (frame) => {
        try {
          const msg = JSON.parse(frame.payload as string);
          if (msg.type === 'message') receivedByPage2.push(msg.text);
        } catch { /* non-JSON frames */ }
      });
    });

    await page1.goto('/chat/room/1');
    await page2.goto('/chat/room/1');

    // Give page2 time to connect and subscribe
    await page2.waitForSelector('[data-testid="connected-indicator"]');

    // Page1 sends a message
    await page1.fill('[data-testid="message-input"]', 'Hi from page 1');
    await page1.click('[data-testid="send-button"]');

    // Page2 UI should reflect it
    await expect(page2.locator('[data-testid="message-list"]'))
      .toContainText('Hi from page 1', { timeout: 3000 });

    expect(receivedByPage2).toContain('Hi from page 1');

    await context1.close();
    await context2.close();
  });

  test('reconnect indicator appears after connection drop', async ({ page }) => {
    await page.routeWebSocket('ws://localhost:3000/chat', (ws) => {
      ws.onMessage((msg) => {
        const parsed = JSON.parse(msg as string);
        if (parsed.type === 'trigger_drop') {
          ws.close(); // close without clean close frame
        }
      });
    });

    await page.goto('/chat/room/1');

    await page.evaluate(() => {
      (window as any).__ws.send(JSON.stringify({ type: 'trigger_drop' }));
    });

    await expect(page.locator('[data-testid="reconnecting-banner"]'))
      .toBeVisible({ timeout: 2000 });
  });
});

# Connect interactively — stdin goes to the WebSocket, received frames print to stdout
websocat ws://localhost:8080/ws

# Send a single message and print all received frames until close
echo '{"type":"ping"}' | websocat ws://localhost:8080/ws

# Connect with a Bearer token header
websocat -H "Authorization: Bearer $TOKEN" wss://api.example.com/ws

# Subscribe and collect 5 frames then disconnect
websocat --text ws://localhost:8080/stream \
  <<< '{"action":"subscribe","topic":"prices"}' | head -5

# TLS with a self-signed cert
websocat --insecure wss://localhost:8443/ws

import ws from 'k6/ws';
import { check } from 'k6';
import { Trend, Counter, Rate } from 'k6/metrics';

const connectTime = new Trend('ws_connect_time_ms', true);
const msgLatency  = new Trend('ws_msg_latency_ms', true);
const msgErrors   = new Counter('ws_msg_errors');
const successRate = new Rate('ws_success_rate');

export default function () {
  const url = 'wss://chat.example.com/ws';
  const params = {
    headers: { Authorization: `Bearer ${__ENV.WS_TOKEN}` },
    tags: { endpoint: 'chat' },
  };

  const connectedAt = Date.now();

  const res = ws.connect(url, params, function (socket) {
    socket.on('open', () => {
      connectTime.add(Date.now() - connectedAt);
      socket.send(JSON.stringify({ type: 'subscribe', room: 'lobby' }));
    });

    socket.on('message', (data) => {
      const msg = JSON.parse(data);
      if (msg.timestamp) {
        msgLatency.add(Date.now() - msg.timestamp);
      }
      const valid = check(msg, {
        'has type field':   (m) => m.type !== undefined,
        'latency under 1s': () => (Date.now() - (msg.timestamp || Date.now())) < 1000,
      });
      successRate.add(valid);
      if (!valid) msgErrors.add(1);
    });

    socket.on('error', (e) => {
      msgErrors.add(1);
      successRate.add(false);
    });

    socket.setInterval(() => {
      socket.send(JSON.stringify({ type: 'ping', clientTs: Date.now() }));
    }, 5000);

    socket.setTimeout(() => socket.close(), 60000);
  });

  check(res, { 'handshake succeeded (101)': (r) => r && r.status === 101 });
}

// k6 run --env WS_URL=wss://chat.example.com/ws --env WS_TOKEN=xxx chat-soak.js

import ws from 'k6/ws';
import { check, sleep } from 'k6';
import { Trend, Counter, Rate, Gauge } from 'k6/metrics';

const connectLatency  = new Trend('ws_connect_latency_ms', true);
const msgRoundTrip    = new Trend('ws_msg_rtt_ms', true);
const connDrops       = new Counter('ws_connection_drops');
const msgErrors       = new Counter('ws_message_errors');
const activeSessions  = new Gauge('ws_active_sessions');
const deliverySuccess = new Rate('ws_delivery_success_rate');

export const options = {
  scenarios: {
    soak: {
      executor: 'ramping-vus',
      startVUs: 0,
      stages: [
        { duration: '1m',  target: 100 },  // ramp to 100 concurrent connections
        { duration: '10m', target: 100 },  // hold for 10 minutes
        { duration: '1m',  target: 0   },  // ramp down
      ],
    },
  },
  thresholds: {
    ws_connect_latency_ms:    ['p(95)<500'],
    ws_msg_rtt_ms:            ['p(50)<100', 'p(95)<500', 'p(99)<1000'],
    ws_delivery_success_rate: ['rate>0.995'],
    ws_connection_drops:      ['count<5'],
  },
};

export function setup() {
  const res = ws.connect(__ENV.WS_URL, {}, function (socket) {
    socket.setTimeout(() => socket.close(), 1000);
  });
  if (!res || res.status !== 101) {
    throw new Error('Setup check failed: server did not accept WebSocket upgrade');
  }
  return { startTime: Date.now() };
}

export default function () {
  const url = __ENV.WS_URL || 'ws://localhost:8080/ws';
  const params = {
    headers: { Authorization: `Bearer ${__ENV.WS_TOKEN || 'test-token'}` },
    tags: { scenario: 'chat_soak' },
  };

  const connStart = Date.now();

  const res = ws.connect(url, params, function (socket) {
    activeSessions.add(1);
    const pendingEchos = new Map(); // echoId -> sentAt

    socket.on('open', () => {
      connectLatency.add(Date.now() - connStart);
      socket.send(JSON.stringify({
        type: 'subscribe',
        room: `load-test-room-${__VU % 10}`,  // spread across 10 rooms
      }));
    });

    socket.on('message', (data) => {
      try {
        const msg = JSON.parse(data);
        if (msg.type === 'echo_reply' && pendingEchos.has(msg.echoId)) {
          msgRoundTrip.add(Date.now() - pendingEchos.get(msg.echoId));
          pendingEchos.delete(msg.echoId);
          deliverySuccess.add(true);
        } else if (msg.type === 'error') {
          msgErrors.add(1);
          deliverySuccess.add(false);
        }
      } catch (e) {
        msgErrors.add(1);
        deliverySuccess.add(false);
      }
    });

    socket.on('error', () => {
      connDrops.add(1);
      deliverySuccess.add(false);
    });

    socket.on('close', (code) => {
      activeSessions.add(-1);
      if (code !== 1000 && code !== 1001) {
        connDrops.add(1);
      }
    });

    // Send an echo probe every 10 seconds
    socket.setInterval(() => {
      const id = `${__VU}-${Date.now()}`;
      pendingEchos.set(id, Date.now());
      socket.send(JSON.stringify({ type: 'echo', echoId: id }));
    }, 10000);

    socket.setTimeout(() => {
      if (pendingEchos.size > 0) deliverySuccess.add(false);
      socket.close(1000, 'load test complete');
    }, 60000);
  });

  check(res, { 'WebSocket upgrade succeeded': (r) => r && r.status === 101 });
  sleep(1);
}

export function teardown(data) {
  console.log(`Soak duration: ${((Date.now() - data.startTime) / 1000).toFixed(0)}s`);
}

import httpx
import json
import pytest


def parse_sse_lines(lines: list[str]) -> list[dict]:
    """Parse a list of SSE text lines into event dicts."""
    events, current = [], {}
    for line in lines:
        line = line.strip()
        if line.startswith('id: '):
            current['id'] = line[4:]
        elif line.startswith('event: '):
            current['type'] = line[7:]
        elif line.startswith('data: '):
            current['data'] = line[6:]
        elif line == '' and current:
            events.append(current)
            current = {}
    return events


def test_sse_content_type():
    with httpx.stream('GET', 'http://localhost:8080/events') as r:
        assert r.status_code == 200
        assert r.headers['content-type'].startswith('text/event-stream')


def test_sse_delivers_three_events():
    lines = []
    with httpx.stream('GET', 'http://localhost:8080/events') as r:
        for line in r.iter_lines():
            lines.append(line)
            if len(parse_sse_lines(lines)) >= 3:
                break
    assert len(parse_sse_lines(lines)) >= 3


def test_sse_event_schema():
    lines = []
    with httpx.stream('GET', 'http://localhost:8080/events') as r:
        for line in r.iter_lines():
            lines.append(line)
            events = parse_sse_lines(lines)
            if events:
                break

    payload = json.loads(parse_sse_lines(lines)[0]['data'])
    assert 'type' in payload
    assert 'timestamp' in payload

import asyncio
import json
import jsonschema
import pytest
import websockets

NOTIFICATION_SCHEMA = {
    "type": "object",
    "required": ["type", "id", "text", "timestamp"],
    "properties": {
        "type":      {"type": "string", "enum": ["notification", "welcome", "ack"]},
        "id":        {"type": "integer"},
        "text":      {"type": "string", "minLength": 1},
        "timestamp": {"type": "integer"},
        "seq":       {"type": "integer"},
    },
    "additionalProperties": False,
}


@pytest.mark.asyncio
async def test_all_received_messages_match_schema():
    received = []
    async with websockets.connect('ws://localhost:8080/notifications') as ws:
        for _ in range(5):
            raw = await asyncio.wait_for(ws.recv(), timeout=5.0)
            msg = json.loads(raw)
            # Raises jsonschema.ValidationError on schema violation
            jsonschema.validate(instance=msg, schema=NOTIFICATION_SCHEMA)
            received.append(msg)

    assert len(received) == 5

from pact import MessageConsumer, Provider
from pact.matchers import Like

pact = (
    MessageConsumer('NotificationUI')
    .has_pact_with(Provider('NotificationService'))
)

(
    pact
    .given('a notification exists for user 42')
    .expects_to_receive('a notification message')
    .with_content({
        'type':      'notification',
        'id':        Like(1),
        'text':      Like('You have a new message'),
        'timestamp': Like(1714000000),
    })
    .with_metadata({'contentType': 'application/json'})
)

// test-helpers/ws-echo-server.js
const WebSocket = require('ws');

const server = new WebSocket.Server({ port: process.env.WS_TEST_PORT || 8765 });

server.on('connection', (socket) => {
  socket.send(JSON.stringify({ type: 'welcome' }));
  socket.on('message', (data) => socket.send(data)); // echo
});

server.on('listening', () => {
  process.send && process.send('ready');
  console.log(`WS echo server on port ${server.address().port}`);
});

process.on('SIGTERM', () => server.close(() => process.exit(0)));

name: WebSocket Tests

on: [push, pull_request]

jobs:
  ws-tests:
    runs-on: ubuntu-latest

    services:
      app:
        image: your-app:latest
        ports:
          - 8080:8080
        options: >-
          --health-cmd "curl -f http://localhost:8080/health || exit 1"
          --health-interval 5s
          --health-timeout 3s
          --health-retries 10

    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - name: Install test dependencies
        run: pip install -r requirements-test.txt

      - name: Run WebSocket functional tests
        run: pytest tests/websocket/ -v --timeout=30
        env:
          WS_URL: ws://localhost:8080/ws

      - name: Install Playwright browsers
        run: npx playwright install --with-deps chromium

      - name: Run Playwright WebSocket E2E tests
        run: npx playwright test tests/e2e/websocket/

      - name: Run k6 WebSocket smoke test (on PR)
        if: github.event_name == 'pull_request'
        uses: grafana/k6-action@v0.3.1
        with:
          filename: k6/ws-smoke.js
          flags: --vus 5 --duration 30s
        env:
          WS_URL: ws://localhost:8080/ws

  ws-soak:
    # Run the 10-minute soak on a schedule, not on every PR
    if: github.event_name == 'schedule'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: grafana/k6-action@v0.3.1
        with:
          filename: k6/chat-soak.js
        env:
          WS_URL: ${{ secrets.STAGING_WS_URL }}
          WS_TOKEN: ${{ secrets.STAGING_WS_TOKEN }}