1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "bufio"
7 "context"
8 "html"
9 "net/http"
10 "regexp"
11 "strings"
12 "testing"
13 "time"
14
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/require"
17
18 "bnema.dev/ermite/internal/domain"
19 "bnema.dev/ermite/internal/ports"
20 "bnema.dev/ermite/internal/usecase"
21)
22
23// sseEvent is one event read from /-/events.
24type sseEvent struct {
25 id, name, data string
26 comment bool
27}
28
29// sseStream is a client of /-/events.
30type sseStream struct {
31 resp *http.Response
32 events chan sseEvent
33 cancel context.CancelFunc
34}
35
36// sse opens /-/events?<query> like the browser does and reads it in the
37// background. It returns once the server answered, which it does when the
38// watch is registered: what happens after is not missed.
39func (h *harness) sse(query, lastID string) *sseStream {
40 h.t.Helper()
41 ctx, cancel := context.WithCancel(h.ctx)
42 h.t.Cleanup(cancel)
43 req, err := http.NewRequestWithContext(ctx, http.MethodGet, h.url("/-/events?"+query), nil)
44 require.NoError(h.t, err)
45 if lastID != "" {
46 req.Header.Set("Last-Event-ID", lastID)
47 }
48 resp, err := http.DefaultClient.Do(req)
49 require.NoError(h.t, err)
50 t := h.t
51 h.t.Cleanup(func() { _ = resp.Body.Close() })
52 require.Equal(t, http.StatusOK, resp.StatusCode)
53 s := &sseStream{resp: resp, events: make(chan sseEvent, 256), cancel: cancel}
54 go func() {
55 defer close(s.events)
56 sc := bufio.NewScanner(resp.Body)
57 sc.Buffer(make([]byte, 0, 64<<10), 1<<20)
58 var cur sseEvent
59 var data []string
60 for sc.Scan() {
61 line := sc.Text()
62 switch {
63 case line == "":
64 if cur.id != "" || cur.name != "" || len(data) > 0 {
65 cur.data = strings.Join(data, "\n")
66 s.events <- cur
67 }
68 cur, data = sseEvent{}, nil
69 case strings.HasPrefix(line, ":"):
70 s.events <- sseEvent{comment: true, data: strings.TrimSpace(line[1:])}
71 default:
72 k, v, _ := strings.Cut(line, ":")
73 v = strings.TrimPrefix(v, " ")
74 switch k {
75 case "id":
76 cur.id = v
77 case "event":
78 cur.name = v
79 case "data":
80 data = append(data, v)
81 }
82 }
83 }
84 }()
85 return s
86}
87
88// next returns the next event (not a comment) or fails after d.
89func (s *sseStream) next(t *testing.T, d time.Duration) sseEvent {
90 t.Helper()
91 timeout := time.After(d)
92 for {
93 select {
94 case ev, ok := <-s.events:
95 if !ok {
96 t.Fatal("the stream ended")
97 }
98 if ev.comment {
99 continue
100 }
101 return ev
102 case <-timeout:
103 t.Fatalf("no event within %v", d)
104 }
105 }
106}
107
108// AC5: a push shows up as the row of the repository on the home stream
109// within a second; creating, hiding and deleting repositories updates or
110// removes the row.
111func TestWebEventsHome(t *testing.T) {
112 h := newHarness(t, harnessOpts{})
113 h.seed("pub", domain.VisibilityPublic)
114 s := h.sse("topic=home", "")
115 assert.Equal(t, "text/event-stream", s.resp.Header.Get("Content-Type"))
116 assert.Contains(t, s.resp.Header.Get("Cache-Control"), "no-cache")
117
118 wt := h.workTree("main", "a.txt")
119 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
120 start := time.Now()
121 ev := s.next(t, time.Second)
122 assert.Less(t, time.Since(start), 2*time.Second)
123 assert.Equal(t, "home-list", ev.name)
124 assert.NotEmpty(t, ev.id)
125 assert.Contains(t, ev.data, `<tr id="repo-pub"`)
126 assert.Contains(t, ev.data, `new push`)
127 assert.Contains(t, ev.data, `href="/pub"`)
128
129 // A new repository is a row; so is a repository created by a push.
130 _, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name("fresh"), Description: "a <b>fresh</b> one", Visibility: domain.VisibilityPublic})
131 require.NoError(t, err)
132 ev = s.next(t, watchWait)
133 assert.Equal(t, "home-list", ev.name)
134 assert.Contains(t, ev.data, `<tr id="repo-fresh"`)
135 assert.Contains(t, ev.data, "a <b>fresh</b> one", "text is escaped")
136 assert.NotContains(t, ev.data, "new push")
137
138 // A repository that turns private is removed from the anonymous list.
139 priv := domain.VisibilityPrivate
140 _, err = h.forge.Repos.Update(h.ctx, domain.Root(), h.name("fresh"), ports.UpdateRepoParams{Visibility: &priv})
141 require.NoError(t, err)
142 ev = s.next(t, watchWait)
143 assert.Equal(t, "home-list", ev.name)
144 assert.Contains(t, ev.data, `<tr id="repo-fresh" data-gone`)
145 assert.NotContains(t, ev.data, "fresh one")
146
147 // And a deleted one.
148 require.NoError(t, h.forge.Repos.Delete(h.ctx, domain.Root(), h.name("pub")))
149 ev = s.next(t, watchWait)
150 assert.Contains(t, ev.data, `<tr id="repo-pub" data-gone`)
151}
152
153// AC5: the stream of a repository carries its last push; the refs page its
154// refs; the commits page the banner with the number of new commits.
155func TestWebEventsRepoFragments(t *testing.T) {
156 h := newHarness(t, harnessOpts{})
157 h.seed("x", domain.VisibilityPublic)
158 wt := h.workTree("main", "a.txt")
159 push := func() { h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/x.git"), "main") }
160 push()
161
162 summary := h.sse("topic=repo:x&view=summary", "")
163 refs := h.sse("topic=repo:x&view=refs", "")
164
165 _, _, page := h.get("/x/commits/main")
166 require.Contains(t, page, `<div id="commits-head" class="live-slot" role="status" data-slot>`, "the slot of the banner, a status region that stays in the page")
167 m := regexp.MustCompile(`hx-sse:connect="([^"]+)"`).FindStringSubmatch(page)
168 require.Len(t, m, 2, "the commits page is live")
169 url := html.UnescapeString(m[1])
170 require.True(t, strings.HasPrefix(url, "/-/events?"), url)
171 commits := h.sse(strings.TrimPrefix(url, "/-/events?"), "")
172
173 h.writeFile(wt, "b.txt", []byte("b\n"))
174 h.writeFile(wt, "c.txt", []byte("c\n"))
175 start := time.Now()
176 push()
177
178 ev := summary.next(t, time.Second)
179 assert.Less(t, time.Since(start), 2*time.Second)
180 assert.Equal(t, "repo-last-push", ev.name)
181 assert.Contains(t, ev.data, `<dd id="last-push-x">`)
182 assert.Contains(t, ev.data, "now")
183
184 ev = refs.next(t, watchWait)
185 assert.Equal(t, "refs-list", ev.name)
186 assert.Contains(t, ev.data, `<tbody id="ref-rows" data-rows>`)
187 assert.Contains(t, ev.data, `>main</a>`)
188 ev = refs.next(t, watchWait)
189 assert.Equal(t, "refs-list", ev.name)
190 assert.Contains(t, ev.data, `<dl class="kv" id="refs-kv">`)
191
192 ev = commits.next(t, watchWait)
193 assert.Equal(t, "commits-head", ev.name)
194 assert.Contains(t, ev.data, `id="commits-head"`)
195 assert.Contains(t, ev.data, "2 new commits — reload")
196 assert.Contains(t, ev.data, `href="/x/commits/main"`)
197
198 // One more commit: the count is since the head the page lists.
199 h.writeFile(wt, "d.txt", []byte("d\n"))
200 push()
201 assert.Contains(t, commits.next(t, watchWait).data, "3 new commits — reload")
202
203 // A push to another branch is not news for this page.
204 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/x.git"), "main:refs/heads/side")
205 idOnly := commits.next(t, watchWait)
206 assert.NotEmpty(t, idOnly.id, "the event is acknowledged, so that a reconnection resumes after it")
207 assert.Empty(t, idOnly.name)
208 assert.Empty(t, idOnly.data)
209 h.writeFile(wt, "e.txt", []byte("e\n"))
210 push()
211 assert.Contains(t, commits.next(t, watchWait).data, "4 new commits — reload")
212
213 // Pages that do not list the head of a branch do not announce anything.
214 _, _, detached := h.get("/x/commits/main/a.txt")
215 assert.NotContains(t, detached, `id="commits-head"`)
216 assert.NotContains(t, detached, "hx-sse:connect")
217 _, _, old := h.get("/x/commits/main?page=nope")
218 assert.NotContains(t, old, "hx-sse:connect")
219}
220
221// AC5: Last-Event-ID resumes after that event; an id nobody issued is a gap,
222// which tells the page to load itself again.
223func TestWebEventsResumeAndGap(t *testing.T) {
224 h := newHarness(t, harnessOpts{})
225 h.seed("pub", domain.VisibilityPublic)
226 s := h.sse("topic=home", "")
227 wt := h.workTree("main", "a.txt")
228 var ids []string
229 for i := range 3 {
230 if i > 0 {
231 h.writeFile(wt, "f"+string(rune('0'+i))+".txt", []byte("x\n"))
232 }
233 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
234 ids = append(ids, s.next(t, watchWait).id)
235 }
236 s.cancel()
237
238 r := h.sse("topic=home", ids[0])
239 for _, want := range ids[1:] {
240 ev := r.next(t, watchWait)
241 assert.Equal(t, want, ev.id)
242 assert.Equal(t, "home-list", ev.name)
243 }
244 h.writeFile(wt, "live.txt", []byte("x\n"))
245 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
246 live := r.next(t, watchWait)
247 assert.NotContains(t, ids, live.id)
248
249 g := h.sse("topic=home", "not-an-id")
250 gap := g.next(t, watchWait)
251 assert.Equal(t, "gap", gap.name)
252 assert.Empty(t, gap.id)
253 assert.Equal(t, "home-list", g.next(t, watchWait).name, "the oldest buffered event follows")
254}
255
256// A small ring: the id that fell out of it is a gap.
257func TestWebEventsGapAfterOverflow(t *testing.T) {
258 h := newHarness(t, harnessOpts{eventBuffer: 3})
259 h.seed("pub", domain.VisibilityPublic)
260 s := h.sse("topic=home", "")
261 wt := h.workTree("main", "a.txt")
262 var ids []string
263 for i := range 6 {
264 if i > 0 {
265 h.writeFile(wt, "f"+string(rune('0'+i))+".txt", []byte("x\n"))
266 }
267 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
268 ids = append(ids, s.next(t, watchWait).id)
269 }
270 r := h.sse("topic=home", ids[1])
271 assert.Equal(t, "gap", r.next(t, watchWait).name)
272 assert.Equal(t, ids[3], r.next(t, watchWait).id)
273}
274
275// AC2, AC5: nothing of a private repository reaches the anonymous stream,
276// and its topic is a missing one.
277func TestWebEventsPrivate(t *testing.T) {
278 h := newHarness(t, harnessOpts{})
279 h.seed("pub", domain.VisibilityPublic)
280 h.seed("priv", domain.VisibilityPrivate)
281 s := h.sse("topic=home", "")
282
283 missing := h.status("GET", "/-/events?topic=repo:nothing", nil)
284 assert.Equal(t, http.StatusNotFound, missing)
285 assert.Equal(t, missing, h.status("GET", "/-/events?topic=repo:priv", nil), "private is indistinguishable from missing")
286
287 wt := h.workTree("main", "a.txt")
288 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
289 desc := "secret change"
290 _, err := h.forge.Repos.Update(h.ctx, domain.Root(), h.name("priv"), ports.UpdateRepoParams{Description: &desc})
291 require.NoError(t, err)
292 // Made public later, it does not reveal what happened while it was
293 // private; the update that makes it public is shown.
294 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
295 ev := s.next(t, watchWait)
296 assert.Contains(t, ev.data, `id="repo-pub"`, "the first event is the public push: %s", ev.data)
297 h.sentinelSSE(t, s, "sentinel")
298 _, _, body := h.get("/")
299 assert.Contains(t, body, "repo-pub")
300 assert.NotContains(t, body, "repo-priv")
301}
302
303// sentinelSSE proves that nothing came since the last event: it creates a
304// public repository and requires its row to be the next event.
305func (h *harness) sentinelSSE(t *testing.T, s *sseStream, name string) {
306 t.Helper()
307 _, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name(name), Visibility: domain.VisibilityPublic})
308 require.NoError(t, err)
309 ev := s.next(t, watchWait)
310 assert.Contains(t, ev.data, `id="repo-`+name+`"`, "an event came before the sentinel: %s", ev.data)
311}
312
313// A topic is home or repo:<name>; anything else is refused before a stream
314// starts.
315func TestWebEventsTopics(t *testing.T) {
316 h := newHarness(t, harnessOpts{})
317 h.seed("pub", domain.VisibilityPublic)
318 for path, want := range map[string]int{
319 "/-/events": http.StatusBadRequest,
320 "/-/events?topic=": http.StatusBadRequest,
321 "/-/events?topic=other": http.StatusBadRequest,
322 "/-/events?topic=repo:": http.StatusBadRequest,
323 "/-/events?topic=repo:Bad%20Name": http.StatusNotFound,
324 "/-/events?topic=repo:..": http.StatusNotFound,
325 "/-/events?topic=repo:nothing": http.StatusNotFound,
326 "/-/events?topic=repo:pub.git": http.StatusNotFound,
327 "/-/events?topic=home,repo:pub": http.StatusBadRequest,
328 "/-/events?topic=repo:pub&view=zzz": http.StatusOK,
329 } {
330 if want == http.StatusOK {
331 s := h.sse(strings.TrimPrefix(path, "/-/events?"), "")
332 s.cancel()
333 continue
334 }
335 assert.Equal(t, want, h.status("GET", path, nil), path)
336 }
337 // POST is not allowed: a browser POST without Origin and session is a 403.
338 assert.Equal(t, http.StatusForbidden, h.status("POST", "/-/events?topic=home", nil))
339}
340
341// The pages connect to the stream of their topic.
342func TestWebPagesAreLive(t *testing.T) {
343 h := newHarness(t, harnessOpts{})
344 h.webRepo("demo")
345 for path, want := range map[string]string{
346 "/": `hx-sse:connect="/-/events?topic=home"`,
347 "/demo": `hx-sse:connect="/-/events?topic=repo%3Ademo&view=summary"`,
348 "/demo/refs": `hx-sse:connect="/-/events?topic=repo%3Ademo&view=refs"`,
349 } {
350 _, _, body := h.get(path)
351 assert.Contains(t, body, want, path)
352 }
353 _, _, nf := h.get("/nothing")
354 assert.NotContains(t, nf, "hx-sse:connect")
355}
356
357// The stream ends cleanly when the server stops its event log (the start of
358// the shutdown): the client sees the end of the body, and the handler is gone
359// (the test server cannot close with a handler left).
360func TestWebEventsEndOnShutdown(t *testing.T) {
361 h := newHarness(t, harnessOpts{})
362 h.seed("pub", domain.VisibilityPublic)
363 home := h.sse("topic=home", "")
364 repo := h.sse("topic=repo:pub", "")
365
366 h.forge.EventLog.Close(context.WithoutCancel(h.ctx))
367 for _, s := range []*sseStream{home, repo} {
368 select {
369 case _, ok := <-s.events:
370 for ok {
371 _, ok = <-s.events
372 }
373 case <-time.After(5 * time.Second):
374 t.Fatal("the stream did not end")
375 }
376 }
377 require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 0 }, 5*time.Second, 5*time.Millisecond)
378}
379
380// The web counts each visitor against the API's limit of streams per client:
381// one address cannot take every watch slot of the forge.
382func TestWebEventsLimitPerVisitor(t *testing.T) {
383 h := newHarness(t, harnessOpts{})
384 for range usecase.MaxWatchersPerClient {
385 h.sse("topic=home", "")
386 }
387 req, err := http.NewRequestWithContext(h.ctx, http.MethodGet, h.url("/-/events?topic=home"), nil)
388 require.NoError(t, err)
389 resp, err := http.DefaultClient.Do(req)
390 require.NoError(t, err)
391 defer func() { _ = resp.Body.Close() }()
392 assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
393 assert.NotEmpty(t, resp.Header.Get("Retry-After"))
394 assert.Equal(t, usecase.MaxWatchersPerClient, h.forge.EventLog.Watchers())
395}
396
397// A visitor that goes away frees its watch.
398func TestWebEventsClientDisconnect(t *testing.T) {
399 h := newHarness(t, harnessOpts{})
400 h.seed("pub", domain.VisibilityPublic)
401 require.Equal(t, 0, h.forge.EventLog.Watchers())
402 s := h.sse("topic=home", "")
403 require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 1 }, 5*time.Second, 5*time.Millisecond)
404 s.cancel()
405 require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 0 }, 5*time.Second, 5*time.Millisecond)
406}