1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "bufio"
7 "context"
8 "errors"
9 "fmt"
10 "io"
11 "net/http"
12 "net/http/httptest"
13 "strings"
14 "sync/atomic"
15 "testing"
16 "time"
17
18 "github.com/stretchr/testify/assert"
19 "github.com/stretchr/testify/mock"
20 "github.com/stretchr/testify/require"
21
22 "bnema.dev/ermite/internal/domain"
23 "bnema.dev/ermite/internal/ports"
24 "bnema.dev/ermite/internal/ports/mocks"
25)
26
27func TestFormatEvent(t *testing.T) {
28 t.Parallel()
29 assert.Equal(t, "event: home-list\nid: b-1\ndata: <tr id=\"x\">\ndata: <td>1</td>\ndata: </tr>\n\n",
30 formatEvent("b-1", "home-list", "<tr id=\"x\">\n <td>1</td>\n</tr>", true), "one data line per line")
31 assert.Equal(t, "event: gap\ndata: gap\n\n", formatEvent("", "gap", "gap", true))
32 assert.Equal(t, "id: b-2\n\n", formatEvent("b-2", "", "", false), "an id alone")
33 assert.Equal(t, "event: e\ndata: a\ndata: b\ndata: c\n\n", formatEvent("", "e", "a\r\nb\rc", true), "no line break escapes the field")
34 assert.Equal(t, "event: e\ndata: \n\n", formatEvent("", "e", "", true))
35}
36
37func TestParseTopic(t *testing.T) {
38 t.Parallel()
39 for in, want := range map[string]string{"home": "", "owner": "", "repo:x": "x", "repo:a.b-c": "a.b-c"} {
40 repo, ok := parseTopic(in)
41 assert.True(t, ok, in)
42 assert.Equal(t, want, repo, in)
43 }
44 for _, in := range []string{"", "repo:", "other", "home,repo:x", "Repo:x", "repo"} {
45 _, ok := parseTopic(in)
46 assert.False(t, ok, in)
47 }
48}
49
50func TestLiveURL(t *testing.T) {
51 t.Parallel()
52 assert.Equal(t, "/-/events?topic=home", liveURL("home", ""))
53 assert.Equal(t, "/-/events?topic=repo%3Ax&view=refs", liveURL("repo:x", "view=refs"))
54}
55
56func TestNewCommitsText(t *testing.T) {
57 t.Parallel()
58 assert.Equal(t, "1 new commit — reload", newCommits("1", 1))
59 assert.Equal(t, "3 new commits — reload", newCommits("3", 3))
60 assert.Equal(t, "100+ new commits — reload", newCommits("100+", 2))
61}
62
63// watchProbe drives the generated EventService mock: what the watch does, and
64// a record of the principal and the parameters of the watches it was asked for.
65type watchProbe struct {
66 err error
67 events []ports.WatchEvent
68 opened atomic.Int32
69 closed atomic.Int32
70 resume atomic.Value // string: ResumeAfter of the last watch
71 topics atomic.Value // []string: Topics of the last watch
72 client atomic.Value // string: Client of the last watch
73 who atomic.Value // domain.Principal: caller of the last watch
74 release chan struct{}
75 push chan ports.WatchEvent // events sent on demand, after the fixed ones
76}
77
78// service is the EventService the web watches: the mock, answering with the
79// probe.
80func (f *watchProbe) service(t testing.TB) *mocks.EventService {
81 t.Helper()
82 m := mocks.NewEventService(t)
83 m.EXPECT().Watch(mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(f.run).Maybe()
84 return m
85}
86
87func (f *watchProbe) run(ctx context.Context, p domain.Principal, params ports.WatchParams, send func(ports.WatchEvent) error) error {
88 f.opened.Add(1)
89 defer f.closed.Add(1)
90 f.resume.Store(params.ResumeAfter)
91 f.topics.Store(params.Topics)
92 f.client.Store(params.Client)
93 f.who.Store(p)
94 if f.err != nil {
95 return f.err
96 }
97 if params.Ready != nil {
98 params.Ready()
99 }
100 for _, ev := range f.events {
101 if err := send(ev); err != nil {
102 return err
103 }
104 }
105 for {
106 select {
107 case <-ctx.Done():
108 return nil
109 case <-f.release:
110 return nil
111 case ev := <-f.push:
112 if err := send(ev); err != nil {
113 return err
114 }
115 }
116 }
117}
118
119// eventsServer serves the web on a fake EventService, with a write timeout
120// shorter than the heartbeat, like a server that sets one.
121func eventsServer(t *testing.T, f *watchProbe, beat time.Duration, writeTimeout time.Duration) *httptest.Server {
122 t.Helper()
123 return eventsServerWith(t, f, beat, writeTimeout, nil)
124}
125
126// eventsServerWith is eventsServer with sessions, when sess is not nil.
127func eventsServerWith(t *testing.T, f *watchProbe, beat time.Duration, writeTimeout time.Duration, sess ports.WebSessionService) *httptest.Server {
128 t.Helper()
129 srv := newTestServer(t, Config{
130 Repos: reposServing(t, publicRepo(t, "x")),
131 Browse: mocks.NewBrowseService(t),
132 Events: f.service(t),
133 Heartbeat: beat, Sessions: sess, Limiter: limiterFor(sess),
134 })
135 ts := httptest.NewUnstartedServer(srv)
136 ts.Config.WriteTimeout = writeTimeout
137 ts.Start()
138 t.Cleanup(ts.Close)
139 return ts
140}
141
142func open(t *testing.T, ts *httptest.Server, query string, hdr map[string]string) (*http.Response, context.CancelFunc) {
143 t.Helper()
144 ctx, cancel := context.WithCancel(context.Background())
145 req, err := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL+"/-/events?"+query, nil)
146 require.NoError(t, err)
147 for k, v := range hdr {
148 req.Header.Set(k, v)
149 }
150 resp, err := ts.Client().Do(req)
151 require.NoError(t, err)
152 t.Cleanup(func() { cancel(); _ = resp.Body.Close() })
153 return resp, cancel
154}
155
156// An idle stream gets a comment every heartbeat, and the write timeout of the
157// server does not end it.
158func TestEventsHeartbeatSurvivesWriteTimeout(t *testing.T) {
159 t.Parallel()
160 f := &watchProbe{release: make(chan struct{})}
161 ts := eventsServer(t, f, 40*time.Millisecond, 100*time.Millisecond)
162 resp, cancel := open(t, ts, "topic=home", nil)
163 require.Equal(t, http.StatusOK, resp.StatusCode)
164 assert.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
165
166 sc := bufio.NewScanner(resp.Body)
167 beats := 0
168 deadline := time.AfterFunc(5*time.Second, cancel)
169 defer deadline.Stop()
170 start := time.Now()
171 for sc.Scan() && beats < 8 {
172 if strings.HasPrefix(sc.Text(), ": heartbeat") {
173 beats++
174 }
175 }
176 assert.Equal(t, 8, beats)
177 assert.Greater(t, time.Since(start), 250*time.Millisecond, "the stream outlived the 100 ms write timeout")
178}
179
180// The id of the event, its name and the lines of its data come through; the
181// Last-Event-ID header is the resume point; a Gap becomes a gap event.
182func TestEventsStreamsEventsAndResumes(t *testing.T) {
183 t.Parallel()
184 f := &watchProbe{release: make(chan struct{}), events: []ports.WatchEvent{
185 {ID: "b-7", Event: domain.PushEvent{Repo: repoName(t, "x")}},
186 {Gap: true},
187 }}
188 ts := eventsServer(t, f, time.Hour, 0)
189 resp, _ := open(t, ts, "topic=repo:x&view=summary", map[string]string{"Last-Event-ID": "b-6"})
190 require.Equal(t, http.StatusOK, resp.StatusCode)
191 sc := bufio.NewScanner(resp.Body)
192 var lines []string
193 for sc.Scan() {
194 lines = append(lines, sc.Text())
195 if sc.Text() == "data: gap" {
196 break
197 }
198 }
199 assert.Contains(t, lines, "id: b-7", "an event with nothing to show keeps its id")
200 assert.Contains(t, lines, "event: gap")
201 assert.Equal(t, "b-6", f.resume.Load())
202 assert.Equal(t, []string{"repo:x"}, f.topics.Load())
203 assert.Equal(t, domain.Anonymous(), f.who.Load(), "a visitor watches as anonymous")
204 assert.NotEmpty(t, f.client.Load(), "the watch is attributed to the address of the visitor")
205}
206
207// Too many watchers is a 503 the browser can retry, not a stream; what the
208// watch is refused for decides the status.
209func TestEventsRefusals(t *testing.T) {
210 t.Parallel()
211 tests := []struct {
212 name string
213 err error
214 want int
215 retryAfter bool
216 }{
217 {"resource exhausted", fmt.Errorf("%w: too many watchers", domain.ErrResourceExhausted), http.StatusServiceUnavailable, true},
218 {"not found", domain.ErrNotFound, http.StatusNotFound, false},
219 {"invalid", domain.ErrInvalid, http.StatusNotFound, false},
220 {"unauthenticated", domain.ErrUnauthenticated, http.StatusUnauthorized, false},
221 {"internal", errors.New("boom"), http.StatusInternalServerError, false},
222 }
223 for _, tt := range tests {
224 t.Run(tt.name, func(t *testing.T) {
225 t.Parallel()
226 ts := eventsServer(t, &watchProbe{err: tt.err}, time.Hour, 0)
227 resp, _ := open(t, ts, "topic=home", nil)
228 assert.Equal(t, tt.want, resp.StatusCode)
229 assert.Equal(t, tt.retryAfter, resp.Header.Get("Retry-After") != "")
230 assert.NotContains(t, resp.Header.Get("Content-Type"), "event-stream")
231 assert.Equal(t, "no-store", resp.Header.Get("Cache-Control"))
232 })
233 }
234}
235
236// A panic in the watch runs outside the handler goroutine, where net/http
237// would not recover it: it ends the stream with a 500 instead of the process.
238func TestEventsWatchPanicIsAnError(t *testing.T) {
239 t.Parallel()
240 m := mocks.NewEventService(t)
241 m.EXPECT().Watch(mock.Anything, mock.Anything, mock.Anything, mock.Anything).
242 RunAndReturn(func(context.Context, domain.Principal, ports.WatchParams, func(ports.WatchEvent) error) error {
243 panic("boom")
244 })
245 ts := httptest.NewServer(newTestServer(t, Config{
246 Repos: reposServing(t, publicRepo(t, "x")),
247 Browse: mocks.NewBrowseService(t),
248 Events: m,
249 }))
250 t.Cleanup(ts.Close)
251 resp, _ := open(t, ts, "topic=home", nil)
252 assert.Equal(t, http.StatusInternalServerError, resp.StatusCode)
253}
254
255// The topic of a repository the visitor cannot see is refused before any
256// watch is opened: a private repository is a missing one.
257func TestEventsPrivateRepositoryTopic(t *testing.T) {
258 t.Parallel()
259 f := &watchProbe{release: make(chan struct{})}
260 srv := newTestServer(t, Config{
261 Repos: reposServing(t, publicRepo(t, "pub")), Browse: mocks.NewBrowseService(t), Events: f.service(t),
262 })
263 ts := httptest.NewServer(srv)
264 t.Cleanup(ts.Close)
265 for _, query := range []string{"topic=repo:secret", "topic=repo:Bad"} {
266 resp, _ := open(t, ts, query, nil)
267 assert.Equal(t, http.StatusNotFound, resp.StatusCode, query)
268 }
269 assert.Zero(t, f.opened.Load(), "no watch was opened for them")
270}
271
272// The owner's stream is the owner's: the watch is opened as the owner.
273func TestEventsOwnerWatchesAsOwner(t *testing.T) {
274 t.Parallel()
275 f := &watchProbe{release: make(chan struct{})}
276 sess, _ := newRevocable(t, "tok")
277 ts := eventsServerWith(t, f, time.Hour, 0, sess)
278 resp, _ := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=tok"})
279 require.Equal(t, http.StatusOK, resp.StatusCode)
280 assert.Equal(t, domain.Owner(), f.who.Load())
281}
282
283// The handler returns, and the watch it opened ends, when the visitor goes
284// away or the API ends the stream.
285func TestEventsNoLeak(t *testing.T) {
286 t.Parallel()
287 f := &watchProbe{release: make(chan struct{})}
288 ts := eventsServer(t, f, time.Hour, 0)
289
290 resp, cancel := open(t, ts, "topic=home", nil)
291 require.Equal(t, http.StatusOK, resp.StatusCode)
292 require.Eventually(t, func() bool { return f.opened.Load() == 1 }, 5*time.Second, time.Millisecond)
293 cancel()
294 require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the visitor left")
295
296 resp, _ = open(t, ts, "topic=home", nil)
297 require.Equal(t, http.StatusOK, resp.StatusCode)
298 require.Eventually(t, func() bool { return f.opened.Load() == 2 }, 5*time.Second, time.Millisecond)
299 close(f.release) // the API ends the stream (shutdown)
300 done := make(chan struct{})
301 go func() { _, _ = bufio.NewReader(resp.Body).ReadString(0); close(done) }()
302 select {
303 case <-done:
304 case <-time.After(5 * time.Second):
305 t.Fatal("the stream did not end with the API's")
306 }
307 require.Eventually(t, func() bool { return f.closed.Load() == 2 }, 5*time.Second, time.Millisecond)
308}
309
310func TestEventsOffWithoutEventService(t *testing.T) {
311 t.Parallel()
312 srv := newTestServer(t, Config{Repos: mocks.NewRepoService(t), Browse: mocks.NewBrowseService(t)})
313 rec := httptest.NewRecorder()
314 srv.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/-/events?topic=home", nil))
315 assert.Equal(t, http.StatusNotFound, rec.Code)
316}
317
318// A change of the repository (its default branch, say) makes the pages that
319// show it load themselves again: the summary and the refs page, whose rows
320// and marks depend on it. The commits page shows a branch, not the default.
321func TestRepoUpdatedReloadsSummaryAndRefs(t *testing.T) {
322 t.Parallel()
323 srv := newTestServer(t, Config{Repos: reposServing(t, publicRepo(t, "x")), Browse: mocks.NewBrowseService(t)})
324
325 ev := ports.WatchEvent{Event: domain.RepoUpdated{Repo: repoName(t, "x")}}
326 for view, wantGap := range map[string]bool{"summary": true, "refs": true, "commits": false} {
327 ls := &liveStream{s: srv, repo: "x", view: view}
328 got, err := ls.render(t.Context(), ev)
329 require.NoError(t, err, view)
330 if wantGap {
331 assert.Equal(t, gap, got, view)
332 } else {
333 assert.Empty(t, got, view)
334 }
335 }
336}
337
338func limiterFor(sess ports.WebSessionService) ports.FailureLimiter {
339 if sess == nil {
340 return nil
341 }
342 return &countingLimiter{}
343}
344
345// newRevocable is a ports.WebSessionService whose one session ends when the test sets
346// ended, like SignOutEverywhere does.
347func newRevocable(t *testing.T, valid string) (*mocks.WebSessionService, *atomic.Bool) {
348 t.Helper()
349 var ended atomic.Bool
350 return newVerifier(t, valid, func(tok string) (bool, error) { return tok == valid && !ended.Load(), nil }), &ended
351}
352
353// GO-002: the stream of the owner is checked before each event is sent, not
354// only on the heartbeat: after the session ended, the next event is not sent
355// and the stream ends.
356func TestEventsOwnerStreamEndsBeforeNextEvent(t *testing.T) {
357 t.Parallel()
358 f := &watchProbe{release: make(chan struct{}), push: make(chan ports.WatchEvent)}
359 sess, ended := newRevocable(t, "tok")
360 ts := eventsServerWith(t, f, time.Hour, 0, sess) // no heartbeat in the test
361 resp, cancel := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=tok"})
362 require.Equal(t, http.StatusOK, resp.StatusCode)
363 assert.Contains(t, resp.Header.Get("Cache-Control"), "private")
364 // Without the check the stream would stay open: bound the read.
365 defer time.AfterFunc(5*time.Second, cancel).Stop()
366 rd := bufio.NewReader(resp.Body)
367
368 line := func() string {
369 s, err := rd.ReadString('\n')
370 if err != nil {
371 return "EOF"
372 }
373 return strings.TrimRight(s, "\n")
374 }
375 // The stream is up; a first event goes through while the session lives.
376 require.Equal(t, ": connected", line())
377 line()
378 f.push <- ports.WatchEvent{ID: "b-1", Gap: true}
379 var seen []string
380 for l := line(); l != "data: gap"; l = line() {
381 seen = append(seen, l)
382 require.NotEqual(t, "EOF", l)
383 }
384 assert.Contains(t, seen, "id: b-1")
385
386 // Signed out everywhere: the next event is not sent, the stream ends.
387 ended.Store(true)
388 select {
389 case f.push <- ports.WatchEvent{ID: "b-2", Gap: true}:
390 case <-time.After(5 * time.Second):
391 t.Fatal("the event was not delivered to the web")
392 }
393 rest, _ := io.ReadAll(rd)
394 assert.NotContains(t, string(rest), "b-2", "nothing is sent after the session ended")
395 require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the watch is closed")
396}
397
398// A visitor's stream does not depend on sessions: events keep flowing.
399func TestEventsVisitorStreamIgnoresSessions(t *testing.T) {
400 t.Parallel()
401 f := &watchProbe{release: make(chan struct{}), push: make(chan ports.WatchEvent)}
402 sess, ended := newRevocable(t, "tok")
403 ended.Store(true)
404 ts := eventsServerWith(t, f, time.Hour, 0, sess)
405 resp, _ := open(t, ts, "topic=home", nil)
406 require.Equal(t, http.StatusOK, resp.StatusCode)
407 f.push <- ports.WatchEvent{ID: "b-1", Gap: true}
408 sc := bufio.NewScanner(resp.Body)
409 found := false
410 for sc.Scan() {
411 if sc.Text() == "id: b-1" {
412 found = true
413 break
414 }
415 }
416 assert.True(t, found)
417}
418
419// GO-006: a stream opened with an invalid cookie is anonymous, and its headers
420// say the answer must not be stored and varies on the cookie.
421func TestEventsInvalidCookieStream(t *testing.T) {
422 t.Parallel()
423 f := &watchProbe{release: make(chan struct{})}
424 ts := eventsServerWith(t, f, time.Hour, 0, newSessions(t, "tok"))
425 resp, _ := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=bad"})
426 require.Equal(t, http.StatusOK, resp.StatusCode)
427 assert.Equal(t, "no-store, no-transform", resp.Header.Get("Cache-Control"))
428 assert.Contains(t, resp.Header.Values("Vary"), "Cookie")
429 assert.Contains(t, resp.Header.Get("Set-Cookie"), "Max-Age=0")
430}
431
432// An issues stream on a repository with issues off, unknown or private is
433// refused the same way: 404 with the same body.
434func TestEventsIssuesViewsAreDeniedLikeAMissingRepo(t *testing.T) {
435 t.Parallel()
436 priv := mocks.NewRepoService(t)
437 priv.EXPECT().Get(mock.Anything, mock.Anything, mock.Anything).Return(domain.Repo{}, domain.ErrNotFound)
438 for _, view := range []string{"issues", "issue&n=1"} {
439 answers := map[string]string{}
440 for name, repos := range map[string]ports.RepoService{
441 "off": reposServing(t, issueRepoView(t, domain.IssuesOff)),
442 "unset": reposServing(t, issueRepoView(t, "")),
443 "unknown": reposServing(t),
444 "private": priv,
445 } {
446 f := &watchProbe{release: make(chan struct{})}
447 srv := newTestServer(t, Config{Repos: repos, Browse: mocks.NewBrowseService(t), Events: f.service(t), Issues: mocks.NewIssueService(t), Heartbeat: time.Hour})
448 ts := httptest.NewServer(srv)
449 t.Cleanup(ts.Close)
450 resp, _ := open(t, ts, "topic=repo:proj&view="+view, nil)
451 body, _ := io.ReadAll(resp.Body)
452 assert.Equal(t, http.StatusNotFound, resp.StatusCode, "%s %s", view, name)
453 answers[name] = resp.Header.Get("Cache-Control") + "|" + string(body)
454 assert.Zero(t, f.opened.Load(), "%s %s: no watch is opened", view, name)
455 }
456 for name, a := range answers {
457 assert.Equal(t, answers["unknown"], a, "%s %s", view, name)
458 }
459 }
460 // With issues on, the stream opens.
461 f := &watchProbe{release: make(chan struct{})}
462 srv := newTestServer(t, Config{Repos: reposServing(t, issueRepoView(t, domain.IssuesOwner)), Browse: mocks.NewBrowseService(t), Events: f.service(t), Issues: mocks.NewIssueService(t), Heartbeat: time.Hour})
463 ts := httptest.NewServer(srv)
464 t.Cleanup(ts.Close)
465 resp, _ := open(t, ts, "topic=repo:proj&view=issues", nil)
466 assert.Equal(t, http.StatusOK, resp.StatusCode)
467}
468
469// Without the merge request service the commits page still gets a banner when
470// its branch moves; it only cannot count the new commits.
471func TestCommitsBannerWithoutMergeRequests(t *testing.T) {
472 t.Parallel()
473 srv := newTestServer(t, Config{Repos: mocks.NewRepoService(t), Browse: mocks.NewBrowseService(t)})
474 old, tip := strings.Repeat("a", 40), strings.Repeat("b", 40)
475 ls := &liveStream{s: srv, repo: "x", view: "commits", ref: "main", head: old}
476 got, err := ls.banner(t.Context(), domain.PushEvent{Updates: []domain.RefUpdate{{Ref: "refs/heads/main", Old: old, New: tip}}})
477 require.NoError(t, err)
478 require.Len(t, got, 1)
479 assert.Contains(t, got[0].html, "the branch moved — reload")
480}