1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package web
4
5import (
6 "context"
7 "errors"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "testing/synctest"
12
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/mock"
15 "github.com/stretchr/testify/require"
16
17 "bnema.dev/ermite/internal/domain"
18 "bnema.dev/ermite/internal/ports"
19 "bnema.dev/ermite/internal/ports/mocks"
20)
21
22func frag(s string) []fragment { return []fragment{{event: "refs-list", html: s}} }
23
24func TestFanoutBuildsOncePerKey(t *testing.T) {
25 t.Parallel()
26 base := fanKey{repo: "x", event: "b-1"}
27 for _, tt := range []struct {
28 name string
29 keys []fanKey
30 built int32
31 }{
32 {"same key", []fanKey{base, base, base}, 1},
33 {"owner and anonymous never share", []fanKey{base, {repo: "x", event: "b-1", owner: true}}, 2},
34 {"another event", []fanKey{base, {repo: "x", event: "b-2"}}, 2},
35 {"another repository", []fanKey{base, {repo: "y", event: "b-1"}}, 2},
36 {"no event id: nothing to share by", []fanKey{{repo: "x"}, {repo: "x"}}, 2},
37 } {
38 t.Run(tt.name, func(t *testing.T) {
39 t.Parallel()
40 var f fanout
41 var built atomic.Int32
42 for _, k := range tt.keys {
43 got, err := f.do(context.Background(), k, func() ([]fragment, error) {
44 built.Add(1)
45 return frag("for-owner=" + map[bool]string{true: "yes", false: "no"}[k.owner]), nil
46 })
47 require.NoError(t, err)
48 assert.Equal(t, "for-owner="+map[bool]string{true: "yes", false: "no"}[k.owner], got[0].html,
49 "a watcher never gets the output built for the other class")
50 }
51 assert.Equal(t, tt.built, built.Load())
52 })
53 }
54}
55
56// While the first watcher builds, every other one waits for it: in a synctest
57// bubble Wait returns once all goroutines are blocked, so the count of the
58// watchers that entered and none that returned proves the others wait on the
59// build instead of making their own.
60func TestFanoutConcurrentWatchersShareOneBuild(t *testing.T) {
61 t.Parallel()
62 const watchers = 50
63 synctest.Test(t, func(t *testing.T) {
64 var f fanout
65 var built, entered, returned atomic.Int32
66 release := make(chan struct{})
67 key := fanKey{repo: "x", event: "b-1"}
68 var wg sync.WaitGroup
69 results := make([][]fragment, watchers)
70 for i := range results {
71 wg.Add(1)
72 go func() {
73 defer wg.Done()
74 entered.Add(1)
75 results[i], _ = f.do(context.Background(), key, func() ([]fragment, error) {
76 built.Add(1)
77 <-release
78 return frag("once"), nil
79 })
80 returned.Add(1)
81 }()
82 }
83 synctest.Wait() // the builder is on release, the others on its entry
84 require.Equal(t, int32(watchers), entered.Load())
85 require.Equal(t, int32(1), built.Load(), "one builder")
86 require.Zero(t, returned.Load(), "the other %d wait for it", watchers-1)
87
88 close(release)
89 wg.Wait()
90 assert.Equal(t, int32(1), built.Load())
91 for _, r := range results {
92 assert.Equal(t, frag("once"), r)
93 }
94 })
95}
96
97func TestFanoutFailureIsNotKept(t *testing.T) {
98 t.Parallel()
99 var f fanout
100 key := fanKey{repo: "x", event: "b-1"}
101 _, err := f.do(context.Background(), key, func() ([]fragment, error) { return nil, errors.New("boom") })
102 require.Error(t, err)
103 got, err := f.do(context.Background(), key, func() ([]fragment, error) { return frag("ok"), nil })
104 require.NoError(t, err)
105 assert.Equal(t, frag("ok"), got, "the next watcher builds again")
106}
107
108func TestFanoutWaiterStopsWithItsContext(t *testing.T) {
109 t.Parallel()
110 var f fanout
111 key := fanKey{repo: "x", event: "b-1"}
112 started, release := make(chan struct{}), make(chan struct{})
113 go func() {
114 _, _ = f.do(context.Background(), key, func() ([]fragment, error) {
115 close(started)
116 <-release
117 return frag("late"), nil
118 })
119 }()
120 <-started
121 ctx, cancel := context.WithCancel(context.Background())
122 cancel()
123 _, err := f.do(ctx, key, func() ([]fragment, error) { return frag("never"), nil })
124 assert.ErrorIs(t, err, context.Canceled)
125 close(release)
126}
127
128func TestFanoutIsBounded(t *testing.T) {
129 t.Parallel()
130 var f fanout
131 for i := range fanMax + 10 {
132 key := fanKey{repo: "x", event: "b-" + string(rune('A'+i%26)) + string(rune('a'+i/26))}
133 _, err := f.do(context.Background(), key, func() ([]fragment, error) { return frag("x"), nil })
134 require.NoError(t, err)
135 }
136 assert.LessOrEqual(t, len(f.m), fanMax)
137}
138
139// refsMocks is a web server over mocks that expect the refs fragment to load
140// the repository and its refs once per build of each class of watcher
141// (anonymous, owner), asking the use cases as that class's principal, and to
142// ask for a commit once per build when commits is set.
143func refsMocks(t *testing.T, anonymous, owner int, commits bool) *Server {
144 t.Helper()
145 const id = "0123456789abcdef0123456789abcdef01234567"
146 repos := mocks.NewRepoService(t)
147 browse := mocks.NewBrowseService(t)
148 for _, class := range []struct {
149 p domain.Principal
150 builds int
151 }{{domain.Anonymous(), anonymous}, {domain.Owner(), owner}} {
152 p, builds := class.p, class.builds
153 if builds == 0 {
154 continue
155 }
156 repos.EXPECT().Get(mock.Anything, p, repoName(t, "x")).Return(publicRepo(t, "x").Repo, nil).Times(builds)
157 browse.EXPECT().ListRefs(mock.Anything, p, repoName(t, "x"), mock.MatchedBy(func(params ports.ListRefsParams) bool {
158 return params.PageSize == maxRefs // the refs are asked in one page
159 })).Return(ports.RefPage{Refs: []domain.Ref{
160 {Name: "refs/heads/main", ShortName: "main", Kind: domain.RefKindBranch, CommitID: objectID(t, id)},
161 {Name: "refs/tags/v1", ShortName: "v1", Kind: domain.RefKindTag, CommitID: objectID(t, id)},
162 }}, nil).Times(builds)
163 if commits {
164 browse.EXPECT().ListCommits(mock.Anything, p, repoName(t, "x"), mock.Anything).RunAndReturn(
165 func(_ context.Context, _ domain.Principal, _ domain.RepoName, params ports.ListCommitsParams) (ports.HistoryPage, error) {
166 return ports.HistoryPage{Commits: []domain.Commit{{ID: objectID(t, params.Revision.String()), Summary: "subject"}}}, nil
167 }).Times(builds)
168 }
169 }
170 repos.EXPECT().View(mock.Anything, mock.Anything).Return(publicRepo(t, "x"), nil).Times(anonymous + owner)
171 return newTestServer(t, Config{Repos: repos, Browse: browse})
172}
173
174// A push is rendered once for the watchers of the same class, and the owner
175// and the anonymous visitors never share what was built.
176func TestLiveRefsSharesAPushBetweenWatchers(t *testing.T) {
177 t.Parallel()
178 owner := domain.ContextWithPrincipal(context.WithValue(context.Background(), signedInKey{}, struct{}{}), domain.Owner())
179 anon := context.Background()
180
181 for _, tt := range []struct {
182 name string
183 ctxs []context.Context
184 anonymous, mine int // builds: one per class
185 }{
186 {"five anonymous watchers", []context.Context{anon, anon, anon, anon, anon}, 1, 0},
187 {"owner and anonymous", []context.Context{anon, owner, anon, owner}, 1, 1},
188 } {
189 t.Run(tt.name, func(t *testing.T) {
190 t.Parallel()
191 s := refsMocks(t, tt.anonymous, tt.mine, true)
192 ev := ports.WatchEvent{ID: "b-1", Event: domain.PushEvent{Repo: repoName(t, "x")}}
193 for _, ctx := range tt.ctxs {
194 st := &liveStream{s: s, repo: "x", view: "refs"}
195 frags, err := st.render(ctx, ev)
196 require.NoError(t, err)
197 require.Len(t, frags, 2)
198 assert.Equal(t, "refs-list", frags[0].event)
199 assert.Contains(t, frags[0].html, `id="ref-rows"`)
200 }
201 })
202 }
203}
204
205// The refs are asked in one page of up to maxRefs (the mock only answers that).
206func TestLoadRepoAsksTheRefsInOnePage(t *testing.T) {
207 t.Parallel()
208 s := refsMocks(t, 1, 0, false)
209 d, err := s.loadRepo(context.Background(), "x")
210 require.NoError(t, err)
211 assert.Len(t, d.branches, 1)
212 assert.Len(t, d.tags, 1)
213}