1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "bufio"
7 "context"
8 "errors"
9 "net/http"
10 "net/http/httptest"
11 "strings"
12 "sync/atomic"
13 "testing"
14 "time"
15
16 "connectrpc.com/connect"
17 "github.com/stretchr/testify/assert"
18 "github.com/stretchr/testify/require"
19
20 pb "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22)
23
24func TestFormatEvent(t *testing.T) {
25 assert.Equal(t, "event: home-list\nid: b-1\ndata: <tr id=\"x\">\ndata: <td>1</td>\ndata: </tr>\n\n",
26 formatEvent("b-1", "home-list", "<tr id=\"x\">\n <td>1</td>\n</tr>", true), "one data line per line")
27 assert.Equal(t, "event: gap\ndata: gap\n\n", formatEvent("", "gap", "gap", true))
28 assert.Equal(t, "id: b-2\n\n", formatEvent("b-2", "", "", false), "an id alone")
29 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")
30 assert.Equal(t, "event: e\ndata: \n\n", formatEvent("", "e", "", true))
31}
32
33func TestParseTopic(t *testing.T) {
34 for in, want := range map[string]string{"home": "", "repo:x": "x", "repo:a.b-c": "a.b-c"} {
35 repo, ok := parseTopic(in)
36 assert.True(t, ok, in)
37 assert.Equal(t, want, repo, in)
38 }
39 for _, in := range []string{"", "repo:", "other", "home,repo:x", "Repo:x", "repo"} {
40 _, ok := parseTopic(in)
41 assert.False(t, ok, in)
42 }
43}
44
45func TestLiveURL(t *testing.T) {
46 assert.Equal(t, "/-/events?topic=home", liveURL("home", ""))
47 assert.Equal(t, "/-/events?topic=repo%3Ax&view=refs", liveURL("repo:x", "view=refs"))
48}
49
50func TestNewCommitsText(t *testing.T) {
51 assert.Equal(t, "1 new commit — reload", newCommits("1", 1))
52 assert.Equal(t, "3 new commits — reload", newCommits("3", 3))
53 assert.Equal(t, "100+ new commits — reload", newCommits("100+", 2))
54}
55
56// fakeEvents is an EventService whose behaviour a test sets.
57type fakeEvents struct {
58 ermiteconnect.UnimplementedEventServiceHandler
59 err error
60 events []*pb.Event
61 opened atomic.Int32
62 closed atomic.Int32
63 resume atomic.Value
64 release chan struct{}
65}
66
67func (f *fakeEvents) Watch(ctx context.Context, req *connect.Request[pb.WatchRequest], st *connect.ServerStream[pb.WatchResponse]) error {
68 f.opened.Add(1)
69 defer f.closed.Add(1)
70 f.resume.Store(req.Msg.GetResumeAfter())
71 if f.err != nil {
72 return f.err
73 }
74 _ = st.Send(nil)
75 for _, ev := range f.events {
76 if err := st.Send(&pb.WatchResponse{Event: ev}); err != nil {
77 return err
78 }
79 }
80 select {
81 case <-ctx.Done():
82 case <-f.release:
83 }
84 return nil
85}
86
87// repoOK answers every GetRepo with a public repository.
88type repoOK struct {
89 ermiteconnect.UnimplementedRepoServiceHandler
90}
91
92func (repoOK) GetRepo(_ context.Context, req *connect.Request[pb.GetRepoRequest]) (*connect.Response[pb.GetRepoResponse], error) {
93 return connect.NewResponse(&pb.GetRepoResponse{Repo: &pb.Repo{Name: req.Msg.GetRepo(), DefaultBranch: "main"}}), nil
94}
95
96// eventsServer serves the web on a fake API, with a write timeout shorter
97// than the heartbeat, like a server that sets one.
98func eventsServer(t *testing.T, f *fakeEvents, beat time.Duration, writeTimeout time.Duration) *httptest.Server {
99 t.Helper()
100 api := http.NewServeMux()
101 api.Handle(ermiteconnect.NewEventServiceHandler(f))
102 api.Handle(ermiteconnect.NewRepoServiceHandler(repoOK{}))
103 backend := httptest.NewServer(api)
104 t.Cleanup(backend.Close)
105
106 srv, err := New(Config{
107 Repos: ermiteconnect.NewRepoServiceClient(backend.Client(), backend.URL),
108 Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),
109 Events: ermiteconnect.NewEventServiceClient(backend.Client(), backend.URL),
110 PublicURL: "https://forge.example", SourceURL: "https://forge.example/ermite", Version: "v1",
111 Heartbeat: beat,
112 })
113 require.NoError(t, err)
114 ts := httptest.NewUnstartedServer(srv)
115 ts.Config.WriteTimeout = writeTimeout
116 ts.Start()
117 t.Cleanup(ts.Close)
118 return ts
119}
120
121func open(t *testing.T, ts *httptest.Server, query string, hdr map[string]string) (*http.Response, context.CancelFunc) {
122 t.Helper()
123 ctx, cancel := context.WithCancel(context.Background())
124 req, err := http.NewRequestWithContext(ctx, http.MethodGet, ts.URL+"/-/events?"+query, nil)
125 require.NoError(t, err)
126 for k, v := range hdr {
127 req.Header.Set(k, v)
128 }
129 resp, err := ts.Client().Do(req)
130 require.NoError(t, err)
131 t.Cleanup(func() { cancel(); _ = resp.Body.Close() })
132 return resp, cancel
133}
134
135// An idle stream gets a comment every heartbeat, and the write timeout of the
136// server does not end it.
137func TestEventsHeartbeatSurvivesWriteTimeout(t *testing.T) {
138 f := &fakeEvents{release: make(chan struct{})}
139 ts := eventsServer(t, f, 40*time.Millisecond, 100*time.Millisecond)
140 resp, cancel := open(t, ts, "topic=home", nil)
141 require.Equal(t, http.StatusOK, resp.StatusCode)
142 assert.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
143
144 sc := bufio.NewScanner(resp.Body)
145 beats := 0
146 deadline := time.AfterFunc(5*time.Second, cancel)
147 defer deadline.Stop()
148 start := time.Now()
149 for sc.Scan() && beats < 8 {
150 if strings.HasPrefix(sc.Text(), ": heartbeat") {
151 beats++
152 }
153 }
154 assert.Equal(t, 8, beats)
155 assert.Greater(t, time.Since(start), 250*time.Millisecond, "the stream outlived the 100 ms write timeout")
156}
157
158// The id of the event, its name and the lines of its data come through; the
159// Last-Event-ID header is the resume point; a Gap becomes a gap event.
160func TestEventsStreamsEventsAndResumes(t *testing.T) {
161 f := &fakeEvents{release: make(chan struct{}), events: []*pb.Event{
162 {Id: "b-7", Kind: &pb.Event_Pushed{Pushed: &pb.Pushed{Repo: "x"}}},
163 {Kind: &pb.Event_Gap{Gap: &pb.Gap{}}},
164 }}
165 ts := eventsServer(t, f, time.Hour, 0)
166 resp, _ := open(t, ts, "topic=repo:x&view=summary", map[string]string{"Last-Event-ID": "b-6"})
167 require.Equal(t, http.StatusOK, resp.StatusCode)
168 sc := bufio.NewScanner(resp.Body)
169 var lines []string
170 for sc.Scan() {
171 lines = append(lines, sc.Text())
172 if sc.Text() == "data: gap" {
173 break
174 }
175 }
176 assert.Contains(t, lines, "id: b-7", "an event with nothing to show keeps its id")
177 assert.Contains(t, lines, "event: gap")
178 assert.Equal(t, "b-6", f.resume.Load())
179}
180
181// Too many watchers is a 503 the browser can retry, not a stream.
182func TestEventsResourceExhausted(t *testing.T) {
183 f := &fakeEvents{err: connect.NewError(connect.CodeResourceExhausted, errors.New("too many watchers"))}
184 ts := eventsServer(t, f, time.Hour, 0)
185 resp, _ := open(t, ts, "topic=home", nil)
186 assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
187 assert.NotEmpty(t, resp.Header.Get("Retry-After"))
188 assert.NotContains(t, resp.Header.Get("Content-Type"), "event-stream")
189
190 f.err = connect.NewError(connect.CodeNotFound, errors.New("x"))
191 resp, _ = open(t, ts, "topic=home", nil)
192 assert.Equal(t, http.StatusNotFound, resp.StatusCode)
193 f.err = errors.New("boom")
194 resp, _ = open(t, ts, "topic=home", nil)
195 assert.Equal(t, http.StatusInternalServerError, resp.StatusCode)
196}
197
198// The handler returns, and the watch it opened ends, when the visitor goes
199// away or the API ends the stream.
200func TestEventsNoLeak(t *testing.T) {
201 f := &fakeEvents{release: make(chan struct{})}
202 ts := eventsServer(t, f, time.Hour, 0)
203
204 resp, cancel := open(t, ts, "topic=home", nil)
205 require.Equal(t, http.StatusOK, resp.StatusCode)
206 require.Eventually(t, func() bool { return f.opened.Load() == 1 }, 5*time.Second, time.Millisecond)
207 cancel()
208 require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the visitor left")
209
210 resp, _ = open(t, ts, "topic=home", nil)
211 require.Equal(t, http.StatusOK, resp.StatusCode)
212 require.Eventually(t, func() bool { return f.opened.Load() == 2 }, 5*time.Second, time.Millisecond)
213 close(f.release) // the API ends the stream (shutdown)
214 done := make(chan struct{})
215 go func() { _, _ = bufio.NewReader(resp.Body).ReadString(0); close(done) }()
216 select {
217 case <-done:
218 case <-time.After(5 * time.Second):
219 t.Fatal("the stream did not end with the API's")
220 }
221 require.Eventually(t, func() bool { return f.closed.Load() == 2 }, 5*time.Second, time.Millisecond)
222}
223
224func TestEventsOffWithoutClient(t *testing.T) {
225 srv, err := New(Config{
226 Repos: ermiteconnect.NewRepoServiceClient(http.DefaultClient, "http://x.invalid"),
227 Git: ermiteconnect.NewGitServiceClient(http.DefaultClient, "http://x.invalid"),
228 PublicURL: "https://forge.example",
229 })
230 require.NoError(t, err)
231 rec := httptest.NewRecorder()
232 srv.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/-/events?topic=home", nil))
233 assert.Equal(t, http.StatusNotFound, rec.Code)
234}
235
236// A change of the repository (its default branch, say) makes the pages that
237// show it load themselves again: the summary and the refs page, whose rows
238// and marks depend on it. The commits page shows a branch, not the default.
239func TestRepoUpdatedReloadsSummaryAndRefs(t *testing.T) {
240 api := http.NewServeMux()
241 api.Handle(ermiteconnect.NewRepoServiceHandler(repoOK{}))
242 backend := httptest.NewServer(api)
243 t.Cleanup(backend.Close)
244 srv, err := New(Config{
245 Repos: ermiteconnect.NewRepoServiceClient(backend.Client(), backend.URL),
246 Git: ermiteconnect.NewGitServiceClient(backend.Client(), backend.URL),
247 PublicURL: "https://forge.example",
248 })
249 require.NoError(t, err)
250
251 ev := &pb.Event{Kind: &pb.Event_RepoUpdated{RepoUpdated: &pb.RepoUpdated{Repo: &pb.Repo{Name: "x"}}}}
252 for view, wantGap := range map[string]bool{"summary": true, "refs": true, "commits": false} {
253 ls := &liveStream{s: srv, repo: "x", view: view}
254 got, err := ls.render(t.Context(), ev)
255 require.NoError(t, err, view)
256 if wantGap {
257 assert.Equal(t, gap, got, view)
258 } else {
259 assert.Empty(t, got, view)
260 }
261 }
262}