1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "net/http"
10 "os/exec"
11 "syscall"
12 "testing"
13 "time"
14
15 "connectrpc.com/connect"
16 "github.com/stretchr/testify/assert"
17 "github.com/stretchr/testify/require"
18 "google.golang.org/protobuf/types/known/fieldmaskpb"
19
20 pb "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite/internal/domain"
22 "bnema.dev/ermite/internal/ports"
23 "bnema.dev/ermite/internal/usecase"
24)
25
26// watchWait is how long a test waits for an event that must arrive. Only
27// TestAC5PushEventWithinASecond holds the server to a second.
28const watchWait = 10 * time.Second
29
30// AC5: a push shows up in Watch within a second, with the ref updates.
31func TestAC5PushEventWithinASecond(t *testing.T) {
32 h := newHarness(t, harnessOpts{})
33 h.seed("pub", domain.VisibilityPublic)
34 w := h.watch(h.api(""), "", "home")
35
36 wt := h.workTree("main", "a.txt")
37 start := time.Now()
38 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
39 ev := w.next(t, time.Second)
40 assert.Less(t, time.Since(start), 2*time.Second)
41
42 pushed := ev.GetPushed()
43 require.NotNil(t, pushed, "got %v", ev)
44 assert.Equal(t, "pub", pushed.GetRepo())
45 require.Len(t, pushed.GetRefUpdates(), 1)
46 up := pushed.GetRefUpdates()[0]
47 assert.Equal(t, "refs/heads/main", up.GetRef())
48 assert.Empty(t, up.GetOldId(), "a created ref has no old id")
49 assert.Equal(t, h.bareGit("pub", "rev-parse", "main"), up.GetNewId())
50 assert.NotEmpty(t, ev.GetId())
51 assert.NotNil(t, ev.GetTime())
52
53 // A second push to the same ref carries both ids; a deletion has no new id.
54 old := up.GetNewId()
55 h.writeFile(wt, "b.txt", []byte("b\n"))
56 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
57 up2 := w.next(t, watchWait).GetPushed().GetRefUpdates()[0]
58 assert.Equal(t, old, up2.GetOldId())
59 assert.NotEqual(t, old, up2.GetNewId())
60
61 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main:refs/heads/tmp")
62 w.next(t, watchWait)
63 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), ":refs/heads/tmp")
64 del := w.next(t, watchWait).GetPushed().GetRefUpdates()[0]
65 assert.Equal(t, "refs/heads/tmp", del.GetRef())
66 assert.NotEmpty(t, del.GetOldId())
67 assert.Empty(t, del.GetNewId())
68}
69
70// AC5: creating, updating and deleting repositories through the API publish
71// RepoCreated, RepoUpdated and RepoDeleted; so does a create-on-push.
72func TestAC5RepoLifecycleEvents(t *testing.T) {
73 h := newHarness(t, harnessOpts{})
74 root := h.api(adminToken)
75 w := h.watch(root, "")
76
77 _, err := root.repo.CreateRepo(h.ctx, connect.NewRequest(&pb.CreateRepoRequest{Name: "life", Description: "d", Visibility: pb.Visibility_VISIBILITY_PUBLIC}))
78 require.NoError(t, err)
79 created := w.next(t, watchWait).GetRepoCreated().GetRepo()
80 require.NotNil(t, created)
81 assert.Equal(t, "life", created.GetName())
82 assert.Equal(t, pb.Visibility_VISIBILITY_PUBLIC, created.GetVisibility())
83 assert.Equal(t, "main", created.GetDefaultBranch())
84
85 _, err = root.repo.UpdateRepo(h.ctx, connect.NewRequest(&pb.UpdateRepoRequest{
86 Repo: "life", Description: "new", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"description"}},
87 }))
88 require.NoError(t, err)
89 updated := w.next(t, watchWait).GetRepoUpdated().GetRepo()
90 require.NotNil(t, updated)
91 assert.Equal(t, "new", updated.GetDescription())
92
93 _, err = root.repo.DeleteRepo(h.ctx, connect.NewRequest(&pb.DeleteRepoRequest{Repo: "life"}))
94 require.NoError(t, err)
95 assert.Equal(t, "life", w.next(t, watchWait).GetRepoDeleted().GetRepo())
96
97 // Create on first push publishes RepoCreated once, then the push.
98 wt := h.workTree("main", "a.txt")
99 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/fresh.git"), "main")
100 assert.Equal(t, "fresh", w.next(t, watchWait).GetRepoCreated().GetRepo().GetName())
101 assert.Equal(t, "fresh", w.next(t, watchWait).GetPushed().GetRepo())
102 // Nothing else came in between: the next event is the sentinel.
103 h.sentinel(t, w, "s-life")
104}
105
106// AC5: resume from an id replays what came after, then goes live.
107func TestAC5Resume(t *testing.T) {
108 h := newHarness(t, harnessOpts{})
109 h.seed("pub", domain.VisibilityPublic)
110 api := h.api("")
111 w := h.watch(api, "")
112
113 wt := h.workTree("main", "a.txt")
114 var ids []string
115 for i := range 4 {
116 if i > 0 {
117 h.writeFile(wt, fmt.Sprintf("f%d.txt", i), []byte("x\n"))
118 }
119 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
120 ids = append(ids, w.next(t, watchWait).GetId())
121 }
122 w.cancel()
123
124 // The client lost its connection after ids[1]: it gets 2 and 3, then live.
125 r := h.watch(api, ids[1])
126 for _, want := range ids[2:] {
127 ev := r.next(t, watchWait)
128 assert.Equal(t, want, ev.GetId())
129 assert.NotNil(t, ev.GetPushed())
130 }
131 // Nothing was replayed twice: the next event is the live one, below.
132 h.writeFile(wt, "live.txt", []byte("x\n"))
133 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
134 live := r.next(t, watchWait)
135 assert.NotNil(t, live.GetPushed())
136 assert.NotContains(t, ids, live.GetId())
137
138 // An id nobody issued is a gap, followed by the oldest buffered event: the
139 // creation of the repository, which came before the first push.
140 g := h.watch(api, "not-an-id")
141 assert.NotNil(t, g.next(t, watchWait).GetGap())
142 assert.Equal(t, "pub", g.next(t, watchWait).GetRepoCreated().GetRepo().GetName())
143 assert.Equal(t, ids[0], g.next(t, watchWait).GetId())
144}
145
146// AC5: a gap when the id fell out of a small ring.
147func TestAC5GapAfterOverflow(t *testing.T) {
148 h := newHarness(t, harnessOpts{eventBuffer: 4})
149 h.seed("pub", domain.VisibilityPublic)
150 api := h.api(adminToken)
151 w := h.watch(api, "")
152 wt := h.workTree("main", "a.txt")
153
154 var ids []string
155 for i := range 8 {
156 if i > 0 {
157 h.writeFile(wt, fmt.Sprintf("f%d.txt", i), []byte("x\n"))
158 }
159 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
160 ids = append(ids, w.next(t, watchWait).GetId())
161 }
162
163 // ids[2] is out of the ring (the last four are ids[4:]).
164 r := h.watch(api, ids[2])
165 gap := r.next(t, watchWait)
166 assert.NotNil(t, gap.GetGap())
167 assert.Empty(t, gap.GetId())
168 for _, want := range ids[4:] {
169 assert.Equal(t, want, r.next(t, watchWait).GetId())
170 }
171
172 // The last buffered id still resumes without a gap.
173 ok := h.watch(api, ids[4])
174 assert.Equal(t, ids[5], ok.next(t, watchWait).GetId())
175}
176
177// AC5: private events never reach anonymous callers, nor tokens that may not
178// read the repository; repo topics follow the same rule.
179func TestAC5NoPrivateEventsToAnonymous(t *testing.T) {
180 h := newHarness(t, harnessOpts{})
181 h.seed("pub", domain.VisibilityPublic)
182 h.seed("priv", domain.VisibilityPrivate)
183 other := h.token(scopes(domain.ScopeRepoRead), "pub")
184 reader := h.token(scopes(domain.ScopeRepoRead), "priv")
185
186 anon := h.watch(h.api(""), "")
187 scoped := h.watch(h.api(other), "")
188 rd := h.watch(h.api(reader), "")
189 root := h.watch(h.api(adminToken), "")
190
191 // A private repository's topic is not found for those who cannot read it.
192 assert.Equal(t, "not_found", codeName(watchErr(h.ctx, h.api(""), "repo:priv")))
193 assert.Equal(t, "not_found", codeName(watchErr(h.ctx, h.api(other), "repo:priv")))
194 assert.Equal(t, codeName(watchErr(h.ctx, h.api(""), "repo:nothing")), codeName(watchErr(h.ctx, h.api(""), "repo:priv")))
195
196 wt := h.workTree("main", "a.txt")
197 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
198 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/pub.git"), "main")
199 // Metadata events too.
200 _, err := h.api(adminToken).repo.UpdateRepo(h.ctx, connect.NewRequest(&pb.UpdateRepoRequest{
201 Repo: "priv", Description: "secret plans", UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"description"}},
202 }))
203 require.NoError(t, err)
204
205 // The reader of priv sees its push and its update, and pub's push.
206 assert.Equal(t, "priv", rd.next(t, watchWait).GetPushed().GetRepo())
207 assert.Equal(t, "pub", rd.next(t, watchWait).GetPushed().GetRepo(), "pub is public")
208 assert.Equal(t, "secret plans", rd.next(t, watchWait).GetRepoUpdated().GetRepo().GetDescription())
209 // Root sees everything in order.
210 assert.Equal(t, "priv", root.next(t, watchWait).GetPushed().GetRepo())
211 assert.Equal(t, "pub", root.next(t, watchWait).GetPushed().GetRepo())
212 assert.Equal(t, "priv", root.next(t, watchWait).GetRepoUpdated().GetRepo().GetName())
213
214 // Deleting reaches those who could read the repository before, not others.
215 _, err = h.api(adminToken).repo.DeleteRepo(h.ctx, connect.NewRequest(&pb.DeleteRepoRequest{Repo: "priv"}))
216 require.NoError(t, err)
217 assert.Equal(t, "priv", rd.next(t, watchWait).GetRepoDeleted().GetRepo())
218 assert.Equal(t, "priv", root.next(t, watchWait).GetRepoDeleted().GetRepo())
219
220 for name, w := range map[string]*stream{"anonymous": anon, "scoped": scoped} {
221 ev := w.next(t, watchWait)
222 assert.Equal(t, "pub", ev.GetPushed().GetRepo(), name)
223 }
224 // Nothing about priv came before the sentinel, for either of them.
225 _, err = h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name("sentinel"), Visibility: domain.VisibilityPublic})
226 require.NoError(t, err)
227 for name, w := range map[string]*stream{"anonymous": anon, "scoped": scoped} {
228 assert.Equal(t, "sentinel", w.next(t, watchWait).GetRepoCreated().GetRepo().GetName(), name)
229 }
230
231 // Resuming does not replay private events to anonymous either: after a
232 // gap, the creation and the push of pub, the sentinels the watchers above
233 // saw, and nothing about priv.
234 r := h.watch(h.api(""), "bogus")
235 assert.NotNil(t, r.next(t, watchWait).GetGap())
236 assert.Equal(t, "pub", r.next(t, watchWait).GetRepoCreated().GetRepo().GetName())
237 assert.Equal(t, "pub", r.next(t, watchWait).GetPushed().GetRepo())
238 assert.Equal(t, "sentinel", r.next(t, watchWait).GetRepoCreated().GetRepo().GetName())
239 h.sentinel(t, r, "s-resumed")
240}
241
242// AC5: a repo topic only carries that repository; a revoked token cannot
243// start a new watch.
244func TestAC5RepoTopic(t *testing.T) {
245 h := newHarness(t, harnessOpts{})
246 h.seed("a", domain.VisibilityPublic)
247 h.seed("b", domain.VisibilityPublic)
248 w := h.watch(h.api(""), "", "repo:a")
249
250 wt := h.workTree("main", "x.txt")
251 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/b.git"), "main")
252 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/a.git"), "main")
253 assert.Equal(t, "a", w.next(t, watchWait).GetPushed().GetRepo())
254 // The push to b was not delivered: the next event is the next push to a.
255 h.writeFile(wt, "y.txt", []byte("y\n"))
256 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/a.git"), "main")
257 assert.Equal(t, h.bareGit("a", "rev-parse", "main"), w.next(t, watchWait).GetPushed().GetRefUpdates()[0].GetNewId())
258}
259
260// The stream ends cleanly when the client cancels, and Watch counts go back to zero.
261func TestWatchEndsOnCancel(t *testing.T) {
262 h := newHarness(t, harnessOpts{})
263 w := h.watch(h.api(""), "")
264 require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 1 }, 5*time.Second, time.Millisecond)
265 w.cancel()
266 select {
267 case <-w.done:
268 case <-time.After(5 * watchWait):
269 t.Fatal("the stream did not end")
270 }
271 require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == 0 }, 5*time.Second, time.Millisecond)
272}
273
274// Shutdown ends the streams cleanly: the client sees the stream end without
275// an error and the server exits 0 well before the shutdown timeout.
276func TestServeShutdownEndsWatchStreams(t *testing.T) {
277 requireGit(t)
278 data := t.TempDir()
279 addr := freeAddr(t)
280 cmd := exec.Command(binary(t), "serve")
281 cmd.Env = append(baseEnv(),
282 "ERMITE_ADDR="+addr, "ERMITE_DATA_DIR="+data, "ERMITE_PUBLIC_URL=http://"+addr,
283 "ERMITE_ADMIN_TOKEN="+adminToken, "ERMITE_LOG_LEVEL=debug", "ERMITE_LOG_FORMAT=json",
284 )
285 var logs syncBuffer
286 cmd.Stdout, cmd.Stderr = &logs, &logs
287 require.NoError(t, cmd.Start())
288 exited := make(chan error, 1)
289 go func() { exited <- cmd.Wait() }()
290 t.Cleanup(func() { _ = cmd.Process.Kill() })
291
292 base := "http://" + addr
293 require.Eventually(t, func() bool {
294 resp, err := http.Get(base + "/-/health")
295 if err != nil {
296 return false
297 }
298 _ = resp.Body.Close()
299 return resp.StatusCode == 200
300 }, 15*time.Second, 50*time.Millisecond, "%s", logs.String())
301
302 c := newClients(base+"/-/api", adminToken)
303 ctx, cancel := context.WithCancel(context.Background())
304 defer cancel()
305 st, err := c.event.Watch(ctx, connect.NewRequest(&pb.WatchRequest{}))
306 require.NoError(t, err)
307 ended := make(chan error, 1)
308 go func() {
309 for st.Receive() {
310 }
311 ended <- st.Err()
312 }()
313
314 // The server version comes from the real binary.
315 v, err := c.server.GetVersion(ctx, connect.NewRequest(&pb.GetVersionRequest{}))
316 require.NoError(t, err)
317 assert.NotEmpty(t, v.Msg.GetVersion())
318
319 start := time.Now()
320 require.NoError(t, cmd.Process.Signal(syscall.SIGTERM))
321 select {
322 case err := <-ended:
323 assert.NoError(t, err, "the stream ends cleanly")
324 case <-time.After(15 * watchWait):
325 t.Fatalf("the stream did not end\n%s", logs.String())
326 }
327 select {
328 case err := <-exited:
329 require.NoError(t, err, "%s", logs.String())
330 case <-time.After(15 * watchWait):
331 t.Fatalf("the server did not exit\n%s", logs.String())
332 }
333 assert.Less(t, time.Since(start), 8*time.Second, "shutdown did not wait for its timeout")
334 assert.NotContains(t, logs.String(), `"level":"error"`, "no error logged:\n%s", logs.String())
335 assert.False(t, errors.Is(err, context.Canceled))
336}
337
338// A token that is revoked while its Watch is open ends the stream with
339// unauthenticated, and the private events that follow never reach it.
340func TestWatchEndsWhenTheTokenIsRevoked(t *testing.T) {
341 h := newHarness(t, harnessOpts{revalidateEvery: 20 * time.Millisecond})
342 h.seed("priv", domain.VisibilityPrivate)
343 created, err := h.forge.Tokens.Create(h.ctx, domain.Root(), ports.CreateTokenParams{
344 Label: "watcher", Scopes: scopes(domain.ScopeRepoRead),
345 })
346 require.NoError(t, err)
347 w := h.watch(h.api(created.Secret), "")
348 wt := h.workTree("main", "a.txt")
349 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
350 assert.Equal(t, "priv", w.next(t, 5*watchWait).GetPushed().GetRepo(), "the token may read priv")
351
352 require.NoError(t, h.forge.Tokens.Revoke(h.ctx, domain.Root(), created.Token.ID))
353 h.writeFile(wt, "b.txt", []byte("b\n"))
354 h.mustGit(wt, "push", "-q", h.urlAuth(adminToken, "/priv.git"), "main")
355
356 select {
357 case <-w.done:
358 case <-time.After(5 * watchWait):
359 t.Fatal("the stream of a revoked token did not end")
360 }
361 assert.Equal(t, "unauthenticated", codeName(w.err))
362 for ev := range w.events {
363 t.Fatalf("an event reached the revoked token: %v", ev)
364 }
365}
366
367// Watch refuses too many topics, and too many streams from one client.
368func TestWatchLimits(t *testing.T) {
369 h := newHarness(t, harnessOpts{})
370 api := h.api(adminToken)
371
372 topics := make([]string, usecase.MaxWatchTopics+1)
373 for i := range topics {
374 topics[i] = "home"
375 }
376 assert.Equal(t, "invalid_argument", codeName(watchErr(h.ctx, api, topics...)))
377
378 for range usecase.MaxWatchersPerClient {
379 h.watch(api, "")
380 }
381 require.Eventually(t, func() bool { return h.forge.EventLog.Watchers() == usecase.MaxWatchersPerClient }, 5*time.Second, time.Millisecond)
382 assert.Equal(t, "resource_exhausted", codeName(watchErr(h.ctx, api)))
383 // A stream of another client address is not affected (the harness trusts
384 // the loopback proxy only when configured: ask the use case directly).
385 ctx, cancel := context.WithCancel(h.ctx)
386 defer cancel()
387 registered := false
388 err := h.forge.EventLog.Watch(ctx, domain.Root(), ports.WatchParams{
389 Client: "another-client", Ready: func() { registered = true; cancel() },
390 }, func(ports.WatchEvent) error { return nil })
391 require.NoError(t, err)
392 assert.True(t, registered, "a watch of another client address is registered")
393}
394
395// sentinel proves that nothing reached the stream since the last event it
396// returned: it creates a public repository, which every watcher of "home" may
397// see, and requires its RepoCreated to be the next event. Events arrive in the
398// order they were published.
399func (h *harness) sentinel(t *testing.T, s *stream, name string) {
400 t.Helper()
401 _, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name(name), Visibility: domain.VisibilityPublic})
402 require.NoError(t, err)
403 ev := s.next(t, watchWait)
404 assert.Equal(t, name, ev.GetRepoCreated().GetRepo().GetName(), "an event came before the sentinel: %v", ev)
405}