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