package web
import (
"bufio"
"context"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
"time"
"craftdesign.group/quarry/internal/api"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/ops"
)
// fakeChanges is the event plane a live page waits on, driven by the
// test: WatchNoun hands back a channel the test ticks by hand.
type fakeChanges struct {
ch chan struct{}
}
func (f *fakeChanges) WatchNoun(string, string) (<-chan struct{}, func()) {
return f.ch, func() {}
}
// countingEngine answers RunDetail differently on each call — the
// stand-in for a run that is actually moving between two polls of the
// same handler.
type countingEngine struct {
mu sync.Mutex
calls int
first ci.Run
next ci.Run
}
func (e *countingEngine) RunDetail(string) (ci.Run, error) {
e.mu.Lock()
defer e.mu.Unlock()
e.calls++
if e.calls == 1 {
return e.first, nil
}
return e.next, nil
}
func (e *countingEngine) StepLogs(_, _ string, _ int, _ bool) ([]ci.LogChunk, error) { return nil, nil }
func (e *countingEngine) Workers() ([]ci.Worker, error) { return nil, nil }
func (e *countingEngine) WorkerTokens() ([]ci.WorkerToken, error) { return nil, nil }
func (e *countingEngine) MintWorkerToken(string, []string) (ci.WorkerToken, error) {
return ci.WorkerToken{}, nil
}
func (e *countingEngine) RevokeWorkerToken(string) error { return nil }
// The run page opens the pushed stream only while the run is actually
// live — a finished run has nothing left to wait on, and a page with no
// live region must not carry a hook a script would act on.
func TestRunPageOpensSSEOnlyWhileRunning(t *testing.T) {
running := fakeEngine{run: ci.Run{RunID: "run-live", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "running"}}}}
srv := runPageServer(t, running, []api.CIRun{{RunID: "run-live", Checkin: "c1", Workflow: "w",
Pipeline: "p", Status: "running", Started: "2026-08-29T01:00:00Z"}})
body := asDan(t, srv, "/r/alpha/ci/run/run-live")
if !strings.Contains(body, `data-init="@get('/r/alpha/ci/run.sse/run-live')"`) {
t.Fatalf("running page did not carry the SSE hook:\n%s", body)
}
done := fakeEngine{run: ci.Run{RunID: "run-done", Status: "completed",
Steps: map[string]ci.Step{"build": {Status: "completed"}}}}
srv2 := runPageServer(t, done, []api.CIRun{{RunID: "run-done", Checkin: "c1", Workflow: "w",
Pipeline: "p", Status: "completed", Started: "2026-08-29T01:00:00Z"}})
body2 := asDan(t, srv2, "/r/alpha/ci/run/run-done")
if strings.Contains(body2, "ci/run.sse/run-done") {
t.Fatalf("finished run still carries the SSE hook:\n%s", body2)
}
}
// ciSSEServer wires the run.sse route with a fake engine and an
// optional Changes plane, mirroring runPageServer's shape.
func ciSSEServer(t *testing.T, eng EngineViews, run api.CIRun, changes Changes) *httptest.Server {
t.Helper()
ix := landingIndex(t, false)
r := ops.NewRegistry()
deps := api.Deps{Version: "v",
Repos: api.NewRepos(map[string]*api.Repo{"alpha": {Index: ix}}),
CI: fakeCI{runs: []api.CIRun{run}}}
if err := api.RegisterAll(r, deps); err != nil {
t.Fatal(err)
}
opts := []Option{WithSessions(fakeSessions{user: "dan"}), WithCaps(stubCaps{letters: "i8"}), WithEngine(eng)}
if changes != nil {
opts = append(opts, WithChanges(changes))
}
srv := httptest.NewServer(HandlerWith(r, nil, nil, opts))
t.Cleanup(srv.Close)
return srv
}
// With a Changes plane wired, the stream sends the run's current state
// on connect and again only when the plane ticks — never on a timer of
// its own.
func TestRunSSEPatchesOnChangesTick(t *testing.T) {
eng := &countingEngine{
first: ci.Run{RunID: "run-9", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "running"}}},
next: ci.Run{RunID: "run-9", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "completed", Attempts: 1}}},
}
changes := &fakeChanges{ch: make(chan struct{}, 1)}
srv := ciSSEServer(t, eng, api.CIRun{RunID: "run-9", Checkin: "c1", Workflow: "w",
Pipeline: "p", Status: "running", Started: "2026-08-29T01:00:00Z"}, changes)
req, err := newSSERequest(srv.URL + "/r/alpha/ci/run.sse/run-9")
if err != nil {
t.Fatal(err)
}
resp, err := srv.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
rd := bufio.NewReader(resp.Body)
first := readFrames(t, rd, []string{"datastar-patch-elements", "running"}, 3*time.Second)
if !strings.Contains(first, "event: datastar-patch-elements") || !strings.Contains(first, `id="run-body"`) {
t.Fatalf("no patch on connect:\n%s", first)
}
if strings.Contains(first, "completed") {
t.Fatalf("the updated status appeared before any tick:\n%s", first)
}
changes.ch <- struct{}{}
got := readFrames(t, rd, []string{"completed"}, 3*time.Second)
if !strings.Contains(got, "completed") {
t.Fatalf("no patch after the plane ticked:\n%s", got)
}
}
// With no Changes plane at all, the stream still moves — the fallback
// ticker, not a page left frozen.
func TestRunSSEFallsBackToATickerWithNoChangesPlane(t *testing.T) {
eng := &countingEngine{
first: ci.Run{RunID: "run-7", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "running"}}},
next: ci.Run{RunID: "run-7", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "completed", Attempts: 1}}},
}
srv := ciSSEServer(t, eng, api.CIRun{RunID: "run-7", Checkin: "c1", Workflow: "w",
Pipeline: "p", Status: "running", Started: "2026-08-29T01:00:00Z"}, nil)
req, err := newSSERequest(srv.URL + "/r/alpha/ci/run.sse/run-7")
if err != nil {
t.Fatal(err)
}
resp, err := srv.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
rd := bufio.NewReader(resp.Body)
// The initial connect frame, then the fallback ticker (3s) must
// still bring the updated state without any Changes plane wired.
got := readFrames(t, rd, []string{"completed"}, 4*time.Second)
if !strings.Contains(got, "completed") {
t.Fatalf("no-plane ticker never delivered the moved state:\n%s", got)
}
}
// The handler must return promptly when the reader goes away — the
// same shape TestChatSSEIsGatedToReadableRepos relies on: ServeHTTP is
// called directly with a context that expires, so a handler that
// ignores cancellation would hang the test rather than fail it softly.
func TestRunSSEStopsWithTheRequestContext(t *testing.T) {
eng := &countingEngine{first: ci.Run{RunID: "run-5", Status: "running",
Steps: map[string]ci.Step{"build": {Status: "running"}}}}
ix := landingIndex(t, false)
r := ops.NewRegistry()
deps := api.Deps{Version: "v",
Repos: api.NewRepos(map[string]*api.Repo{"alpha": {Index: ix}}),
CI: fakeCI{runs: []api.CIRun{{RunID: "run-5", Checkin: "c1", Workflow: "w",
Pipeline: "p", Status: "running", Started: "2026-08-29T01:00:00Z"}}}}
if err := api.RegisterAll(r, deps); err != nil {
t.Fatal(err)
}
h := HandlerWith(r, nil, nil, []Option{
WithSessions(fakeSessions{user: "dan"}), WithCaps(stubCaps{letters: "i8"}), WithEngine(eng)})
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
defer cancel()
req := httptest.NewRequest("GET", "/r/alpha/ci/run.sse/run-5", nil).WithContext(ctx)
req.AddCookie(&http.Cookie{Name: sessionCookie, Value: "good"})
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req) // returns once ctx is done; a hang fails the test via its own timeout
if !strings.Contains(rec.Header().Get("Content-Type"), "event-stream") {
t.Fatalf("run.sse did not stream: %d %s", rec.Code, rec.Body)
}
}