WebSocket and SSE in tests

TestClient drives two kinds of long-lived responses in-process (since 18.29.0): a WebSocket route's upgrade handler, over an in-memory duplex, and a text/event-stream body, read as it arrives. Both wait with timeouts, so a handler that never answers fails the test instead of hanging it.

WebSocket#

client.websocket(path).connect() sends the opening handshake through the app. When the route answers with WebSocket::upgrade's 101, the handler runs on a thread of its own — the same handler the server would run — and you get a TestWebSocket:

use rust_web_server::app::App;
use rust_web_server::test_client::TestClient;
use rust_web_server::websocket::{Message, WebSocket};

#[test]
fn echo() {
    let app = App::with_state(()).get("/echo", |request, _params, _conn, _state| {
        WebSocket::upgrade(request, |mut ws| {
            while let Ok(Some(Message::Text(text))) = ws.recv() {
                let _ = ws.send_text(&text);
            }
        })
    });
    let client = TestClient::new(app);

    let mut ws = client.websocket("/echo").connect().expect("a 101");
    ws.send_text("hi");
    assert_eq!("hi", ws.recv_text());
    assert_eq!(Some(1000), ws.close(1000, "done")); // the handler's close code
}

Anything but an upgrade comes back as the Err response, so refusals are testable too:

let refused = client.websocket("/admin/live").connect().err().expect("refused");
assert_eq!(401, refused.status());

Dropping a TestWebSocket ends the connection: the handler's recv returns an error. A handler that reaches for tokio::runtime::Handle::current() finds the test's runtime when the test is #[tokio::test].

Server-Sent Events#

TestResponse::sse() reads a text/event-stream body — streamed by Sse::stream (stream_pipe) as it arrives, or buffered by Sse::into_response at once — and parses it into SseFrame { event, data, id, retry }:

use std::time::Duration;
use rust_web_server::app::App;
use rust_web_server::sse::{Sse, SseEvent};
use rust_web_server::test_client::TestClient;

#[test]
fn ticks() {
    let app = App::with_state(()).get("/ticks", |_req, _params, _conn, _state| {
        let (events, response) = Sse::stream();
        std::thread::spawn(move || {
            for n in 1..=3 {
                let _ = events.send(SseEvent::data(n.to_string()).event_type("tick").id(n.to_string()));
            }
        });
        response
    });
    let client = TestClient::new(app);

    let events = client.get("/ticks").send().sse_events(3, Duration::from_secs(5));
    assert_eq!(3, events.len());
    assert_eq!("tick", events[0].event);
    assert_eq!("1", events[0].data);
    assert_eq!(Some("1"), events[0].id.as_deref());
}

sse_events(n, timeout) returns up to n events within timeout in all — fewer when the stream ends or time runs out, so assert on the count. For finer control keep the TestSse from .sse():

Parsing follows the HTML Standard's rules: lines end in CRLF, LF or CR; data lines join with \n; the event type defaults to message; a block without data isn't dispatched; retry must be digits. SseParser is the same parser on its own (parser.feed(bytes) returns the events those bytes completed). It refuses an event or line over 8 MiB (sse::MAX_EVENT_BYTES) instead of buffering without bound.

A streamed body is read on a thread of its own; it ends with the stream, or at the first read after the TestSse is dropped (for Sse::stream, the next event or keep-alive comment). TestResponse::is_streamed() tells whether a response has a streamed body, and take_stream() hands you its reader for anything that isn't SSE.