1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "context"
7 "strings"
8 "sync"
9 "testing"
10 "testing/synctest"
11
12 "github.com/stretchr/testify/assert"
13 "github.com/stretchr/testify/mock"
14 "github.com/stretchr/testify/require"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18 "bnema.dev/ermite/internal/usecase"
19)
20
21func TestWatchOwnerTopicIsUnknownToEveryoneElse(t *testing.T) {
22 t.Parallel()
23 r := newEventRig(t, 0)
24 noop := func(ports.WatchEvent) error { return nil }
25 unknown := r.events.Watch(context.Background(), domain.Anonymous(), ports.WatchParams{Topics: []string{"nope"}}, noop)
26 require.ErrorIs(t, unknown, domain.ErrInvalid)
27
28 partial := tokenWith("t1", now.Add(0), "pub") // limited to a repository
29 for _, p := range []domain.Principal{domain.Anonymous(), partial, issuePrincipal(domain.ScopeIssueWrite)} {
30 err := r.events.Watch(context.Background(), p, ports.WatchParams{Topics: []string{"owner"}}, noop)
31 require.ErrorIs(t, err, domain.ErrInvalid)
32 assert.Equal(t, strings.ReplaceAll(unknown.Error(), "nope", "owner"), err.Error(), "same answer as an unknown topic")
33 }
34}
35
36func TestWatchOwnerTopicSpansRepositoriesAndHonoursIssuesOff(t *testing.T) {
37 t.Parallel()
38 synctest.Test(t, func(t *testing.T) {
39 r := newEventRig(t, 16)
40 a, b := name(t, "aaa"), name(t, "bbb")
41 var mu sync.Mutex
42 bMode := domain.IssuesOwner
43 r.meta.EXPECT().Get(mock.Anything, a).Return(domain.Repo{Name: a, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(domain.IssuesOwner)}, nil)
44 r.meta.EXPECT().Get(mock.Anything, b).RunAndReturn(func(context.Context, domain.RepoName) (domain.Repo, error) {
45 mu.Lock()
46 defer mu.Unlock()
47 return domain.Repo{Name: b, Visibility: domain.VisibilityPrivate, CreatedAt: now, Settings: withIssues(bMode)}, nil
48 })
49 ctx, cancel := context.WithCancel(context.Background())
50 defer cancel()
51 w := r.watch(ctx, domain.Owner(), ports.WatchParams{Topics: []string{"owner"}})
52 synctest.Wait()
53
54 r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: a, Number: 1}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
55 _, ok := w.next(t).Event.(domain.IssueOpened)
56 require.True(t, ok)
57 r.feed <- domain.MergeRequestOpened{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
58 _, ok = w.next(t).Event.(domain.MergeRequestOpened)
59 require.True(t, ok, "a second repository")
60
61 // Pushes are not inbox news.
62 r.feed <- r.pushEvent(a, sentinelOID, "refs/heads/x")
63 r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: b, Number: 3}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
64 _, ok = w.next(t).Event.(domain.IssueOpened)
65 require.True(t, ok, "the push was skipped")
66
67 mu.Lock()
68 bMode = domain.IssuesOff
69 mu.Unlock()
70 r.feed <- domain.IssueOpened{Issue: domain.Issue{Repo: b, Number: 4}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
71 r.feed <- domain.MergeRequestClosed{MergeRequest: domain.MergeRequest{Repo: b, Number: 2}, At: now, Visibility: domain.VisibilityPrivate, CreatedAt: now}
72 e := w.next(t)
73 _, ok = e.Event.(domain.MergeRequestClosed)
74 require.True(t, ok, "the issue event of the issues-off repository is not delivered, got %T", e.Event)
75 })
76}
77
78func withIssues(m domain.IssuesMode) domain.RepoSettings {
79 st := domain.DefaultRepoSettings()
80 st.Issues = m
81 return st
82}
83
84func 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
92func 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
138func 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
156func 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
182func 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}