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