summercms/modules/surf/streaming_smoke_test.go

package surf

import (
	"bufio"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"git.golem15.com/golem15/summercms/modules/backpack"
	"git.golem15.com/golem15/summercms/modules/pact"
	"git.golem15.com/golem15/summercms/modules/party"
)

// Smoke test for the streaming route kind of phase 19.7 (RESEARCH Pitfall 1):
// a MountStream route flushes to the client while the handler still runs, a
// buffered route does not, and a panic is recovered without rewriting a
// response that has started.

// streamPlugin registers one buffered route; its stream routes are mounted
// through the mountPier hook, as the MCP module would.
type streamPlugin struct {
	release chan struct{}
}

func (streamPlugin) ID() string                   { return "acme.stream" }
func (streamPlugin) Requires() []string           { return nil }
func (streamPlugin) Register(*backpack.App) error { return nil }
func (streamPlugin) Boot(*backpack.App) error     { return nil }

func (p streamPlugin) events(w http.ResponseWriter, _ *http.Request) {
	w.Header().Set("Content-Type", "text/event-stream")
	_, _ = io.WriteString(w, "event: a\n\n")
	if f, ok := w.(http.Flusher); ok {
		f.Flush()
	}
	<-p.release
	_, _ = io.WriteString(w, "event: b\n\n")
}

func (p streamPlugin) Routes(r pact.Router) error {
	r.Get("/acme/buffered", p.events)
	return nil
}

// firstLine reads one line of the body in the background and reports it on
// the returned channel.
func firstLine(t *testing.T, res *http.Response) <-chan string {
	t.Helper()
	out := make(chan string, 1)
	go func() {
		line, err := bufio.NewReader(res.Body).ReadString('\n')
		if err != nil {
			line = "error: " + err.Error()
		}
		out <- strings.TrimSpace(line)
	}()
	return out
}

func TestStreamingRouteSmoke(t *testing.T) {
	plugin := streamPlugin{release: make(chan struct{})}
	panicAfterFlush := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = io.WriteString(w, "event: a\n\n")
		w.(http.Flusher).Flush()
		panic("after flush")
	})
	panicBeforeWrite := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {
		panic("before write")
	})
	mountPier = func(r *Router, _ *backpack.App, _ []party.Plugin) error {
		r.BindPlugin("summercms.pier")
		if err := r.MountStream("GET", "/mcp/events", http.HandlerFunc(plugin.events)); err != nil {
			return err
		}
		if err := r.MountStream("GET", "/mcp/panic-late", panicAfterFlush); err != nil {
			return err
		}
		if err := r.MountStream("GET", "/mcp/panic-early", panicBeforeWrite); err != nil {
			return err
		}
		// A stream route may not replace a plugin route.
		if err := r.MountStream("GET", "/acme/buffered", panicBeforeWrite); err == nil || !strings.Contains(err.Error(), "acme.stream") {
			t.Errorf("MountStream over a plugin route: %v", err)
		}
		return nil
	}
	defer func() { mountPier = nil }()

	app := transportApp(t, "development", map[string]string{
		"http.yaml": transportBodyLimits,
		"mcp.yaml":  "enabled: true\n",
	})
	h, err := Assemble(app, []party.Plugin{plugin})
	if err != nil {
		t.Fatal(err)
	}
	srv := httptest.NewServer(h)
	defer srv.Close()
	client := srv.Client()

	// The stream route delivers the first event before the handler returns.
	stream, err := client.Get(srv.URL + "/mcp/events")
	if err != nil {
		t.Fatal(err)
	}
	defer stream.Body.Close()
	if stream.StatusCode != http.StatusOK {
		t.Fatalf("stream status = %d", stream.StatusCode)
	}
	select {
	case line := <-firstLine(t, stream):
		if line != "event: a" {
			t.Fatalf("stream first line = %q", line)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("stream route did not flush its first event while the handler was running")
	}

	// The buffered route delivers nothing, not even its headers, until the
	// handler returns: the whole GET stays pending until the release.
	type bufferedResult struct {
		status int
		body   string
		err    error
	}
	bufferedDone := make(chan bufferedResult, 1)
	go func() {
		res, err := client.Get(srv.URL + "/acme/buffered")
		if err != nil {
			bufferedDone <- bufferedResult{err: err}
			return
		}
		defer res.Body.Close()
		body, err := io.ReadAll(res.Body)
		bufferedDone <- bufferedResult{status: res.StatusCode, body: string(body), err: err}
	}()
	select {
	case got := <-bufferedDone:
		t.Fatalf("buffered route answered before the handler returned: %+v", got)
	case <-time.After(300 * time.Millisecond):
	}
	close(plugin.release)
	select {
	case got := <-bufferedDone:
		if got.err != nil || got.status != http.StatusOK || got.body != "event: a\n\nevent: b\n\n" {
			t.Fatalf("buffered route after release = %+v", got)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("buffered route never answered")
	}
	rest, err := io.ReadAll(stream.Body)
	if err != nil || !strings.Contains(string(rest), "event: b") {
		t.Fatalf("stream rest = %q, %v", rest, err)
	}

	// A panic after the first flush ends the body; the 200 already sent stays.
	late, err := client.Get(srv.URL + "/mcp/panic-late")
	if err != nil {
		t.Fatal(err)
	}
	defer late.Body.Close()
	if late.StatusCode != http.StatusOK {
		t.Fatalf("panic after flush status = %d", late.StatusCode)
	}
	body, _ := io.ReadAll(late.Body)
	if !strings.HasPrefix(string(body), "event: a") || strings.Contains(string(body), "500") || strings.Contains(string(body), "after flush") {
		t.Fatalf("panic after flush body = %q", body)
	}

	// A panic before any write answers the opaque 500.
	early, err := client.Get(srv.URL + "/mcp/panic-early")
	if err != nil {
		t.Fatal(err)
	}
	defer early.Body.Close()
	body, _ = io.ReadAll(early.Body)
	if early.StatusCode != http.StatusInternalServerError || strings.Contains(string(body), "before write") {
		t.Fatalf("panic before write = %d %q", early.StatusCode, body)
	}
}