1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "context"
7 "net/http"
8 "strings"
9 "sync"
10 "sync/atomic"
11 "testing"
12 "time"
13
14 "connectrpc.com/connect"
15 "github.com/stretchr/testify/assert"
16 "github.com/stretchr/testify/require"
17
18 pb "bnema.dev/ermite-api/gen/ermite"
19 "bnema.dev/ermite/internal/domain"
20 "bnema.dev/ermite/internal/ports"
21 "bnema.dev/ermite/internal/usecase"
22)
23
24// pullHook is CIConfig.PullStage for the tests that interleave other changes
25// with the workers of the merge requests: armed for one stage, it runs fn (or
26// waits to be released) the next time a worker reaches it, once.
27type pullHook struct {
28 mu sync.Mutex
29 stage string
30 fn func()
31 reached chan struct{}
32 release chan struct{}
33}
34
35func (h *pullHook) call(stage string) {
36 h.mu.Lock()
37 if stage != h.stage {
38 h.mu.Unlock()
39 return
40 }
41 h.stage = ""
42 fn, reached, release := h.fn, h.reached, h.release
43 h.mu.Unlock()
44 if fn != nil {
45 fn()
46 return
47 }
48 close(reached)
49 <-release // held until the test releases it; its cleanup always does
50}
51
52// block arms the hook: the worker stops at stage and stays there until release
53// is called (the cleanup of the test calls it too, so a failed test never leaves
54// a worker stuck). reached is closed when the worker has stopped.
55func (h *pullHook) block(t *testing.T, stage string) (reached <-chan struct{}, release func()) {
56 t.Helper()
57 r, rel := make(chan struct{}), make(chan struct{})
58 h.mu.Lock()
59 h.stage, h.fn, h.reached, h.release = stage, nil, r, rel
60 h.mu.Unlock()
61 var once sync.Once
62 release = func() { once.Do(func() { close(rel) }) }
63 t.Cleanup(release)
64 return r, release
65}
66
67// await waits for the worker to reach a blocked stage, failing the test when it
68// does not within the bound.
69func await(t *testing.T, reached <-chan struct{}, what string) {
70 t.Helper()
71 select {
72 case <-reached:
73 case <-time.After(15 * time.Second):
74 t.Fatalf("the worker never reached %s", what)
75 }
76}
77
78// run arms the hook with a function the worker runs at stage.
79func (h *pullHook) run(stage string, fn func()) {
80 h.mu.Lock()
81 h.stage, h.fn = stage, fn
82 h.mu.Unlock()
83}
84
85func newHookedHarness(t *testing.T) (*ciHarness, *pullHook) {
86 t.Helper()
87 h := &pullHook{}
88 return newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PullStage = h.call }), h
89}
90
91func (c *ciHarness) parents(sha string) string {
92 c.t.Helper()
93 return strings.TrimSpace(c.git2("log", "-1", "--format=%P", sha))
94}
95
96// S1: an event runs for the commits it was made for, with the trust of whoever
97// caused it, however the branch moves before the event is handled.
98func TestPullEventKeepsItsCommitsAndTrustWhenALimitedTokenPushesNext(t *testing.T) {
99 c, hook := newHookedHarness(t)
100 runner := c.registerRunner("alpha", "ubuntu-latest")
101 main := strings.TrimSpace(c.push("main", pullWF("", ""), ""))
102 c.mustGit(c.wt, "checkout", "-q", "-b", "agent/x")
103 x1 := c.push("agent/x", map[string]string{"x1.txt": "1\n"}, "")
104 secret, _ := c.branchToken(scopes(domain.ScopeRepoRead, domain.ScopeRepoWrite), []string{"agent/"}, "proj")
105
106 reached, release := hook.block(t, "event")
107 n := c.openMR(c.root, "agent/x") // the owner opens it: trusted
108 await(t, reached, "the opening event")
109 // While the owner's event waits, the agent pushes a new commit.
110 c.writeFile(c.wt, "x2.txt", []byte("2\n"))
111 x2 := strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
112 c.mustGit(c.wt, "push", "-q", c.urlAuth(secret, "/proj.git"), "agent/x")
113 release()
114
115 c.waitFor("both runs", func() bool { return len(c.runs()) == 2 })
116 runs := c.runs() // newest first: the synchronize, then the opened
117 opened, sync := runs[1], runs[0]
118 assert.True(t, opened.GetTrusted(), "the owner's event keeps the owner's trust")
119 assert.Equal(t, main+" "+x1, c.parents(opened.GetSha()), "and its own commits, not the agent's later one")
120 assert.False(t, sync.GetTrusted(), "the agent's push is the agent's")
121 assert.Equal(t, main+" "+x2, c.parents(sync.GetSha()))
122 assert.Equal(t, "claude", sync.GetActor())
123 // The refs are the latest ones.
124 c.waitFor("the refs follow the latest push", func() bool {
125 return c.ref(domain.PullHeadRef(n)) == x2 && strings.HasSuffix(c.parents(c.ref(domain.PullMergeRef(n))), " "+x2)
126 })
127 _ = runner
128}
129
130// S1: a push made with a job token starts no pull_request run, not even through
131// the event of an earlier opening.
132func TestPullEventIsNotRunForAJobTokenPush(t *testing.T) {
133 c, hook := newHookedHarness(t)
134 runner := c.registerRunner("alpha", "ubuntu-latest")
135 files := pullWF("", "")
136 files[".github/workflows/push.yml"] = "on:\n push:\n branches: [feat]\npermissions:\n contents: write\njobs:\n j:\n runs-on: ubuntu-latest\n steps:\n - run: echo push\n"
137 main := strings.TrimSpace(c.push("main", files, ""))
138 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
139 f1 := c.push("feat", map[string]string{"f1.txt": "1\n"}, "")
140 pushJob := runner.await() // the push workflow's job holds a token with contents: write
141
142 reached, release := hook.block(t, "event")
143 c.openMR(c.root, "feat")
144 await(t, reached, "the opening event")
145 c.writeFile(c.wt, "f2.txt", []byte("2\n"))
146 out, err := c.tokenPush(c.wt, pushJob.GetGithubToken(), "HEAD:refs/heads/feat")
147 require.NoError(t, err, out)
148 release()
149
150 prJob := runner.await()
151 runner.runJob(prJob, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
152 runner.runJob(pushJob, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
153 var pulls []*pb.Run
154 for _, r := range c.runs() {
155 if r.GetEvent() == "pull_request" {
156 pulls = append(pulls, r)
157 }
158 }
159 require.Len(t, pulls, 1, "the job's push started no run")
160 assert.Equal(t, main+" "+f1, c.parents(pulls[0].GetSha()))
161 time.Sleep(200 * time.Millisecond)
162 for _, r := range c.runs() {
163 assert.NotEqual(t, "synchronize", r.GetTitle())
164 }
165 assert.Len(t, c.runs(), 2, "the push run and the opened run")
166}
167
168// S4: a request closed while its merge is being computed gets neither the refs
169// nor the run of the opening, and the closed run does not use that merge.
170func TestPullCloseDuringComputeDiscardsTheOpening(t *testing.T) {
171 c, hook := newHookedHarness(t)
172 runner := c.registerRunner("alpha", "ubuntu-latest")
173 c.push("main", pullWF(" types: [opened, closed]", ""), "")
174 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
175 f1 := c.push("feat", map[string]string{"f.txt": "f\n"}, "")
176
177 hook.run("computed", func() {
178 _, err := c.root.mr.CloseMergeRequest(c.ctx, connect.NewRequest(&pb.CloseMergeRequestRequest{Repo: "proj", Number: 1}))
179 assert.NoError(t, err)
180 })
181 n := c.openMR(c.root, "feat")
182 job := runner.await()
183 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
184 gh := c.ghContext(job)
185 assert.Equal(t, "closed", gh["event"].(map[string]any)["action"])
186 assert.Equal(t, f1, gh["sha"], "no test merge was published: the closed run uses the head")
187 assert.Empty(t, c.ref(domain.PullMergeRef(n)), "the merge ref of a closed request is never published")
188 assert.Empty(t, c.ref(domain.PullHeadRef(n)))
189 time.Sleep(200 * time.Millisecond)
190 assert.Len(t, c.runs(), 1, "no run for the opening")
191}
192
193// E1: a slow refresh of a merge request's refs does not delay the push
194// workflows.
195func TestPullSlowRefreshDoesNotDelayPushRuns(t *testing.T) {
196 c, hook := newHookedHarness(t)
197 runner := c.registerRunner("alpha", "ubuntu-latest")
198 both := "name: both\non:\n push:\n branches: [main]\n pull_request:\njobs:\n j:\n runs-on: ubuntu-latest\n steps:\n - run: echo hi\n"
199 c.push("main", map[string]string{".github/workflows/both.yml": both}, "")
200 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
201 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
202 c.push("feat", map[string]string{"f.txt": "f\n"}, "")
203 n := c.openMR(c.root, "feat")
204 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
205 c.waitFor("the merge ref", func() bool { return c.ref(domain.PullMergeRef(n)) != "" })
206
207 reached, release := hook.block(t, "refresh")
208 c.mustGit(c.wt, "checkout", "-q", "main")
209 m2 := c.push("main", map[string]string{"m2.txt": "m\n"}, "") // the target moves
210 await(t, reached, "the refresh") // the refresh worker is stuck
211 c.waitFor("the push run of main, while the refresh waits", func() bool {
212 for _, r := range c.runs() {
213 if r.GetEvent() == "push" && r.GetSha() == m2 {
214 return true
215 }
216 }
217 return false
218 })
219 release()
220 c.waitFor("the refresh then follows", func() bool {
221 return strings.HasPrefix(c.parents(c.ref(domain.PullMergeRef(n))), m2+" ")
222 })
223}
224
225// pushRuns counts the push runs, without failing the test (it runs in the hook
226// of a worker).
227func (c *ciHarness) pushRuns() int {
228 res, err := c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: c.repo}))
229 if err != nil {
230 return -1
231 }
232 n := 0
233 for _, r := range res.Msg.GetRuns() {
234 if r.GetEvent() == "push" {
235 n++
236 }
237 }
238 return n
239}
240
241// A push's workflows are queued before its merge requests are looked at: when
242// the discovery starts, the push run is already being created, so a slow
243// discovery never delays it.
244func TestPullDiscoveryDoesNotDelayPushRuns(t *testing.T) {
245 c, hook := newHookedHarness(t)
246 runner := c.registerRunner("alpha", "ubuntu-latest")
247 both := "name: both\non:\n push:\n branches: [main]\n pull_request:\njobs:\n j:\n runs-on: ubuntu-latest\n steps:\n - run: echo hi\n"
248 c.push("main", map[string]string{".github/workflows/both.yml": both}, "")
249 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
250 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
251 c.push("feat", map[string]string{"f.txt": "f\n"}, "")
252 n := c.openMR(c.root, "feat")
253 runner.runJob(runner.await(), pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
254 c.waitFor("the merge ref", func() bool { return c.ref(domain.PullMergeRef(n)) != "" })
255 c.waitFor("the workers are idle", func() bool { return len(c.runs()) == 2 })
256
257 before := c.pushRuns()
258 var sawRun atomic.Bool
259 hook.run("discover", func() {
260 // Inside the discovery of the push: its push run must come into being
261 // while it goes on (the push handler waits here, up to the bound).
262 deadline := time.Now().Add(10 * time.Second)
263 for time.Now().Before(deadline) {
264 if c.pushRuns() > before {
265 sawRun.Store(true)
266 return
267 }
268 time.Sleep(10 * time.Millisecond)
269 }
270 })
271 c.mustGit(c.wt, "checkout", "-q", "main")
272 m2 := c.push("main", map[string]string{"m2.txt": "m\n"}, "")
273 assert.True(t, sawRun.Load(), "the push run was created while the discovery ran")
274 c.waitFor("the merge ref follows the target", func() bool {
275 return strings.HasPrefix(c.parents(c.ref(domain.PullMergeRef(n))), m2+" ")
276 })
277}
278
279// Past the bound of the refresh queue the refs are repaired by a rescan, and
280// the activities are never lost: every synchronize still runs.
281func TestPullSaturatedRefreshQueueKeepsEventsAndRepairsRefs(t *testing.T) {
282 h := &pullHook{}
283 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PullStage = h.call; cfg.PullRefreshes = 1 })
284 runner := c.registerRunner("alpha", "ubuntu-latest")
285 _ = runner
286 c.push("main", pullWF("", ""), "")
287 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
288 c.push("feat", map[string]string{"f.txt": "f\n"}, "")
289 n1 := c.openMR(c.root, "feat")
290 c.mustGit(c.wt, "checkout", "-q", "-b", "feat2", "main")
291 c.push("feat2", map[string]string{"g.txt": "g\n"}, "")
292 n2 := c.openMR(c.root, "feat2")
293 c.waitFor("both opened runs and both merge refs", func() bool {
294 return len(c.runs()) == 2 && c.ref(domain.PullMergeRef(n1)) != "" && c.ref(domain.PullMergeRef(n2)) != ""
295 })
296
297 reached, release := h.block(t, "refresh")
298 c.mustGit(c.wt, "checkout", "-q", "feat")
299 c.push("feat", map[string]string{"f2.txt": "2\n"}, "") // its refresh is taken by the worker, which stops
300 await(t, reached, "the first refresh")
301 c.mustGit(c.wt, "checkout", "-q", "feat2")
302 g2 := c.push("feat2", map[string]string{"g2.txt": "2\n"}, "") // waits: the queue is at its bound of one
303 c.mustGit(c.wt, "checkout", "-q", "feat")
304 f3 := c.push("feat", map[string]string{"f3.txt": "3\n"}, "") // over the bound: the repository is marked dirty
305 release()
306
307 c.waitFor("the three synchronize runs, none lost to the cap", func() bool { return len(c.runs()) == 5 })
308 c.waitFor("the refs end at the latest heads", func() bool {
309 return c.ref(domain.PullHeadRef(n1)) == f3 && strings.HasSuffix(c.parents(c.ref(domain.PullMergeRef(n1))), " "+f3) &&
310 c.ref(domain.PullHeadRef(n2)) == g2 && strings.HasSuffix(c.parents(c.ref(domain.PullMergeRef(n2))), " "+g2)
311 })
312}
313
314// GO-010: the target tip an event is tested against is read when the push
315// happens: a target that moves before the event is handled changes nothing for
316// the run, which keeps the merge of the commits it was made for.
317func TestPullEventKeepsTheTargetTipOfItsPush(t *testing.T) {
318 c, hook := newHookedHarness(t)
319 c.registerRunner("alpha", "ubuntu-latest")
320 m1 := c.push("main", pullWF("", ""), "")
321 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
322 c.push("feat", map[string]string{"f.txt": "f\n"}, "")
323 c.openMR(c.root, "feat")
324 c.waitFor("the opened run", func() bool { return len(c.runs()) == 1 })
325
326 reached, release := hook.block(t, "event")
327 f2 := c.push("feat", map[string]string{"f2.txt": "2\n"}, "") // the synchronize event is taken and stops
328 await(t, reached, "the synchronize event")
329 c.mustGit(c.wt, "checkout", "-q", "main")
330 m2 := c.push("main", map[string]string{"m2.txt": "m\n"}, "") // the target moves meanwhile
331 release()
332
333 c.waitFor("the synchronize run", func() bool { return len(c.runs()) == 2 })
334 assert.Equal(t, m1+" "+f2, c.parents(c.runs()[0].GetSha()), "the run merges the target as it was at the push, not as it is now")
335 assert.NotEqual(t, m2, strings.Fields(c.parents(c.runs()[0].GetSha()))[0])
336 c.waitFor("the merge ref follows the target", func() bool {
337 return c.parents(c.ref(domain.PullMergeRef(1))) == m2+" "+f2
338 })
339}
340
341// countingPull counts the test merges the CI asks of Git.
342type countingPull struct {
343 ports.PullGit
344 merges atomic.Int32
345}
346
347func (p *countingPull) TestMerge(ctx context.Context, repo domain.RepoName, base, head domain.ObjectID) (domain.ObjectID, bool, error) {
348 p.merges.Add(1)
349 return p.PullGit.TestMerge(ctx, repo, base, head)
350}
351
352// GO-009: conflicting merge requests are not tried again by a rescan until a
353// branch moves, even when there are more of them than the memory holds, and the
354// rescan neither loops nor marks its repository dirty again.
355func TestPullRescanDoesNotRetryConflicts(t *testing.T) {
356 p := &countingPull{}
357 var rescans atomic.Int32
358 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
359 p.PullGit = cfg.Pull
360 cfg.Pull = p
361 cfg.PullAttempts = 2
362 cfg.PullStage = func(stage string) {
363 if stage == "rescan" {
364 rescans.Add(1)
365 }
366 }
367 })
368 c.push("main", map[string]string{"a.txt": "base\n"}, "")
369 const conflicting = 4
370 for i := range conflicting {
371 b := "c" + string(rune('a'+i))
372 c.mustGit(c.wt, "checkout", "-q", "-b", b, "main")
373 c.push(b, map[string]string{"a.txt": b + "\n"}, "")
374 }
375 c.mustGit(c.wt, "checkout", "-q", "main")
376 c.push("main", map[string]string{"a.txt": "main\n"}, "") // every branch now conflicts with main
377 for i := range conflicting {
378 c.openMR(c.root, "c"+string(rune('a'+i)))
379 }
380 c.waitFor("every conflict was tried", func() bool { return p.merges.Load() >= conflicting })
381 time.Sleep(300 * time.Millisecond)
382 for i := range conflicting {
383 assert.Empty(t, c.ref(domain.PullMergeRef(int64(i+1))), "a conflict has no merge ref")
384 }
385
386 // A rescan (what a saturated queue or a restart leads to) tries again only
387 // what the memory forgot: at most the number of merge requests, and it ends.
388 before, scans := p.merges.Load(), rescans.Load()
389 require.NoError(t, c.forge.CI.RescanPulls(c.ctx, c.name("proj")))
390 assert.LessOrEqual(t, p.merges.Load()-before, int32(conflicting))
391 time.Sleep(300 * time.Millisecond)
392 after := p.merges.Load()
393 assert.Equal(t, scans+1, rescans.Load(), "the rescan was not marked dirty again")
394 assert.Equal(t, after, p.merges.Load(), "and nothing goes on trying")
395}
396
397// readingMRs counts what the CI asks of the merge request store: the full
398// listings and the single reads of requests.
399type readingMRs struct {
400 ports.MergeRequestStore
401 lists, opens, gets atomic.Int32
402}
403
404func (s *readingMRs) List(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
405 s.lists.Add(1)
406 return s.MergeRequestStore.List(ctx, repo)
407}
408
409func (s *readingMRs) ListOpen(ctx context.Context, repo domain.RepoName) ([]domain.MergeRequest, error) {
410 s.opens.Add(1)
411 return s.MergeRequestStore.ListOpen(ctx, repo)
412}
413
414func (s *readingMRs) Get(ctx context.Context, repo domain.RepoName, n int64) (domain.MergeRequest, error) {
415 s.gets.Add(1)
416 return s.MergeRequestStore.Get(ctx, repo, n)
417}
418
419// GO-011: the snapshot of a push looks up the open merge requests of the pushed
420// branches in the index and reads those only: the many closed ones of the
421// repository are not read, and the repository's merge requests are not listed.
422func TestPullPushDoesNotReadHistoricalMergeRequests(t *testing.T) {
423 store := &readingMRs{}
424 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
425 store.MergeRequestStore = cfg.MRs
426 cfg.MRs = store
427 })
428 c.registerRunner("alpha", "ubuntu-latest")
429 c.push("main", pullWF("", ""), "")
430 const closed = 30
431 for i := range closed {
432 b := "old" + string(rune('a'+i%26)) + string(rune('a'+i/26))
433 c.mustGit(c.wt, "checkout", "-q", "-b", b, "main")
434 c.push(b, map[string]string{b + ".txt": "x\n"}, "")
435 n := c.openMR(c.root, b)
436 _, err := c.root.mr.CloseMergeRequest(c.ctx, connect.NewRequest(&pb.CloseMergeRequestRequest{Repo: "proj", Number: n}))
437 require.NoError(t, err)
438 }
439 c.mustGit(c.wt, "checkout", "-q", "-b", "live", "main")
440 c.push("live", map[string]string{"l.txt": "l\n"}, "")
441 live := c.openMR(c.root, "live")
442 runsOf := func() int {
443 n := 0
444 for _, r := range c.runs() {
445 if r.GetRef() == domain.PullMergeRef(live) {
446 n++
447 }
448 }
449 return n
450 }
451 // (The closed requests may or may not have had their opened run: they were
452 // closed at once.) The live one has its own.
453 c.waitFor("the opened run of the live request", func() bool { return runsOf() == 1 })
454 time.Sleep(300 * time.Millisecond)
455
456 lists, gets := store.lists.Load(), store.gets.Load()
457 c.push("live", map[string]string{"l2.txt": "2\n"}, "")
458 c.waitFor("the synchronize run of the live request", func() bool { return runsOf() == 2 })
459 time.Sleep(300 * time.Millisecond)
460 assert.Zero(t, store.lists.Load()-lists, "the merge requests of the repository are not listed for a push")
461 // The synchronize snapshot reads the one open request on the pushed branch (and
462 // the workers read it again to handle the event and refresh the refs): a
463 // small number of reads, nothing like the thirty closed ones.
464 assert.Less(t, store.gets.Load()-gets, int32(closed), "the closed requests are not read")
465}
466
467// GO-011: past the bound of one push's snapshot the rest is skipped, an error
468// naming the repository and the number skipped is logged, and the refs of the
469// skipped requests are repaired by the rescan.
470func TestPullSnapshotBoundSkipsLogsAndMarksDirty(t *testing.T) {
471 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PullSnapshot = 2 })
472 c.registerRunner("alpha", "ubuntu-latest")
473 c.push("main", pullWF("", ""), "")
474 c.mustGit(c.wt, "checkout", "-q", "-b", "shared")
475 c.push("shared", map[string]string{"s.txt": "s\n"}, "")
476 const requests = 5
477 var numbers []int64
478 for i := range requests {
479 // Several open requests on the same source branch (distinct targets).
480 t0 := "t" + string(rune('a'+i))
481 c.mustGit(c.wt, "checkout", "-q", "-b", t0, "main")
482 c.push(t0, map[string]string{t0 + ".txt": "t\n"}, "")
483 res, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{
484 Repo: "proj", Title: "t" + t0, SourceBranch: "shared", TargetBranch: t0,
485 }))
486 require.NoError(t, err)
487 numbers = append(numbers, res.Msg.GetMergeRequest().GetNumber())
488 }
489 c.waitFor("the opened runs", func() bool { return len(c.runs()) == requests })
490
491 c.mustGit(c.wt, "checkout", "-q", "shared")
492 latest := c.push("shared", map[string]string{"s2.txt": "2\n"}, "")
493
494 c.waitFor("the error naming the repository and the skipped requests", func() bool {
495 logs := c.logs.String()
496 return strings.Contains(logs, "snapshot of a push is cut short") && strings.Contains(logs, "more than the bound") &&
497 strings.Contains(logs, `"bound":2`) && strings.Contains(logs, `"repo":"proj"`) && !strings.Contains(logs, `"skipped"`)
498 })
499 c.waitFor("only the bounded part got a synchronize run", func() bool { return len(c.runs()) == requests+2 })
500 time.Sleep(300 * time.Millisecond)
501 assert.Len(t, c.runs(), requests+2, "the three others were skipped")
502 // The refs of every request still end at the pushed head: the repository was
503 // marked dirty and the rescan refreshed them.
504 for _, n := range numbers {
505 c.waitFor("the head ref of a skipped request is repaired", func() bool {
506 return c.ref(domain.PullHeadRef(n)) == latest
507 })
508 }
509}
510
511// GO-013: a push that touches no merge request (not a source, not a target)
512// queues no refresh and marks nothing for a rescan; a push of a target queues the
513// refresh of exactly the requests that target it.
514func TestPullUnrelatedPushQueuesNoRefreshAndMarksNothing(t *testing.T) {
515 var refreshes, rescans atomic.Int32
516 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
517 cfg.PullStage = func(stage string) {
518 switch stage {
519 case "refresh":
520 refreshes.Add(1)
521 case "rescan":
522 rescans.Add(1)
523 }
524 }
525 })
526 c.registerRunner("alpha", "ubuntu-latest")
527 c.push("main", pullWF("", ""), "")
528 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
529 c.push("feat", map[string]string{"f.txt": "f\n"}, "")
530 n := c.openMR(c.root, "feat")
531 c.waitFor("the merge ref", func() bool { return c.ref(domain.PullMergeRef(n)) != "" })
532 time.Sleep(300 * time.Millisecond)
533
534 // Pushes of branches that are neither the source nor the target of any open
535 // merge request: nothing is queued, nothing is marked.
536 beforeRefresh, beforeRescan := refreshes.Load(), rescans.Load()
537 for _, b := range []string{"unrelated", "other/topic", "another"} {
538 c.mustGit(c.wt, "checkout", "-q", "-b", b, "main")
539 c.push(b, map[string]string{b + ".txt": "x\n"}, "")
540 }
541 time.Sleep(500 * time.Millisecond)
542 assert.Equal(t, beforeRefresh, refreshes.Load(), "no refresh for pushes that touch no merge request")
543 assert.Equal(t, beforeRescan, rescans.Load(), "and no repository marked for a rescan")
544
545 // A push of the target queues the refresh of the request that targets it.
546 c.mustGit(c.wt, "checkout", "-q", "main")
547 m2 := c.push("main", map[string]string{"m2.txt": "m\n"}, "")
548 c.waitFor("the merge ref follows the target", func() bool {
549 return strings.HasPrefix(c.parents(c.ref(domain.PullMergeRef(n))), m2+" ")
550 })
551 assert.Greater(t, refreshes.Load(), beforeRefresh, "a refresh was queued for the target move")
552 assert.Equal(t, beforeRescan, rescans.Load(), "through the index, not a rescan")
553}
554
555// GO-013: more merge requests target a pushed branch than the bound: the lookup
556// is cut, an error is logged, and the repository is rescanned, so that every
557// request ends with a merge ref on the new target.
558func TestPullTargetLookupTruncationMarksDirty(t *testing.T) {
559 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PullSnapshot = 2 })
560 c.registerRunner("alpha", "ubuntu-latest")
561 c.push("main", pullWF("", ""), "")
562 const requests = 5
563 var numbers []int64
564 for i := range requests {
565 b := "f" + string(rune('a'+i))
566 c.mustGit(c.wt, "checkout", "-q", "-b", b, "main")
567 c.push(b, map[string]string{b + ".txt": "x\n"}, "")
568 numbers = append(numbers, c.openMR(c.root, b))
569 }
570 c.waitFor("the opened runs", func() bool { return len(c.runs()) == requests })
571 c.mustGit(c.wt, "checkout", "-q", "main")
572 m2 := c.push("main", map[string]string{"m2.txt": "m\n"}, "")
573 c.waitFor("the truncation is logged", func() bool {
574 logs := c.logs.String()
575 return strings.Contains(logs, "refresh of a push is cut short") && strings.Contains(logs, "more than the bound") && strings.Contains(logs, `"repo":"proj"`)
576 })
577 for _, n := range numbers {
578 c.waitFor("the merge ref of a request is on the new target", func() bool {
579 return strings.HasPrefix(c.parents(c.ref(domain.PullMergeRef(n))), m2+" ")
580 })
581 }
582}
583
584// countingGit counts the diffs the CI asks of Git.
585type countingGit struct {
586 ports.MergeGit
587 diffs atomic.Int32
588}
589
590func (g *countingGit) DiffCommits(ctx context.Context, repo domain.RepoName, base, head domain.ObjectID, l domain.DiffLimits) (domain.DiffStats, domain.Diff, error) {
591 g.diffs.Add(1)
592 return g.MergeGit.DiffCommits(ctx, repo, base, head, l)
593}
594
595// E2: the pages of one merge request share the diff.
596func TestPullFilesPagesShareOneDiff(t *testing.T) {
597 g := &countingGit{}
598 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
599 g.MergeGit = cfg.Git
600 cfg.Git = g
601 })
602 c.push("main", pullWF("", ""), "")
603 c.mustGit(c.wt, "checkout", "-q", "-b", "feat")
604 for i := range 5 {
605 c.writeFile(c.wt, "d/"+string(rune('a'+i))+".txt", []byte("x\n"))
606 }
607 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "feat")
608 c.openMR(c.root, "feat")
609 for _, q := range []string{"?per_page=2&page=1", "?per_page=2&page=2", "?per_page=2&page=3", ""} {
610 code, _, _ := c.apiGet(adminToken, "/repos/owner/proj/pulls/1/files"+q)
611 require.Equal(t, http.StatusOK, code)
612 }
613 assert.EqualValues(t, 1, g.diffs.Load(), "one diff for four pages")
614 // A push to the branch is another pair of commits: another diff.
615 c.writeFile(c.wt, "d/z.txt", []byte("z\n"))
616 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "feat")
617 code, _, _ := c.apiGet(adminToken, "/repos/owner/proj/pulls/1/files")
618 require.Equal(t, http.StatusOK, code)
619 assert.EqualValues(t, 2, g.diffs.Load())
620}