Testing Real-Time Apps

import test, { before, after } from "node:test";
import assert from "node:assert/strict";
import { io } from "socket.io-client";
import { startServer } from "../src/server.js";
import { frame, id } from "../support/helpers.js";

let server;
const clients = [];
before(async () => { server = await startServer(); });
after(async () => { clients.forEach((c) => c.disconnect()); await server.close(); });

async function connect(user) {
  const socket = io(`http://127.0.0.1:${server.port}`, { transports: ["websocket"], auth: { user }, reconnection: false, forceNew: true });
  clients.push(socket);
  await new Promise((resolve, reject) => { socket.once("connect", resolve); socket.once("connect_error", reject); });
  return socket;
}
const once = (socket, event) => new Promise((resolve) => socket.once(event, resolve));

test("the middleware rejects a connection without credentials", async () => {
  const socket = io(`http://127.0.0.1:${server.port}`, { transports: ["websocket"], reconnection: false, forceNew: true });
  clients.push(socket);
  const err = await new Promise((resolve) => socket.once("connect_error", resolve));
  assert.equal(err.message, "unauthorized");
});

test("ack, broadcast to the room and duplicate detection", async () => {
  const [alice, bob] = await Promise.all([connect("alice"), connect("bob")]);
  await alice.emitWithAck("chatterbox", frame("join", "general", { since: null }, id()));
  await bob.emitWithAck("chatterbox", frame("join", "general", { since: null }, id()));

  const received = once(bob, "chatterbox");   // register BEFORE sending
  const msg = frame("message", "general", { text: "Hello, Chatterbox" }, id());
  const ack = await alice.emitWithAck("chatterbox", msg);
  assert.deepEqual([ack.type, ack.payload.seq, ack.payload.dup], ["ack", 1, false]);
  assert.equal((await received).payload.text, "Hello, Chatterbox");

  const dup = await alice.emitWithAck("chatterbox", msg);
  assert.deepEqual([dup.payload.seq, dup.payload.dup], [1, true]);
});

import test, { before, after } from "node:test";
import assert from "node:assert/strict";
import WebSocket from "ws";
import { startServer } from "../src/server.js";
import { frame, id } from "../support/helpers.js";

let server;
before(async () => { server = await startServer(); });
after(() => server.close());

/** Opens a socket and returns { ws, next(type) } where next() resolves with the next frame of that type. */
async function open(user) {
  const ws = new WebSocket(`ws://127.0.0.1:${server.port}/ws?user=${user}`);
  const queue = [];
  const waiters = [];
  ws.on("message", (data) => {
    queue.push(JSON.parse(data.toString()));
    waiters.splice(0).forEach((w) => w());
  });
  await new Promise((resolve, reject) => { ws.once("open", resolve); ws.once("error", reject); });
  const next = async (type, ms = 2000) => {
    const deadline = Date.now() + ms;
    for (;;) {
      const i = queue.findIndex((e) => e.type === type);
      if (i >= 0) return queue.splice(i, 1)[0];
      const left = deadline - Date.now();
      if (left <= 0) throw new Error(`timed out waiting for a ${type} frame`);
      await new Promise((resolve) => { waiters.push(resolve); setTimeout(resolve, left); });
    }
  };
  return { ws, next, send: (env) => ws.send(JSON.stringify(env)) };
}

test("a message is acknowledged to the sender and broadcast to the room", async () => {
  const alice = await open("alice");
  const bob = await open("bob");
  for (const c of [alice, bob]) { c.send(frame("join", "general", { since: null }, id())); await c.next("ack"); }

  alice.send(frame("message", "general", { text: "Hello, Chatterbox" }, id()));
  assert.equal((await alice.next("ack")).payload.seq, 1);
  assert.equal((await bob.next("message")).payload.text, "Hello, Chatterbox");
  alice.ws.close(); bob.ws.close();
});

test("a malformed frame closes the socket with 4400", async () => {
  const c = await open("carol");
  const closed = new Promise((resolve) => c.ws.once("close", (code) => resolve(code)));
  c.ws.send("not json");
  assert.equal(await closed, 4400);
});

test("the upgrade without credentials is refused with 401", async () => {
  const ws = new WebSocket(`ws://127.0.0.1:${server.port}/ws`);
  const status = await new Promise((resolve) => ws.once("unexpected-response", (_req, res) => resolve(res.statusCode)));
  assert.equal(status, 401);
});

{
  "name": "chatterbox-tests",
  "private": true,
  "type": "module",
  "scripts": { "test": "node --test" },
  "dependencies": { "socket.io": "4.8.4", "ws": "8.22.0" },
  "devDependencies": { "socket.io-client": "4.8.4" }
}

using Microsoft.AspNetCore.SignalR;

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddSignalR();
builder.Services.AddSingleton<MessageLog>();
var app = builder.Build();
app.MapHub<ChatHub>("/chat");
app.Run();

public sealed class MessageLog
{
    private readonly Dictionary<string, int> _seq = new();
    public int Next(string room) { lock (_seq) { _seq[room] = _seq.GetValueOrDefault(room) + 1; return _seq[room]; } }
}

public sealed class ChatHub(MessageLog log) : Hub
{
    public Task Join(string room) => Groups.AddToGroupAsync(Context.ConnectionId, room);

    public async Task<int> Send(string room, string text)
    {
        if (text.Length is 0 or > 4000) throw new HubException("bad_request");
        var seq = log.Next(room);
        await Clients.Group(room).SendAsync("message", new { room, text, seq });
        return seq;
    }
}

public partial class Program;   // makes Program visible to WebApplicationFactory<Program>

using Microsoft.AspNetCore.SignalR;
using Microsoft.AspNetCore.Http.Connections;
using Microsoft.AspNetCore.Mvc.Testing;
using Microsoft.AspNetCore.SignalR.Client;
using Xunit;

public class ChatHubTests(WebApplicationFactory<Program> factory) : IClassFixture<WebApplicationFactory<Program>>
{
    // A HubConnection that talks to the in-memory TestServer through a real WebSocket.
    private async Task<HubConnection> ConnectAsync()
    {
        var server = factory.Server;
        var connection = new HubConnectionBuilder()
            .WithUrl(new Uri(server.BaseAddress, "/chat"), o =>
            {
                o.Transports = HttpTransportType.WebSockets;
                o.SkipNegotiation = true;                                   // WebSockets only: no negotiate request needed
                o.HttpMessageHandlerFactory = _ => server.CreateHandler();
                o.WebSocketFactory = async (context, ct) =>
                    await server.CreateWebSocketClient().ConnectAsync(context.Uri, ct);
            })
            .Build();
        await connection.StartAsync();
        return connection;
    }

    [Fact]
    public async Task A_message_reaches_the_room_with_a_sequence_number()
    {
        await using var alice = await ConnectAsync();
        await using var bob = await ConnectAsync();
        var received = new TaskCompletionSource<(string Room, string Text, int Seq)>();
        bob.On<System.Text.Json.JsonElement>("message", m =>
            received.TrySetResult((m.GetProperty("room").GetString()!, m.GetProperty("text").GetString()!, m.GetProperty("seq").GetInt32())));
        await alice.InvokeAsync("Join", "general");
        await bob.InvokeAsync("Join", "general");

        var seq = await alice.InvokeAsync<int>("Send", "general", "Hello, Chatterbox");

        var message = await received.Task.WaitAsync(TimeSpan.FromSeconds(5));
        Assert.Equal(1, seq);
        Assert.Equal(("general", "Hello, Chatterbox", 1), message);
    }

    [Fact]
    public async Task An_invalid_message_surfaces_as_a_HubException_on_the_client()
    {
        await using var alice = await ConnectAsync();
        var error = await Assert.ThrowsAsync<HubException>(() => alice.InvokeAsync<int>("Send", "general", ""));
        Assert.Contains("bad_request", error.Message);
    }
}

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <TargetFramework>net10.0</TargetFramework>
    <Nullable>enable</Nullable>
    <ImplicitUsings>enable</ImplicitUsings>
    <IsPackable>false</IsPackable>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="*" />
    <PackageReference Include="xunit" Version="*" />
    <PackageReference Include="xunit.runner.visualstudio" Version="*" />
    <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" Version="10.0.*" />
    <PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="10.0.*" />
  </ItemGroup>
  <ItemGroup>
    <ProjectReference Include="../App/App.csproj" />
  </ItemGroup>
</Project>

<?php
// client_test.php - tests with the Swoole coroutine client; run with `php client_test.php`.
use Swoole\Coroutine;
use Swoole\Coroutine\Http\Client;

use function Swoole\Coroutine\run;

function frame(string $type, string $room, array $payload, string $id): string
{
    return json_encode(['v' => 1, 'type' => $type, 'id' => $id, 'room' => $room, 'ts' => 0, 'payload' => $payload]);
}

function connect(string $host, int $port): Client
{
    $client = new Client($host, $port);
    if (!$client->upgrade('/')) {
        throw new RuntimeException("upgrade failed: {$client->errCode}");
    }
    return $client;
}

/** Reads frames until one has the wanted type; a read that takes longer than $timeout seconds fails. */
function next_of(Client $c, string $type, float $timeout = 2.0): array
{
    $deadline = microtime(true) + $timeout;
    while (($left = $deadline - microtime(true)) > 0) {
        $f = $c->recv($left);
        if ($f === false || $f === '') {
            break;
        }
        $env = json_decode($f->data, true);
        if ($env['type'] === $type) {
            return $env;
        }
    }
    throw new RuntimeException("no $type frame within {$timeout}s");
}

function check(bool $ok, string $label): void
{
    echo ($ok ? 'PASS ' : 'FAIL ') . $label . PHP_EOL;
    if (!$ok) {
        exit(1);
    }
}

run(function () {
    $host = getenv('CHAT_HOST') ?: '127.0.0.1';
    $alice = connect($host, 9501);
    $bob = connect($host, 9501);
    foreach ([$alice, $bob] as $i => $c) {
        $c->push(frame('join', 'general', ['since' => null], "join$i"));
        next_of($c, 'ack');
    }

    $msg = frame('message', 'general', ['text' => 'Hello, Chatterbox'], 'm1');
    $alice->push($msg);
    $ack = next_of($alice, 'ack');
    check($ack['payload']['seq'] === 1 && $ack['payload']['dup'] === false, 'message is acknowledged with seq 1');
    check(next_of($bob, 'message')['payload']['text'] === 'Hello, Chatterbox', 'bob receives the broadcast');

    $alice->push($msg);
    $dup = next_of($alice, 'ack');
    check($dup['payload']['dup'] === true && $dup['payload']['seq'] === 1, 'the same id is a duplicate');

    $bob->push(frame('message', 'secret', ['text' => 'hi'], 'm2'));
    check(next_of($bob, 'error')['payload']['code'] === 'forbidden', 'a non-member gets forbidden');
    $alice->close();
    $bob->close();
});

import { defineConfig } from "@playwright/test";

export default defineConfig({
  testDir: "./tests",
  timeout: 30_000,
  webServer: { command: "node server.js", url: "http://127.0.0.1:3100/", reuseExistingServer: false },
  use: {
    baseURL: "http://127.0.0.1:3100",
    launchOptions: {
      args: ["--use-fake-device-for-media-stream", "--use-fake-ui-for-media-stream"],
    },
    trace: "retain-on-failure",
  },
});

{
  "name": "chatterbox-e2e",
  "private": true,
  "type": "module",
  "scripts": { "test": "playwright test" },
  "dependencies": { "ws": "8.22.0" },
  "devDependencies": { "@playwright/test": "1.64.1" }
}

// server.js - static files, the Chatterbox subset on /ws and a raw signalling relay on /signal.
import { createServer } from "node:http";
import { readFile } from "node:fs/promises";
import { WebSocketServer } from "ws";
import { createState, handle } from "./handler.js";

const PORT = Number(process.env.PORT ?? 3100);
const state = createState();
const chat = new WebSocketServer({ noServer: true });
const signal = new WebSocketServer({ noServer: true });
const users = new Map();      // chat socket -> user
const roomSockets = new Map(); // room -> Set of chat sockets
const peers = new Map();       // call room -> Set of signalling sockets

const http = createServer(async (req, res) => {
  const { pathname } = new URL(req.url, "http://x");
  const name = pathname === "/" ? "chat.html" : pathname.slice(1);
  try {
    const body = await readFile(new URL(`./public/${name}`, import.meta.url));
    res.writeHead(200, { "content-type": name.endsWith(".html") ? "text/html" : "text/plain" }).end(body);
  } catch { res.writeHead(404).end("not found"); }
});

http.on("upgrade", (req, socket, head) => {
  const url = new URL(req.url, "http://x");
  const wss = url.pathname === "/ws" ? chat : url.pathname === "/signal" ? signal : null;
  if (!wss) return socket.destroy();
  wss.handleUpgrade(req, socket, head, (ws) => {
    ws.query = url.searchParams;
    wss.emit("connection", ws);
  });
});

chat.on("connection", (ws) => {
  users.set(ws, ws.query.get("user"));
  ws.on("message", (data) => {
    const env = JSON.parse(data.toString());
    const { replies, broadcast } = handle(state, users.get(ws), env);
    if (env.type === "join") (roomSockets.get(env.room) ?? roomSockets.set(env.room, new Set()).get(env.room)).add(ws);
    replies.forEach((r) => ws.send(JSON.stringify(r)));
    broadcast.forEach((b) => roomSockets.get(b.room)?.forEach((c) => c.send(JSON.stringify(b.envelope))));
  });
  ws.on("close", () => users.delete(ws));
});

signal.on("connection", (ws) => {
  const room = ws.query.get("room");
  const set = peers.get(room) ?? peers.set(room, new Set()).get(room);
  ws.send(JSON.stringify({ type: "peers", count: set.size }));  // the second arrival creates the offer
  set.add(ws);
  ws.on("message", (data) => set.forEach((c) => c !== ws && c.send(data.toString())));
  ws.on("close", () => set.delete(ws));
});

http.listen(PORT, "127.0.0.1", () => console.log(`listening on ${PORT}`));

<!doctype html>
<meta charset="utf-8">
<title>Chatterbox test page</title>
<ul id="log" aria-label="messages"></ul>
<form id="form"><input id="text" aria-label="Message"><button>Send</button></form>
<script>
  const user = new URLSearchParams(location.search).get("user");
  const ws = new WebSocket(`ws://${location.host}/ws?user=${user}`);
  const newId = () => crypto.randomUUID().replaceAll("-", "").slice(0, 26);
  let joinId;
  const send = (type, payload) => {
    const id = newId();
    ws.send(JSON.stringify({ v: 1, type, id, room: "general", ts: 0, payload }));
    return id;
  };
  ws.onopen = () => { joinId = send("join", { since: null }); };
  ws.onmessage = (e) => {
    const env = JSON.parse(e.data);
    if (env.type === "ack" && env.payload.ref === joinId) document.body.dataset.joined = "true";
    if (env.type !== "message") return;
    const li = document.createElement("li");
    li.textContent = `${env.payload.from.userId}: ${env.payload.text}`;
    log.append(li);
  };
  setInterval(() => ws.readyState === WebSocket.OPEN && send("presence", { status: "online" }), 20000);  // application heartbeat
  form.onsubmit = (e) => { e.preventDefault(); send("message", { text: text.value }); text.value = ""; };
</script>

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

test("the page sends and receives Chatterbox envelopes", async ({ page }) => {
  const sent = [];
  const received = [];
  page.on("websocket", (ws) => {
    ws.on("framesent", (f) => sent.push(JSON.parse(f.payload)));
    ws.on("framereceived", (f) => received.push(JSON.parse(f.payload)));
  });

  await page.goto("/?user=alice");
  await expect(page.locator("body")).toHaveAttribute("data-joined", "true");
  await page.getByRole("textbox", { name: "Message" }).fill("Hello, Chatterbox");
  await page.getByRole("button", { name: "Send" }).click();

  await expect(page.getByRole("list", { name: "messages" })).toContainText("alice: Hello, Chatterbox");
  const message = sent.find((e) => e.type === "message");
  expect(message).toMatchObject({ v: 1, room: "general", payload: { text: "Hello, Chatterbox" } });
  const ack = received.find((e) => e.type === "ack" && e.payload.ref === message.id);
  expect(ack.payload).toMatchObject({ ok: true, seq: expect.any(Number), dup: false });
});

test("two browser contexts are two users", async ({ browser }) => {
  const alice = await (await browser.newContext()).newPage();
  const bob = await (await browser.newContext()).newPage();
  await Promise.all([alice.goto("/?user=alice"), bob.goto("/?user=bob")]);
  for (const page of [alice, bob]) await expect(page.locator("body")).toHaveAttribute("data-joined", "true");

  await alice.getByRole("textbox", { name: "Message" }).fill("Anyone there?");
  await alice.getByRole("button", { name: "Send" }).click();
  await expect(bob.getByRole("list", { name: "messages" })).toContainText("alice: Anyone there?");
});

<!doctype html>
<meta charset="utf-8">
<title>Huddle test page</title>
<video id="local" autoplay muted playsinline></video><video id="remote" autoplay playsinline></video>
<p id="state">idle</p>
<script>
  const room = new URLSearchParams(location.search).get("room");
  const pc = new RTCPeerConnection();                      // host candidates only: both pages are on one machine
  const sig = new WebSocket(`ws://${location.host}/signal?room=${room}`);
  const send = (m) => sig.send(JSON.stringify(m));
  pc.onicecandidate = (e) => e.candidate && send({ candidate: e.candidate });
  sig.onopen = () => { state.textContent = "waiting"; };
  pc.ontrack = (e) => { remote.srcObject = e.streams[0]; };
  pc.onconnectionstatechange = () => { state.textContent = pc.connectionState; };
  const media = navigator.mediaDevices.getUserMedia({ video: true, audio: true }).then((stream) => {
    local.srcObject = stream;
    stream.getTracks().forEach((t) => pc.addTrack(t, stream));
  });
  sig.onmessage = async (e) => {
    const m = JSON.parse(e.data);
    await media;
    if (m.type === "peers" && m.count > 0) {                // I arrived second: I make the offer
      await pc.setLocalDescription(await pc.createOffer());
      send({ description: pc.localDescription });
    } else if (m.description) {
      await pc.setRemoteDescription(m.description);
      if (m.description.type === "offer") {
        await pc.setLocalDescription(await pc.createAnswer());
        send({ description: pc.localDescription });
      }
    } else if (m.candidate) {
      await pc.addIceCandidate(m.candidate);
    }
  };
  window.inboundVideoBytes = async () => {
    const report = await pc.getStats();
    let bytes = 0;
    report.forEach((s) => { if (s.type === "inbound-rtp" && s.kind === "video") bytes += s.bytesReceived; });
    return bytes;
  };
</script>

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

test("two contexts get a call with media flowing both ways", async ({ browser }) => {
  const first = await (await browser.newContext()).newPage();
  await first.goto("/huddle.html?room=daily");
  await expect(first.locator("#state")).toHaveText("waiting");   // signalling is open before the second page arrives

  const second = await (await browser.newContext()).newPage();
  await second.goto("/huddle.html?room=daily");

  for (const page of [first, second]) {
    await expect(page.locator("#state")).toHaveText("connected");
    await expect.poll(() => page.evaluate(() => window.inboundVideoBytes()), { timeout: 15_000 }).toBeGreaterThan(0);
  }
});

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

test("the client sends a presence heartbeat every 20 seconds, without waiting 20 seconds", async ({ page }) => {
  const heartbeats = [];
  page.on("websocket", (ws) =>
    ws.on("framesent", (f) => {
      const env = JSON.parse(f.payload);
      if (env.type === "presence") heartbeats.push(env);
    }));

  await page.clock.install();                 // before goto: the page's timers are now fake
  await page.goto("/?user=carol");
  await expect(page.locator("body")).toHaveAttribute("data-joined", "true");
  expect(heartbeats).toHaveLength(0);

  await page.clock.fastForward(61_000);       // three heartbeat intervals in no time
  await expect.poll(() => heartbeats.length).toBeGreaterThanOrEqual(1);
  expect(heartbeats[0]).toMatchObject({ type: "presence", payload: { status: "online" } });
});