Testing Go SSE Handlers with httptest

Part of Go Streaming Patterns, under Backend Stream Generation & Connection Management.

The default testing tools assume a handler that returns. httptest.NewRecorder never sees a flush and ResponseRecorder has no client to disconnect, so the two properties that matter most for a stream — incremental delivery and clean teardown — are exactly the ones the obvious test cannot check.

Symptom & Developer Intent

  • A test using httptest.NewRecorder hangs forever because the handler never returns.
  • Tests pass while production streams deliver nothing, because the missing Flush() is invisible to a recorder.
  • Disconnect handling is untested, so goroutine leaks reach production.
  • Tests are slow because each one waits on real timers.
  • Reading the response body blocks until the handler finishes, which for a stream is never.

The intent is a test suite that asserts frames arrive incrementally, that a disconnect tears everything down, and that no goroutines survive.

Root Cause Analysis

httptest.NewRecorder returns a *ResponseRecorder, which buffers everything written and exposes it only after the handler returns. It does implement http.Flusher, but flushing writes into the same buffer — so a handler that forgets to flush produces an identical recording to one that flushes correctly. The single most common Go SSE bug is therefore invisible to the most common Go test.

What each test harness can actually observe Comparison of httptest.NewRecorder and httptest.NewServer across incremental delivery, flush visibility and client disconnect. What each test harness can actually observe NewRecorder NewServer Incremental reads no — buffered yes — real reader Missing Flush() visible no yes — read times out Client disconnect no client exists cancel the request Speed instant loopback, fast
The recorder cannot distinguish a handler that flushes from one that does not, which is precisely the bug Go SSE handlers most often have.

It also has no notion of a client going away. The request context is whatever the test supplied, so a handler whose teardown hangs off r.Context().Done() never exercises that path.

httptest.NewServer fixes both. It runs a real HTTP server on a loopback port, so the response is a real io.Reader that yields bytes as they are flushed, and cancelling the client request produces a genuine context cancellation in the handler.

Step-by-Step Resolution

The shape of a streaming test Test pipeline: start a real server, connect, publish after connecting, read one frame with a deadline, then cancel and assert teardown. The shape of a streaming test goleak then proves the writer goroutine actually exited NewServer real loopback Connect assert content type Publish + read with a deadline Cancel assert hub.Len()==0 per test after connect teardown
Publishing after connecting is the detail that makes the assertion meaningful — a frame published earlier could have been buffered rather than flushed.

Step 1 — Use a real server and read incrementally 

func TestStreamDeliversIncrementally(t *testing.T) {
    hub := NewHub()
    srv := httptest.NewServer(http.HandlerFunc(streamHandler(hub)))
    defer srv.Close()

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        t.Fatalf("connect: %v", err)
    }
    defer resp.Body.Close()

    if got := resp.Header.Get("Content-Type"); !strings.HasPrefix(got, "text/event-stream") {
        t.Fatalf("content type = %q", got)
    }

    // Publish AFTER connecting, so the frame can only arrive if the handler flushed.
    hub.Publish(Event{ID: "1", Data: `{"tick":1}`})

    reader := bufio.NewReader(resp.Body)
    frame := readFrame(t, reader, 2*time.Second)

    if !strings.Contains(frame, `"tick":1`) {
        t.Fatalf("frame = %q", frame)
    }
}

Step 2 — Read one frame at a time, with a deadline

A stream never ends, so every read needs a timeout or the test hangs on failure.

// readFrame reads up to the blank line that terminates one event block.
func readFrame(t *testing.T, r *bufio.Reader, timeout time.Duration) string {
    t.Helper()

    type result struct {
        s   string
        err error
    }
    ch := make(chan result, 1)

    go func() {
        var b strings.Builder
        for {
            line, err := r.ReadString('\n')
            if err != nil {
                ch <- result{b.String(), err}
                return
            }
            if line == "\n" || line == "\r\n" { // blank line: frame complete
                ch <- result{b.String(), nil}
                return
            }
            b.WriteString(line)
        }
    }()

    select {
    case res := <-ch:
        if res.err != nil {
            t.Fatalf("read frame: %v", res.err)
        }
        return res.s
    case <-time.After(timeout):
        t.Fatal("timed out waiting for a frame — handler probably did not Flush()")
        return ""
    }
}

The timeout message is the point: a handler missing Flush() fails here with an explanation instead of hanging.

Step 3 — Assert the flush explicitly with a recording writer

For unit-level tests without a server, wrap the writer and count flushes.

type flushRecorder struct {
    *httptest.ResponseRecorder
    flushes int
}

func (f *flushRecorder) Flush() {
    f.flushes++
    f.ResponseRecorder.Flush()
}

func TestHandlerFlushesEveryEvent(t *testing.T) {
    rec := &flushRecorder{ResponseRecorder: httptest.NewRecorder()}
    ctx, cancel := context.WithCancel(context.Background())
    req := httptest.NewRequest(http.MethodGet, "/events", nil).WithContext(ctx)

    hub := NewHub()
    done := make(chan struct{})
    go func() { streamHandler(hub)(rec, req); close(done) }()

    hub.Publish(Event{ID: "1", Data: "a"})
    hub.Publish(Event{ID: "2", Data: "b"})
    time.Sleep(20 * time.Millisecond)

    cancel()
    <-done

    if rec.flushes < 2 {
        t.Fatalf("flushes = %d, want >= 2 (one per event)", rec.flushes)
    }
}

Step 4 — Test disconnect and teardown 

func TestDisconnectUnregistersClient(t *testing.T) {
    hub := NewHub()
    srv := httptest.NewServer(http.HandlerFunc(streamHandler(hub)))
    defer srv.Close()

    ctx, cancel := context.WithCancel(context.Background())
    req, _ := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL, nil)
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        t.Fatal(err)
    }

    waitFor(t, func() bool { return hub.Len() == 1 }, time.Second)

    cancel()          // the client goes away: ctx.Done() fires inside the handler
    resp.Body.Close()

    waitFor(t, func() bool { return hub.Len() == 0 }, time.Second)
}

func waitFor(t *testing.T, cond func() bool, d time.Duration) {
    t.Helper()
    deadline := time.Now().Add(d)
    for time.Now().Before(deadline) {
        if cond() {
            return
        }
        time.Sleep(5 * time.Millisecond)
    }
    t.Fatalf("condition not met within %v", d)
}

Step 5 — Fail the suite on leaked goroutines 

func TestMain(m *testing.M) {
    // A leaked writer goroutine per departed client is the classic SSE leak.
    defer goleak.VerifyTestMain(m)
    os.Exit(m.Run())
}

Without a leak check, a handler that forgets to return on cancellation passes every functional test and leaks in production.

Validation & Monitoring

Which test catches which bug Table mapping the three common Go SSE bugs to the assertion that catches each. Which test catches which bug Caught by Symptom without it Missing Flush() read deadline silent stream in prod Client not unregistered hub.Len() == 0 memory grows Writer goroutine leak goleak OOM after days
Three bugs, three different assertions — and the registry check alone would pass a handler that leaks a goroutine on every disconnect.
# Run with the race detector: hub registries are shared across goroutines.
go test -race ./...

# Confirm tests are fast — a slow suite means real timers are being waited on.
go test -race -count=1 ./... -v | grep -E '^(ok|--- )'

# Confirm the flush assertion actually catches the bug: remove the Flush() call
# from the handler and check that TestStreamDeliversIncrementally fails.

That last step is worth doing once for real. A test that passes whether or not the handler flushes is worse than no test, because it creates confidence that does not exist.

For heartbeat and timeout behaviour, inject the clock rather than sleeping:

type Clock interface {
    Now() time.Time
    NewTicker(d time.Duration) *time.Ticker
}
// Production passes a real clock; tests pass a fake and advance it instantly.

Production Checklist

Frequently Asked Questions

Why does my test hang when I use httptest.NewRecorder?

Because the handler never returns — it is a stream — and the recorder exposes its body only after the handler completes. Use httptest.NewServer so the response body is a real reader that yields flushed bytes, and cancel the request context to end the handler.

How do I assert that the handler actually flushed?

Two ways. At integration level, publish after connecting and read with a deadline: an unflushed frame never arrives and the test fails on timeout. At unit level, wrap the recorder in a type whose Flush() increments a counter and assert one flush per event.

Do I need goleak, or is the registry assertion enough?

Both, because they catch different bugs. The registry assertion proves the client was unregistered; goleak proves the writer goroutine actually exited. A handler can unregister and still leave a goroutine blocked on a channel forever, and only the leak detector sees that.

How do I test heartbeat behaviour without waiting fifteen seconds?

Inject the clock. Take a small interface providing Now and NewTicker, pass a real implementation in production and a fake one in tests, and advance the fake instantly. Sleeping for real intervals makes the suite slow and flaky on loaded CI machines.