detached d66b108

test(events): owner topic access matrix, token end, bearer refusal, stale index

commit
d66b108197d1e412dd67884ec50bc2df888739e3
author
brice · · 2026-10-19 06:49 UTC
parent
063865b
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +285 -1
Unified
UnifiedSplit
M internal/adapters/kvstore/export_test.go +24 -1
@@ -2,7 +2,13 @@
22
33 package kvstore
44
5-import "context"
5+import (
6+ "context"
7+
8+ kv "bnema.dev/ermite-kv"
9+
10+ "bnema.dev/ermite/internal/domain"
11+)
612
713 // SetAfterSettingsSnapshot installs the hook that runs between the snapshot of
814 // DeleteEnvironment and its batch, and returns the function that removes it.
@@ -34,3 +40,20 @@ func (s *Store) CountKeysForTest(ctx context.Context, prefix string) int {
3440 }
3541 return n
3642 }
43+
44+// DropIssueRecordForTest deletes the issue record and leaves its open index
45+// entry: the state of an index that is stale.
46+func (i *Issues) DropIssueRecordForTest(ctx context.Context, repo domain.RepoName, n int64) error {
47+ var b kv.Batch
48+ b.Delete(issueKey(repo, n))
49+ _, err := i.db.Apply(ctx, &b)
50+ return err
51+}
52+
53+// DropMRRecordForTest is DropIssueRecordForTest for a merge request.
54+func (m *MergeRequests) DropMRRecordForTest(ctx context.Context, repo domain.RepoName, n int64) error {
55+ var b kv.Batch
56+ b.Delete(mrKey(repo, n))
57+ _, err := m.db.Apply(ctx, &b)
58+ return err
59+}
@@ -2,7 +2,13 @@
2 2
3 package kvstore3 package kvstore
4 4
5-import "context"5+import (
6+ "context"
7+
8+ kv "bnema.dev/ermite-kv"
9+
10+ "bnema.dev/ermite/internal/domain"
11+)
6 12
7 // SetAfterSettingsSnapshot installs the hook that runs between the snapshot of13 // SetAfterSettingsSnapshot installs the hook that runs between the snapshot of
8 // DeleteEnvironment and its batch, and returns the function that removes it.14 // DeleteEnvironment and its batch, and returns the function that removes it.
@@ -34,3 +40,20 @@ func (s *Store) CountKeysForTest(ctx context.Context, prefix string) int {
34 }40 }
35 return n41 return n
36 }42 }
43+
44+// DropIssueRecordForTest deletes the issue record and leaves its open index
45+// entry: the state of an index that is stale.
46+func (i *Issues) DropIssueRecordForTest(ctx context.Context, repo domain.RepoName, n int64) error {
47+ var b kv.Batch
48+ b.Delete(issueKey(repo, n))
49+ _, err := i.db.Apply(ctx, &b)
50+ return err
51+}
52+
53+// DropMRRecordForTest is DropIssueRecordForTest for a merge request.
54+func (m *MergeRequests) DropMRRecordForTest(ctx context.Context, repo domain.RepoName, n int64) error {
55+ var b kv.Batch
56+ b.Delete(mrKey(repo, n))
57+ _, err := m.db.Apply(ctx, &b)
58+ return err
59+}
M internal/adapters/kvstore/issues_test.go +28 -0
@@ -425,3 +425,31 @@ func TestListOpenAllIsEmptyWithoutOpenItems(t *testing.T) {
425425 require.NoError(t, err)
426426 assert.Empty(t, open)
427427 }
428+
429+func TestListOpenAllSkipsAStaleIndexEntry(t *testing.T) {
430+ ctx := context.Background()
431+ s, mrs, n := repoWithMRs(t)
432+ is := s.Issues()
433+ keepI, err := is.Create(ctx, newIssue(t, n, "keep"))
434+ require.NoError(t, err)
435+ goneI, err := is.Create(ctx, newIssue(t, n, "gone"))
436+ require.NoError(t, err)
437+ keepM, err := mrs.Create(ctx, newMR(t, n, "keep"))
438+ require.NoError(t, err)
439+ goneM, err := mrs.Create(ctx, newMR(t, n, "gone"))
440+ require.NoError(t, err)
441+
442+ require.NoError(t, is.DropIssueRecordForTest(ctx, n, goneI.Number))
443+ require.NoError(t, mrs.DropMRRecordForTest(ctx, n, goneM.Number))
444+ require.Positive(t, countKeys(t, s, "isopen/proj/"), "the index entries are still there")
445+ require.Positive(t, countKeys(t, s, "mropen/proj/"))
446+
447+ issues, err := is.ListOpenAll(ctx)
448+ require.NoError(t, err)
449+ require.Len(t, issues, 1)
450+ assert.Equal(t, keepI.Number, issues[0].Number)
451+ open, err := mrs.ListOpenAll(ctx)
452+ require.NoError(t, err)
453+ require.Len(t, open, 1)
454+ assert.Equal(t, keepM.Number, open[0].Number)
455+}
@@ -425,3 +425,31 @@ func TestListOpenAllIsEmptyWithoutOpenItems(t *testing.T) {
425 require.NoError(t, err)425 require.NoError(t, err)
426 assert.Empty(t, open)426 assert.Empty(t, open)
427 }427 }
428+
429+func TestListOpenAllSkipsAStaleIndexEntry(t *testing.T) {
430+ ctx := context.Background()
431+ s, mrs, n := repoWithMRs(t)
432+ is := s.Issues()
433+ keepI, err := is.Create(ctx, newIssue(t, n, "keep"))
434+ require.NoError(t, err)
435+ goneI, err := is.Create(ctx, newIssue(t, n, "gone"))
436+ require.NoError(t, err)
437+ keepM, err := mrs.Create(ctx, newMR(t, n, "keep"))
438+ require.NoError(t, err)
439+ goneM, err := mrs.Create(ctx, newMR(t, n, "gone"))
440+ require.NoError(t, err)
441+
442+ require.NoError(t, is.DropIssueRecordForTest(ctx, n, goneI.Number))
443+ require.NoError(t, mrs.DropMRRecordForTest(ctx, n, goneM.Number))
444+ require.Positive(t, countKeys(t, s, "isopen/proj/"), "the index entries are still there")
445+ require.Positive(t, countKeys(t, s, "mropen/proj/"))
446+
447+ issues, err := is.ListOpenAll(ctx)
448+ require.NoError(t, err)
449+ require.Len(t, issues, 1)
450+ assert.Equal(t, keepI.Number, issues[0].Number)
451+ open, err := mrs.ListOpenAll(ctx)
452+ require.NoError(t, err)
453+ require.Len(t, open, 1)
454+ assert.Equal(t, keepM.Number, open[0].Number)
455+}
M internal/adapters/web/inbox_test.go +114 -0
@@ -3,10 +3,12 @@
33 package web
44
55 import (
6+ "bufio"
67 "context"
78 "io"
89 "net/http"
910 "net/http/httptest"
11+ "strings"
1012 "testing"
1113 "time"
1214
@@ -193,3 +195,115 @@ func TestInboxLiveFragments(t *testing.T) {
193195 func withOwner(ctx context.Context) context.Context {
194196 return domain.ContextWithPrincipal(context.WithValue(ctx, signedInKey{}, struct{}{}), domain.Owner())
195197 }
198+
199+// A bearer token is not a web session: the inbox answers it as it answers
200+// anyone else, with the unknown-page and unknown-topic responses.
201+func TestInboxRefusesABearerToken(t *testing.T) {
202+ t.Parallel()
203+ f := &watchProbe{release: make(chan struct{})}
204+ svc := mocks.NewInboxService(t) // never asked
205+ srv := newTestServer(t, Config{
206+ Repos: reposServing(t, publicRepo(t, "x")), Browse: mocks.NewBrowseService(t), Events: f.service(t), Inbox: svc,
207+ Heartbeat: time.Hour, Sessions: newSessions(t, "tok"), Limiter: &countingLimiter{},
208+ })
209+ bearer := map[string]string{"Authorization": "Bearer tok"}
210+
211+ // The page.
212+ missing := inboxGet(srv, "/-/nothing-here", false)
213+ req := httptest.NewRequest(http.MethodGet, "/-/inbox", nil)
214+ req.Header.Set("Authorization", "Bearer tok")
215+ rec := httptest.NewRecorder()
216+ srv.ServeHTTP(rec, req)
217+ assert.Equal(t, missing.Code, rec.Code)
218+ assert.Equal(t, http.StatusNotFound, rec.Code)
219+ assert.Equal(t, missing.Body.String(), rec.Body.String(), "the unknown-page response")
220+
221+ // The stream.
222+ ts := httptest.NewServer(srv)
223+ t.Cleanup(ts.Close)
224+ unknown, _ := open(t, ts, "topic=nope&view=inbox", nil)
225+ want, _ := io.ReadAll(unknown.Body)
226+ resp, _ := open(t, ts, "topic=owner&view=inbox", bearer)
227+ body, _ := io.ReadAll(resp.Body)
228+ assert.Equal(t, unknown.StatusCode, resp.StatusCode)
229+ assert.Equal(t, string(want), string(body), "the unknown-topic response")
230+ assert.Zero(t, f.opened.Load(), "no watch is opened")
231+}
232+
233+// The owner's inbox stream ends once the session is invalid: the next event is
234+// not sent.
235+func TestInboxStreamEndsWhenTheSessionEnds(t *testing.T) {
236+ t.Parallel()
237+ f := &watchProbe{release: make(chan struct{}), push: make(chan ports.WatchEvent)}
238+ sess, ended := newRevocable(t, "tok")
239+ svc := mocks.NewInboxService(t)
240+ svc.EXPECT().Overview(mock.Anything, mock.Anything, domain.RepoName{}).Return(inboxOverview(t), nil).Maybe()
241+ srv := newTestServer(t, Config{
242+ Repos: reposServing(t, publicRepo(t, "x")), Browse: mocks.NewBrowseService(t), Events: f.service(t), Inbox: svc,
243+ Heartbeat: time.Hour, Sessions: sess, Limiter: &countingLimiter{},
244+ })
245+ ts := httptest.NewServer(srv)
246+ t.Cleanup(ts.Close)
247+ resp, cancel := open(t, ts, "topic=owner&view=inbox", map[string]string{"Cookie": CookieName + "=tok"})
248+ require.Equal(t, http.StatusOK, resp.StatusCode)
249+ defer time.AfterFunc(5*time.Second, cancel).Stop()
250+ rd := bufio.NewReader(resp.Body)
251+ line := func() string {
252+ s, err := rd.ReadString('\n')
253+ if err != nil {
254+ return "EOF"
255+ }
256+ return strings.TrimRight(s, "\n")
257+ }
258+ require.Equal(t, ": connected", line())
259+ line()
260+ ev := func(id string) ports.WatchEvent {
261+ return ports.WatchEvent{ID: id, Event: domain.IssueOpened{Issue: domain.Issue{Repo: repoName(t, "beta"), Number: 7}}}
262+ }
263+ f.push <- ev("b-1")
264+ var seen []string
265+ for l := line(); !strings.HasPrefix(l, "data:"); l = line() {
266+ seen = append(seen, l)
267+ require.NotEqual(t, "EOF", l)
268+ }
269+ assert.Contains(t, seen, "id: b-1")
270+
271+ ended.Store(true)
272+ select {
273+ case f.push <- ev("b-2"):
274+ case <-time.After(5 * time.Second):
275+ t.Fatal("the event was not delivered to the web")
276+ }
277+ rest, _ := io.ReadAll(rd)
278+ assert.NotContains(t, string(rest), "b-2", "nothing is sent after the session ended")
279+ require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the watch is closed")
280+}
281+
282+// A repository event refreshes the inbox: issues turned off or a deletion
283+// remove its items from the fragments.
284+func TestInboxRefreshesOnRepoEvents(t *testing.T) {
285+ t.Parallel()
286+ for name, ev := range map[string]domain.Event{
287+ "issues turned off": domain.RepoUpdated{Repo: repoName(t, "beta")},
288+ "repo deleted": domain.RepoDeleted{Repo: repoName(t, "beta")},
289+ } {
290+ t.Run(name, func(t *testing.T) {
291+ t.Parallel()
292+ after := ports.InboxOverview{Repos: []domain.RepoName{repoName(t, "alpha")}}
293+ svc := mocks.NewInboxService(t)
294+ svc.EXPECT().Overview(mock.Anything, mock.Anything, domain.RepoName{}).Return(after, nil)
295+ srv := inboxServer(t, svc)
296+ ls := &liveStream{s: srv, view: "inbox"}
297+ frags, err := ls.render(withOwner(t.Context()), ports.WatchEvent{ID: "b-1", Event: ev})
298+ require.NoError(t, err)
299+ require.Len(t, frags, 4)
300+ all := ""
301+ for _, f := range frags {
302+ all += f.html
303+ }
304+ assert.Contains(t, all, `id="inbox-issues"`)
305+ assert.NotContains(t, all, "crash on start", "the issue is gone")
306+ assert.NotContains(t, all, "/beta", "the repository is gone from the filter")
307+ })
308+ }
309+}
@@ -3,10 +3,12 @@
3 package web3 package web
4 4
5 import (5 import (
6+ "bufio"
6 "context"7 "context"
7 "io"8 "io"
8 "net/http"9 "net/http"
9 "net/http/httptest"10 "net/http/httptest"
11+ "strings"
10 "testing"12 "testing"
11 "time"13 "time"
12 14
@@ -193,3 +195,115 @@ func TestInboxLiveFragments(t *testing.T) {
193 func withOwner(ctx context.Context) context.Context {195 func withOwner(ctx context.Context) context.Context {
194 return domain.ContextWithPrincipal(context.WithValue(ctx, signedInKey{}, struct{}{}), domain.Owner())196 return domain.ContextWithPrincipal(context.WithValue(ctx, signedInKey{}, struct{}{}), domain.Owner())
195 }197 }
198+
199+// A bearer token is not a web session: the inbox answers it as it answers
200+// anyone else, with the unknown-page and unknown-topic responses.
201+func TestInboxRefusesABearerToken(t *testing.T) {
202+ t.Parallel()
203+ f := &watchProbe{release: make(chan struct{})}
204+ svc := mocks.NewInboxService(t) // never asked
205+ srv := newTestServer(t, Config{
206+ Repos: reposServing(t, publicRepo(t, "x")), Browse: mocks.NewBrowseService(t), Events: f.service(t), Inbox: svc,
207+ Heartbeat: time.Hour, Sessions: newSessions(t, "tok"), Limiter: &countingLimiter{},
208+ })
209+ bearer := map[string]string{"Authorization": "Bearer tok"}
210+
211+ // The page.
212+ missing := inboxGet(srv, "/-/nothing-here", false)
213+ req := httptest.NewRequest(http.MethodGet, "/-/inbox", nil)
214+ req.Header.Set("Authorization", "Bearer tok")
215+ rec := httptest.NewRecorder()
216+ srv.ServeHTTP(rec, req)
217+ assert.Equal(t, missing.Code, rec.Code)
218+ assert.Equal(t, http.StatusNotFound, rec.Code)
219+ assert.Equal(t, missing.Body.String(), rec.Body.String(), "the unknown-page response")
220+
221+ // The stream.
222+ ts := httptest.NewServer(srv)
223+ t.Cleanup(ts.Close)
224+ unknown, _ := open(t, ts, "topic=nope&view=inbox", nil)
225+ want, _ := io.ReadAll(unknown.Body)
226+ resp, _ := open(t, ts, "topic=owner&view=inbox", bearer)
227+ body, _ := io.ReadAll(resp.Body)
228+ assert.Equal(t, unknown.StatusCode, resp.StatusCode)
229+ assert.Equal(t, string(want), string(body), "the unknown-topic response")
230+ assert.Zero(t, f.opened.Load(), "no watch is opened")
231+}
232+
233+// The owner's inbox stream ends once the session is invalid: the next event is
234+// not sent.
235+func TestInboxStreamEndsWhenTheSessionEnds(t *testing.T) {
236+ t.Parallel()
237+ f := &watchProbe{release: make(chan struct{}), push: make(chan ports.WatchEvent)}
238+ sess, ended := newRevocable(t, "tok")
239+ svc := mocks.NewInboxService(t)
240+ svc.EXPECT().Overview(mock.Anything, mock.Anything, domain.RepoName{}).Return(inboxOverview(t), nil).Maybe()
241+ srv := newTestServer(t, Config{
242+ Repos: reposServing(t, publicRepo(t, "x")), Browse: mocks.NewBrowseService(t), Events: f.service(t), Inbox: svc,
243+ Heartbeat: time.Hour, Sessions: sess, Limiter: &countingLimiter{},
244+ })
245+ ts := httptest.NewServer(srv)
246+ t.Cleanup(ts.Close)
247+ resp, cancel := open(t, ts, "topic=owner&view=inbox", map[string]string{"Cookie": CookieName + "=tok"})
248+ require.Equal(t, http.StatusOK, resp.StatusCode)
249+ defer time.AfterFunc(5*time.Second, cancel).Stop()
250+ rd := bufio.NewReader(resp.Body)
251+ line := func() string {
252+ s, err := rd.ReadString('\n')
253+ if err != nil {
254+ return "EOF"
255+ }
256+ return strings.TrimRight(s, "\n")
257+ }
258+ require.Equal(t, ": connected", line())
259+ line()
260+ ev := func(id string) ports.WatchEvent {
261+ return ports.WatchEvent{ID: id, Event: domain.IssueOpened{Issue: domain.Issue{Repo: repoName(t, "beta"), Number: 7}}}
262+ }
263+ f.push <- ev("b-1")
264+ var seen []string
265+ for l := line(); !strings.HasPrefix(l, "data:"); l = line() {
266+ seen = append(seen, l)
267+ require.NotEqual(t, "EOF", l)
268+ }
269+ assert.Contains(t, seen, "id: b-1")
270+
271+ ended.Store(true)
272+ select {
273+ case f.push <- ev("b-2"):
274+ case <-time.After(5 * time.Second):
275+ t.Fatal("the event was not delivered to the web")
276+ }
277+ rest, _ := io.ReadAll(rd)
278+ assert.NotContains(t, string(rest), "b-2", "nothing is sent after the session ended")
279+ require.Eventually(t, func() bool { return f.closed.Load() == 1 }, 5*time.Second, time.Millisecond, "the watch is closed")
280+}
281+
282+// A repository event refreshes the inbox: issues turned off or a deletion
283+// remove its items from the fragments.
284+func TestInboxRefreshesOnRepoEvents(t *testing.T) {
285+ t.Parallel()
286+ for name, ev := range map[string]domain.Event{
287+ "issues turned off": domain.RepoUpdated{Repo: repoName(t, "beta")},
288+ "repo deleted": domain.RepoDeleted{Repo: repoName(t, "beta")},
289+ } {
290+ t.Run(name, func(t *testing.T) {
291+ t.Parallel()
292+ after := ports.InboxOverview{Repos: []domain.RepoName{repoName(t, "alpha")}}
293+ svc := mocks.NewInboxService(t)
294+ svc.EXPECT().Overview(mock.Anything, mock.Anything, domain.RepoName{}).Return(after, nil)
295+ srv := inboxServer(t, svc)
296+ ls := &liveStream{s: srv, view: "inbox"}
297+ frags, err := ls.render(withOwner(t.Context()), ports.WatchEvent{ID: "b-1", Event: ev})
298+ require.NoError(t, err)
299+ require.Len(t, frags, 4)
300+ all := ""
301+ for _, f := range frags {
302+ all += f.html
303+ }
304+ assert.Contains(t, all, `id="inbox-issues"`)
305+ assert.NotContains(t, all, "crash on start", "the issue is gone")
306+ assert.NotContains(t, all, "/beta", "the repository is gone from the filter")
307+ })
308+ }
309+}
M internal/usecase/events_owner_test.go +119 -0
@@ -15,6 +15,7 @@ import (
1515
1616 "bnema.dev/ermite/internal/domain"
1717 "bnema.dev/ermite/internal/ports"
18+ "bnema.dev/ermite/internal/usecase"
1819 )
1920
2021 func TestWatchOwnerTopicIsUnknownToEveryoneElse(t *testing.T) {
@@ -79,3 +80,121 @@ func withIssues(m domain.IssuesMode) domain.RepoSettings {
7980 st.Issues = m
8081 return st
8182 }
83+
84+func allScopesToken(mutate func(*domain.Token)) domain.Principal {
85+ tok := domain.Token{ID: "tok", Scopes: domain.AllScopes()}
86+ if mutate != nil {
87+ mutate(&tok)
88+ }
89+ return domain.TokenPrincipal(tok)
90+}
91+
92+func TestWatchOwnerTopicAccessMatrix(t *testing.T) {
93+ t.Parallel()
94+ missingOne := domain.AllScopes()[1:]
95+ cases := []struct {
96+ name string
97+ p domain.Principal
98+ allowed bool
99+ }{
100+ {"root", domain.Root(), true},
101+ {"owner session", domain.Owner(), true},
102+ {"unrestricted all-scope token", allScopesToken(nil), true},
103+ {"token missing one scope", allScopesToken(func(tk *domain.Token) { tk.Scopes = missingOne }), false},
104+ {"repo-limited all-scope token", allScopesToken(func(tk *domain.Token) { tk.Repos = []domain.RepoName{name(t, "aaa")} }), false},
105+ {"branch-limited all-scope token", allScopesToken(func(tk *domain.Token) { tk.BranchPrefixes = []string{"agent/"} }), false},
106+ {"anonymous", domain.Anonymous(), false},
107+ }
108+ for _, tc := range cases {
109+ t.Run(tc.name, func(t *testing.T) {
110+ t.Parallel()
111+ synctest.Test(t, func(t *testing.T) {
112+ r := newEventRig(t, 16)
113+ a, b := name(t, "aaa"), name(t, "bbb")
114+ for _, n := range []domain.RepoName{a, b} {
115+ r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(domain.IssuesOwner)}, nil).Maybe()
116+ }
117+ ctx, cancel := context.WithCancel(context.Background())
118+ defer cancel()
119+ noop := func(ports.WatchEvent) error { return nil }
120+ if !tc.allowed {
121+ err := r.events.Watch(ctx, tc.p, ports.WatchParams{Topics: []string{"owner"}}, noop)
122+ require.ErrorIs(t, err, domain.ErrInvalid)
123+ assert.EqualError(t, err, domain.ErrInvalid.Error()+`: unknown topic "owner" (want "home" or "repo:<name>")`)
124+ return
125+ }
126+ w := r.watch(ctx, tc.p, ports.WatchParams{Topics: []string{"owner"}})
127+ synctest.Wait()
128+ require.Equal(t, 1, r.events.Watchers())
129+ r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
130+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
131+ assert.Equal(t, a, w.next(t).Event.(domain.IssueOpened).Issue.Repo)
132+ assert.Equal(t, b, w.next(t).Event.(domain.MergeRequestOpened).MergeRequest.Repo)
133+ })
134+ })
135+ }
136+}
137+
138+func TestWatchOwnerTopicFollowsRepoUpdatedAndDeleted(t *testing.T) {
139+ t.Parallel()
140+ synctest.Test(t, func(t *testing.T) {
141+ r := newEventRig(t, 16)
142+ a := r.repo("aaa", domain.VisibilityPrivate)
143+ ctx, cancel := context.WithCancel(context.Background())
144+ defer cancel()
145+ owner := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}})
146+ synctest.Wait()
147+ r.feed <- domain.RepoUpdated{Repo: a, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
148+ r.feed <- domain.RepoDeleted{Repo: a, Visibility: domain.VisibilityPrivate, At: now}
149+ _, ok := owner.next(t).Event.(domain.RepoUpdated)
150+ require.True(t, ok)
151+ _, ok = owner.next(t).Event.(domain.RepoDeleted)
152+ require.True(t, ok)
153+ })
154+}
155+
156+func TestWatchOwnerTopicEndsWhenTheTokenIsRevoked(t *testing.T) {
157+ t.Parallel()
158+ synctest.Test(t, func(t *testing.T) {
159+ r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: -1})
160+ priv := r.repo("priv", domain.VisibilityPrivate)
161+ ctx, cancel := context.WithCancel(context.Background())
162+ defer cancel()
163+ w := r.watch(ctx, allScopesToken(nil), ports.WatchParams{Topics: []string{"owner"}})
164+ synctest.Wait()
165+ require.Equal(t, 1, r.events.Watchers())
166+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: priv, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
167+ w.next(t)
168+
169+ r.revoked.Store("tok", true)
170+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: priv, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
171+ synctest.Wait()
172+ select {
173+ case err := <-w.done:
174+ assert.ErrorIs(t, err, domain.ErrUnauthenticated)
175+ default:
176+ t.Fatal("the owner watch of a revoked token did not end")
177+ }
178+ assert.Empty(t, w.got, "nothing was sent after the revocation")
179+ })
180+}
181+
182+func TestWatchOwnerTopicEndsWhenTheTokenExpires(t *testing.T) {
183+ t.Parallel()
184+ synctest.Test(t, func(t *testing.T) {
185+ r := newEventRig(t, 0)
186+ priv := r.repo("priv", domain.VisibilityPrivate)
187+ // The rig's clock stands at now: this token expired at now.
188+ p := allScopesToken(func(tk *domain.Token) { tk.ExpiresAt = now })
189+ w := r.watch(context.Background(), p, ports.WatchParams{Topics: []string{"owner"}})
190+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: priv, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
191+ synctest.Wait()
192+ select {
193+ case err := <-w.done:
194+ assert.ErrorIs(t, err, domain.ErrUnauthenticated)
195+ default:
196+ t.Fatal("the owner watch of an expired token did not end")
197+ }
198+ assert.Empty(t, w.got)
199+ })
200+}
@@ -15,6 +15,7 @@ import (
15 15
16 "bnema.dev/ermite/internal/domain"16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"17 "bnema.dev/ermite/internal/ports"
18+ "bnema.dev/ermite/internal/usecase"
18 )19 )
19 20
20 func TestWatchOwnerTopicIsUnknownToEveryoneElse(t *testing.T) {21 func TestWatchOwnerTopicIsUnknownToEveryoneElse(t *testing.T) {
@@ -79,3 +80,121 @@ func withIssues(m domain.IssuesMode) domain.RepoSettings {
79 st.Issues = m80 st.Issues = m
80 return st81 return st
81 }82 }
83+
84+func allScopesToken(mutate func(*domain.Token)) domain.Principal {
85+ tok := domain.Token{ID: "tok", Scopes: domain.AllScopes()}
86+ if mutate != nil {
87+ mutate(&tok)
88+ }
89+ return domain.TokenPrincipal(tok)
90+}
91+
92+func TestWatchOwnerTopicAccessMatrix(t *testing.T) {
93+ t.Parallel()
94+ missingOne := domain.AllScopes()[1:]
95+ cases := []struct {
96+ name string
97+ p domain.Principal
98+ allowed bool
99+ }{
100+ {"root", domain.Root(), true},
101+ {"owner session", domain.Owner(), true},
102+ {"unrestricted all-scope token", allScopesToken(nil), true},
103+ {"token missing one scope", allScopesToken(func(tk *domain.Token) { tk.Scopes = missingOne }), false},
104+ {"repo-limited all-scope token", allScopesToken(func(tk *domain.Token) { tk.Repos = []domain.RepoName{name(t, "aaa")} }), false},
105+ {"branch-limited all-scope token", allScopesToken(func(tk *domain.Token) { tk.BranchPrefixes = []string{"agent/"} }), false},
106+ {"anonymous", domain.Anonymous(), false},
107+ }
108+ for _, tc := range cases {
109+ t.Run(tc.name, func(t *testing.T) {
110+ t.Parallel()
111+ synctest.Test(t, func(t *testing.T) {
112+ r := newEventRig(t, 16)
113+ a, b := name(t, "aaa"), name(t, "bbb")
114+ for _, n := range []domain.RepoName{a, b} {
115+ r.meta.EXPECT().Get(mock.Anything, n).Return(domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(domain.IssuesOwner)}, nil).Maybe()
116+ }
117+ ctx, cancel := context.WithCancel(context.Background())
118+ defer cancel()
119+ noop := func(ports.WatchEvent) error { return nil }
120+ if !tc.allowed {
121+ err := r.events.Watch(ctx, tc.p, ports.WatchParams{Topics: []string{"owner"}}, noop)
122+ require.ErrorIs(t, err, domain.ErrInvalid)
123+ assert.EqualError(t, err, domain.ErrInvalid.Error()+`: unknown topic "owner" (want "home" or "repo:<name>")`)
124+ return
125+ }
126+ w := r.watch(ctx, tc.p, ports.WatchParams{Topics: []string{"owner"}})
127+ synctest.Wait()
128+ require.Equal(t, 1, r.events.Watchers())
129+ r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
130+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
131+ assert.Equal(t, a, w.next(t).Event.(domain.IssueOpened).Issue.Repo)
132+ assert.Equal(t, b, w.next(t).Event.(domain.MergeRequestOpened).MergeRequest.Repo)
133+ })
134+ })
135+ }
136+}
137+
138+func TestWatchOwnerTopicFollowsRepoUpdatedAndDeleted(t *testing.T) {
139+ t.Parallel()
140+ synctest.Test(t, func(t *testing.T) {
141+ r := newEventRig(t, 16)
142+ a := r.repo("aaa", domain.VisibilityPrivate)
143+ ctx, cancel := context.WithCancel(context.Background())
144+ defer cancel()
145+ owner := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}})
146+ synctest.Wait()
147+ r.feed <- domain.RepoUpdated{Repo: a, At: now, Visibility: domain.VisibilityPrivate, Previous: domain.VisibilityPrivate, CreatedAt: now}
148+ r.feed <- domain.RepoDeleted{Repo: a, Visibility: domain.VisibilityPrivate, At: now}
149+ _, ok := owner.next(t).Event.(domain.RepoUpdated)
150+ require.True(t, ok)
151+ _, ok = owner.next(t).Event.(domain.RepoDeleted)
152+ require.True(t, ok)
153+ })
154+}
155+
156+func TestWatchOwnerTopicEndsWhenTheTokenIsRevoked(t *testing.T) {
157+ t.Parallel()
158+ synctest.Test(t, func(t *testing.T) {
159+ r := newEventRigConfig(t, usecase.EventsConfig{RevalidateEvery: -1})
160+ priv := r.repo("priv", domain.VisibilityPrivate)
161+ ctx, cancel := context.WithCancel(context.Background())
162+ defer cancel()
163+ w := r.watch(ctx, allScopesToken(nil), ports.WatchParams{Topics: []string{"owner"}})
164+ synctest.Wait()
165+ require.Equal(t, 1, r.events.Watchers())
166+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: priv, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
167+ w.next(t)
168+
169+ r.revoked.Store("tok", true)
170+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: priv, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
171+ synctest.Wait()
172+ select {
173+ case err := <-w.done:
174+ assert.ErrorIs(t, err, domain.ErrUnauthenticated)
175+ default:
176+ t.Fatal("the owner watch of a revoked token did not end")
177+ }
178+ assert.Empty(t, w.got, "nothing was sent after the revocation")
179+ })
180+}
181+
182+func TestWatchOwnerTopicEndsWhenTheTokenExpires(t *testing.T) {
183+ t.Parallel()
184+ synctest.Test(t, func(t *testing.T) {
185+ r := newEventRig(t, 0)
186+ priv := r.repo("priv", domain.VisibilityPrivate)
187+ // The rig's clock stands at now: this token expired at now.
188+ p := allScopesToken(func(tk *domain.Token) { tk.ExpiresAt = now })
189+ w := r.watch(context.Background(), p, ports.WatchParams{Topics: []string{"owner"}})
190+ r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: priv, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
191+ synctest.Wait()
192+ select {
193+ case err := <-w.done:
194+ assert.ErrorIs(t, err, domain.ErrUnauthenticated)
195+ default:
196+ t.Fatal("the owner watch of an expired token did not end")
197+ }
198+ assert.Empty(t, w.got)
199+ })
200+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 41

↑↓ moveenter openesc close