1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "bufio"
7 "context"
8 "errors"
9 "io"
10 "net/http"
11 "net/http/httptest"
12 "strings"
13 "sync/atomic"
14 "testing"
15 "time"
16
17 "connectrpc.com/connect"
18 "github.com/stretchr/testify/assert"
19 "github.com/stretchr/testify/require"
20
21 pb "bnema.dev/ermite-api/gen/ermite"
22 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
23)
24
25func TestFormatEvent(t *testing.T) {
26 assert.Equal(t, "event: home-list\nid: b-1\ndata: <tr id=\"x\">\ndata: <td>1</td>\ndata: </tr>\n\n",
27 formatEvent("b-1", "home-list", "<tr id=\"x\">\n <td>1</td>\n</tr>", true), "one data line per line")
28 assert.Equal(t, "event: gap\ndata: gap\n\n", formatEvent("", "gap", "gap", true))
29 assert.Equal(t, "id: b-2\n\n", formatEvent("b-2", "", "", false), "an id alone")
30 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")
31 assert.Equal(t, "event: e\ndata: \n\n", formatEvent("", "e", "", true))
32}
33
34func TestParseTopic(t *testing.T) {
35 for in, want := range map[string]string{"home": "", "repo:x": "x", "repo:a.b-c": "a.b-c"} {
36 repo, ok := parseTopic(in)
37 assert.True(t, ok, in)
38 assert.Equal(t, want, repo, in)
39 }
40 for _, in := range []string{"", "repo:", "other", "home,repo:x", "Repo:x", "repo"} {
41 _, ok := parseTopic(in)
42 assert.False(t, ok, in)
43 }
44}
45
46func TestLiveURL(t *testing.T) {
47 assert.Equal(t, "/-/events?topic=home", liveURL("home", ""))
48 assert.Equal(t, "/-/events?topic=repo%3Ax&view=refs", liveURL("repo:x", "view=refs"))
49}
50
51func TestNewCommitsText(t *testing.T) {
52 assert.Equal(t, "1 new commit — reload", newCommits("1", 1))
53 assert.Equal(t, "3 new commits — reload", newCommits("3", 3))
54 assert.Equal(t, "100+ new commits — reload", newCommits("100+", 2))
55}
56
57// fakeEvents is an EventService whose behaviour a test sets.
58type fakeEvents struct {
59 ermiteconnect.UnimplementedEventServiceHandler
60 err error
61 events []*pb.Event
62 opened atomic.Int32
63 closed atomic.Int32
64 resume atomic.Value
65 release chan struct{}
66 push chan *pb.Event // events sent on demand, after the fixed ones
67}
68
69func (f *fakeEvents) Watch(ctx context.Context, req *connect.Request[pb.WatchRequest], st *connect.ServerStream[pb.WatchResponse]) error {
70 f.opened.Add(1)
71 defer f.closed.Add(1)
72 f.resume.Store(req.Msg.GetResumeAfter())
73 if f.err != nil {
74 return f.err
75 }
76 _ = st.Send(nil)
77 for _, ev := range f.events {
78 if err := st.Send(&pb.WatchResponse{Event: ev}); err != nil {
79 return err
80 }
81 }
82 for {
83 select {
84 case <-ctx.Done():
85 return nil
86 case <-f.release:
87 return nil
88 case ev := <-f.push:
89 if err := st.Send(&pb.WatchResponse{Event: ev}); err != nil {
90 return err
91 }
92 }
93 }
94}
95
96// repoOK answers every GetRepo with a public repository.
97type repoOK struct {
98 ermiteconnect.UnimplementedRepoServiceHandler
99}
100
101func (repoOK) GetRepo(_ context.Context, req *connect.Request[pb.GetRepoRequest]) (*connect.Response[pb.GetRepoResponse], error) {
102 return connect.NewResponse(&pb.GetRepoResponse{Repo: &pb.Repo{Name: req.Msg.GetRepo(), DefaultBranch: "main"}}), nil
103}
104
105// eventsServer serves the web on a fake API, with a write timeout shorter
106// than the heartbeat, like a server that sets one.
107func eventsServer(t *testing.T, f *fakeEvents, beat time.Duration, writeTimeout time.Duration) *httptest.Server {
108 t.Helper()
109 return eventsServerWith(t, f, beat, writeTimeout, nil)
110}
111
112// eventsServerWith is eventsServer with sessions, when sess is not nil.
113func eventsServerWith(t *testing.T, f *fakeEvents, beat time.Duration, writeTimeout time.Duration, sess SessionAuth) *httptest.Server {
114 t.Helper()
115 api := http.NewServeMux()
116 api.Handle(ermiteconnect.NewEventServiceHandler(f))
117 api.Handle(ermiteconnect.NewRepoServiceHandler(repoOK{}))
118 backend := httptest.NewServer(api)
119 t.Cleanup(backend.Close)
120
121 srv, err := New(Config{
122 Repos: ermiteconnect.NewRepoServiceClient(backend.Client(), backend.URL),
123 Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),
124 Events: ermiteconnect.NewEventServiceClient(backend.Client(), backend.URL),
125 PublicURL: "https://forge.example", SourceURL: "https://forge.example/ermite", Version: "v1",
126 Heartbeat: beat, Sessions: sess, Limiter: limiterFor(sess),
127 })
128 require.NoError(t, err)
129 ts := httptest.NewUnstartedServer(srv)
130 ts.Config.WriteTimeout = writeTimeout
131 ts.Start()
132 t.Cleanup(ts.Close)
133 return ts
134}
135
136func open(t *testing.T, ts *httptest.Server, query string, hdr map[string]string) (*http.Response, context.CancelFunc) {
137 t.Helper()
138 ctx, cancel := context.WithCancel(context.Background())
139 req, err := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL+"/-/events?"+query, nil)
140 require.NoError(t, err)
141 for k, v := range hdr {
142 req.Header.Set(k, v)
143 }
144 resp, err := ts.Client().Do(req)
145 require.NoError(t, err)
146 t.Cleanup(func() { cancel(); _ = resp.Body.Close() })
147 return resp, cancel
148}
149
150// An idle stream gets a comment every heartbeat, and the write timeout of the
151// server does not end it.
152func TestEventsHeartbeatSurvivesWriteTimeout(t *testing.T) {
153 f := &fakeEvents{release: make(chan struct{})}
154 ts := eventsServer(t, f, 40*time.Millisecond, 100*time.Millisecond)
155 resp, cancel := open(t, ts, "topic=home", nil)
156 require.Equal(t, http.StatusOK, resp.StatusCode)
157 assert.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
158
159 sc := bufio.NewScanner(resp.Body)
160 beats := 0
161 deadline := time.AfterFunc(5*time.Second, cancel)
162 defer deadline.Stop()
163 start := time.Now()
164 for sc.Scan() && beats < 8 {
165 if strings.HasPrefix(sc.Text(), ": heartbeat") {
166 beats++
167 }
168 }
169 assert.Equal(t, 8, beats)
170 assert.Greater(t, time.Since(start), 250*time.Millisecond, "the stream outlived the 100 ms write timeout")
171}
172
173// The id of the event, its name and the lines of its data come through; the
174// Last-Event-ID header is the resume point; a Gap becomes a gap event.
175func TestEventsStreamsEventsAndResumes(t *testing.T) {
176 f := &fakeEvents{release: make(chan struct{}), events: []*pb.Event{
177 {Id: "b-7", Kind: &pb.Event_Pushed{Pushed: &pb.Pushed{Repo: "x"}}},
178 {Kind: &pb.Event_Gap{Gap: &pb.Gap{}}},
179 }}
180 ts := eventsServer(t, f, time.Hour, 0)
181 resp, _ := open(t, ts, "topic=repo:x&view=summary", map[string]string{"Last-Event-ID": "b-6"})
182 require.Equal(t, http.StatusOK, resp.StatusCode)
183 sc := bufio.NewScanner(resp.Body)
184 var lines []string
185 for sc.Scan() {
186 lines = append(lines, sc.Text())
187 if sc.Text() == "data: gap" {
188 break
189 }
190 }
191 assert.Contains(t, lines, "id: b-7", "an event with nothing to show keeps its id")
192 assert.Contains(t, lines, "event: gap")
193 assert.Equal(t, "b-6", f.resume.Load())
194}
195
196// Too many watchers is a 503 the browser can retry, not a stream.
197func TestEventsResourceExhausted(t *testing.T) {
198 f := &fakeEvents{err: connect.NewError(connect.CodeResourceExhausted, errors.New("too many watchers"))}
199 ts := eventsServer(t, f, time.Hour, 0)
200 resp, _ := open(t, ts, "topic=home", nil)
201 assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
202 assert.NotEmpty(t, resp.Header.Get("Retry-After"))
203 assert.NotContains(t, resp.Header.Get("Content-Type"), "event-stream")
204
205 f.err = connect.NewError(connect.CodeNotFound, errors.New("x"))
206 resp, _ = open(t, ts, "topic=home", nil)
207 assert.Equal(t, http.StatusNotFound, resp.StatusCode)
208 f.err = errors.New("boom")
209 resp, _ = open(t, ts, "topic=home", nil)
210 assert.Equal(t, http.StatusInternalServerError, resp.StatusCode)
211}
212
213// The handler returns, and the watch it opened ends, when the visitor goes
214// away or the API ends the stream.
215func TestEventsNoLeak(t *testing.T) {
216 f := &fakeEvents{release: make(chan struct{})}
217 ts := eventsServer(t, f, time.Hour, 0)
218
219 resp, cancel := open(t, ts, "topic=home", nil)
220 require.Equal(t, http.StatusOK, resp.StatusCode)
221 require.Eventually(t, func() bool { return f.opened.Load() == 1 }, 5*time.Second, time.Millisecond)
222 cancel()
223 require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the visitor left")
224
225 resp, _ = open(t, ts, "topic=home", nil)
226 require.Equal(t, http.StatusOK, resp.StatusCode)
227 require.Eventually(t, func() bool { return f.opened.Load() == 2 }, 5*time.Second, time.Millisecond)
228 close(f.release) // the API ends the stream (shutdown)
229 done := make(chan struct{})
230 go func() { _, _ = bufio.NewReader(resp.Body).ReadString(0); close(done) }()
231 select {
232 case <-done:
233 case <-time.After(5 * time.Second):
234 t.Fatal("the stream did not end with the API's")
235 }
236 require.Eventually(t, func() bool { return f.closed.Load() == 2 }, 5*time.Second, time.Millisecond)
237}
238
239func TestEventsOffWithoutClient(t *testing.T) {
240 srv, err := New(Config{
241 Repos: ermiteconnect.NewRepoServiceClient(http.DefaultClient, "http://x.invalid"),
242 Git: ermiteconnect.NewGitServiceClient(http.DefaultClient, "http://x.invalid"),
243 PublicURL: "https://forge.example",
244 })
245 require.NoError(t, err)
246 rec := httptest.NewRecorder()
247 srv.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/-/events?topic=home", nil))
248 assert.Equal(t, http.StatusNotFound, rec.Code)
249}
250
251// A change of the repository (its default branch, say) makes the pages that
252// show it load themselves again: the summary and the refs page, whose rows
253// and marks depend on it. The commits page shows a branch, not the default.
254func TestRepoUpdatedReloadsSummaryAndRefs(t *testing.T) {
255 api := http.NewServeMux()
256 api.Handle(ermiteconnect.NewRepoServiceHandler(repoOK{}))
257 backend := httptest.NewServer(api)
258 t.Cleanup(backend.Close)
259 srv, err := New(Config{
260 Repos: ermiteconnect.NewRepoServiceClient(backend.Client(), backend.URL),
261 Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),
262 PublicURL: "https://forge.example",
263 })
264 require.NoError(t, err)
265
266 ev := &pb.Event{Kind: &pb.Event_RepoUpdated{RepoUpdated: &pb.RepoUpdated{Repo: &pb.Repo{Name: "x"}}}}
267 for view, wantGap := range map[string]bool{"summary": true, "refs": true, "commits": false} {
268 ls := &liveStream{s: srv, repo: "x", view: view}
269 got, err := ls.render(t.Context(), ev)
270 require.NoError(t, err, view)
271 if wantGap {
272 assert.Equal(t, gap, got, view)
273 } else {
274 assert.Empty(t, got, view)
275 }
276 }
277}
278
279func limiterFor(sess SessionAuth) FailureLimiter {
280 if sess == nil {
281 return nil
282 }
283 return &countingLimiter{}
284}
285
286// revocable is a SessionAuth whose one session ends when the test says so, like
287// SignOutEverywhere does.
288type revocable struct {
289 fakeSessions
290 ended atomic.Bool
291}
292
293func (r *revocable) Verify(ctx context.Context, tok string) (context.Context, bool, error) {
294 return ctx, tok == r.valid && !r.ended.Load(), nil
295}
296
297// GO-002: the stream of the owner is checked before each event is sent, not
298// only on the heartbeat: after the session ended, the next event is not sent
299// and the stream ends.
300func TestEventsOwnerStreamEndsBeforeNextEvent(t *testing.T) {
301 f := &fakeEvents{release: make(chan struct{}), push: make(chan *pb.Event)}
302 sess := &revocable{fakeSessions: fakeSessions{valid: "tok"}}
303 ts := eventsServerWith(t, f, time.Hour, 0, sess) // no heartbeat in the test
304 resp, cancel := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=tok"})
305 require.Equal(t, http.StatusOK, resp.StatusCode)
306 assert.Contains(t, resp.Header.Get("Cache-Control"), "private")
307 // Without the check the stream would stay open: bound the read.
308 defer time.AfterFunc(5*time.Second, cancel).Stop()
309 rd := bufio.NewReader(resp.Body)
310
311 line := func() string {
312 s, err := rd.ReadString('\n')
313 if err != nil {
314 return "EOF"
315 }
316 return strings.TrimRight(s, "\n")
317 }
318 // The stream is up; a first event goes through while the session lives.
319 require.Equal(t, ": connected", line())
320 line()
321 f.push <- &pb.Event{Id: "b-1", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}
322 var seen []string
323 for l := line(); l != "data: gap"; l = line() {
324 seen = append(seen, l)
325 require.NotEqual(t, "EOF", l)
326 }
327 assert.Contains(t, seen, "id: b-1")
328
329 // Signed out everywhere: the next event is not sent, the stream ends.
330 sess.ended.Store(true)
331 select {
332 case f.push <- &pb.Event{Id: "b-2", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}:
333 case <-time.After(5 * time.Second):
334 t.Fatal("the event was not delivered to the web")
335 }
336 rest, _ := io.ReadAll(rd)
337 assert.NotContains(t, string(rest), "b-2", "nothing is sent after the session ended")
338 require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the watch is closed")
339}
340
341// A visitor's stream does not depend on sessions: events keep flowing.
342func TestEventsVisitorStreamIgnoresSessions(t *testing.T) {
343 f := &fakeEvents{release: make(chan struct{}), push: make(chan *pb.Event)}
344 sess := &revocable{fakeSessions: fakeSessions{valid: "tok"}}
345 sess.ended.Store(true)
346 ts := eventsServerWith(t, f, time.Hour, 0, sess)
347 resp, _ := open(t, ts, "topic=home", nil)
348 require.Equal(t, http.StatusOK, resp.StatusCode)
349 f.push <- &pb.Event{Id: "b-1", Kind: &pb.Event_Gap{Gap: &pb.Gap{}}}
350 sc := bufio.NewScanner(resp.Body)
351 found := false
352 for sc.Scan() {
353 if sc.Text() == "id: b-1" {
354 found = true
355 break
356 }
357 }
358 assert.True(t, found)
359}
360
361// GO-006: a stream opened with an invalid cookie is anonymous, and its headers
362// say the answer must not be stored and varies on the cookie.
363func TestEventsInvalidCookieStream(t *testing.T) {
364 f := &fakeEvents{release: make(chan struct{})}
365 ts := eventsServerWith(t, f, time.Hour, 0, &fakeSessions{valid: "tok"})
366 resp, _ := open(t, ts, "topic=home", map[string]string{"Cookie": CookieName + "=bad"})
367 require.Equal(t, http.StatusOK, resp.StatusCode)
368 assert.Equal(t, "no-store, no-transform", resp.Header.Get("Cache-Control"))
369 assert.Contains(t, resp.Header.Values("Vary"), "Cookie")
370 assert.Contains(t, resp.Header.Get("Set-Cookie"), "Max-Age=0")
371}