diff --git a/internal/httpserver/sse_e2e_test.go b/internal/httpserver/sse_e2e_test.go new file mode 100644 index 0000000..eb1d253 --- /dev/null +++ b/internal/httpserver/sse_e2e_test.go @@ -0,0 +1,145 @@ +package httpserver + +import ( + "bufio" + "context" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "vetting/internal/api" + "vetting/internal/db" + "vetting/internal/events" + "vetting/internal/model" + "vetting/internal/orchestrator" + "vetting/internal/store" +) + +// TestSSE_EndToEnd is the canary test the unit-level render tests miss: +// it mounts the real chi router (with RealIP + Recoverer + Logger +// middleware, the same stack that runs in prod), opens a GET /events, +// publishes an event via Runner.PublishTileUpdate, and asserts the +// payload lands on the wire as a well-formed SSE message. +// +// Motivation: the user reported that after cdd6cae "the pipeline does +// not update without a manual refresh". Template-only tests can't catch +// that — they only check rendered HTML. This test fails if any layer +// between publish and browser (hub subscribe, writeSSE formatting, +// middleware ResponseWriter wrapping that strips Flusher) is broken. +func TestSSE_EndToEnd(t *testing.T) { + conn, err := db.Open(filepath.Join(t.TempDir(), "vetting.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + + hosts := &store.Hosts{DB: conn} + runs := &store.Runs{DB: conn} + stages := &store.Stages{DB: conn} + diffs := &store.SpecDiffs{DB: conn} + arts := &store.Artifacts{DB: conn} + hub := events.NewHub() + runner := &orchestrator.Runner{Runs: runs, Hosts: hosts, Stages: stages, EventHub: hub} + tiles := &api.TileEnricher{Runs: runs, Artifacts: arts, SpecDiffs: diffs} + + // The prod wiring: closures that render tile + pipeline fragments + // for publishTileUpdate. Without these, publishTileUpdate is a no-op + // except for the bare tile event. + orchestrator.TileRenderer = func(_ context.Context, _ model.Host, _ *model.Run) string { + return "
tile
" + } + orchestrator.PipelineRenderer = func(_ *model.Run, _ []model.Stage) string { + return "
pipeline
" + } + + ui := &api.UI{ + Hosts: hosts, Runs: runs, Stages: stages, SpecDiffs: diffs, Artifacts: arts, + EventHub: hub, Runner: runner, Tiles: tiles, + } + agent := &api.Agent{ + Hosts: hosts, Runs: runs, Stages: stages, Artifacts: arts, + SpecDiffs: diffs, Runner: runner, EventHub: hub, + } + + hostID, err := hosts.Create(context.Background(), model.Host{ + Name: "sse-e2e", MAC: "aa:bb:cc:dd:ee:77", + WoLBroadcastIP: "10.0.0.255", WoLPort: 9, + }) + if err != nil { + t.Fatalf("create host: %v", err) + } + + router := NewRouter(Deps{UI: ui, Agent: agent}) + srv := httptest.NewServer(router) + t.Cleanup(srv.Close) + + // Open the SSE stream. + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+"/events", nil) + if err != nil { + t.Fatalf("new request: %v", err) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("GET /events: %v", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } + if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") { + t.Fatalf("Content-Type = %q, want text/event-stream", ct) + } + + reader := bufio.NewReader(resp.Body) + + // Consume the hello preamble first. + if err := waitForSSEEvent(reader, "hello", 1*time.Second); err != nil { + t.Fatalf("hello preamble: %v", err) + } + + // Give the Subscribe goroutine a moment to land in the map so our + // publish doesn't beat it. This is the race the hub's select + // default-drops on — if the subscriber isn't registered yet, the + // publish is silently thrown away. + time.Sleep(50 * time.Millisecond) + + // Publish a tile-update via Runner. This is what the real orchestrator + // does on every state transition. + runner.PublishTileUpdate(context.Background(), hostID) + + // The browser should see one SSE message with `event: tile-`. + wantName := "tile-1" // first host → ID 1 + _ = hostID + if err := waitForSSEEvent(reader, wantName, 2*time.Second); err != nil { + t.Fatalf("tile publish: %v", err) + } +} + +// waitForSSEEvent reads lines from the SSE stream until it sees +// `event: ` or the deadline elapses. It tolerates interleaved +// heartbeats because the hub's 15s heartbeat can fire between the +// hello and the event we're looking for. +func waitForSSEEvent(r *bufio.Reader, name string, timeout time.Duration) error { + deadline := time.Now().Add(timeout) + want := "event: " + name + for time.Now().Before(deadline) { + line, err := r.ReadString('\n') + if err != nil { + return err + } + if strings.TrimRight(line, "\r\n") == want { + return nil + } + } + return &timeoutErr{} +} + +type timeoutErr struct{} + +func (e *timeoutErr) Error() string { return "timeout waiting for sse event" } diff --git a/internal/web/templates/layout.templ b/internal/web/templates/layout.templ index 1641f51..c9acdb2 100644 --- a/internal/web/templates/layout.templ +++ b/internal/web/templates/layout.templ @@ -9,7 +9,7 @@ templ Layout(title string) { { title } — Vetting - +
diff --git a/internal/web/templates/layout_templ.go b/internal/web/templates/layout_templ.go index 1dfc9aa..142fd91 100644 --- a/internal/web/templates/layout_templ.go +++ b/internal/web/templates/layout_templ.go @@ -42,7 +42,7 @@ func Layout(title string) templ.Component { if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err } - templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 2, " — Vetting
Vetting
·
") + templ_7745c5c3_Err = templruntime.WriteString(templ_7745c5c3_Buffer, 2, " — Vetting
Vetting
·
") if templ_7745c5c3_Err != nil { return templ_7745c5c3_Err }