1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "context"
7 "encoding/json"
8 "fmt"
9 "net/http/httptest"
10 "os"
11 "path/filepath"
12 "strings"
13 "testing"
14 "time"
15
16 "connectrpc.com/connect"
17 "github.com/stretchr/testify/assert"
18 "github.com/stretchr/testify/require"
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
26const oneJob = `on:
27 push:
28 %s
29jobs:
30 j:
31 runs-on: ubuntu-latest
32 steps:
33 - run: echo hi
34`
35
36func workflowWith(filter string) string { return fmt.Sprintf(oneJob, filter) }
37
38// AC3: filters, tags, skip directives, invalid files and deletions.
39func TestCITriggerRules(t *testing.T) {
40 c := newCIHarness(t, nil)
41 c.registerRunner("alpha", "ubuntu-latest")
42 count := func() int { return len(c.runs()) }
43 settle := func() { time.Sleep(300 * time.Millisecond) } // a push that runs nothing leaves nothing to wait for
44
45 // A workflow with a branch filter does not run for another branch.
46 c.push("main", map[string]string{".github/workflows/only-main.yml": workflowWith("branches: [main]"), "src/a.go": "package a\n"}, "")
47 c.waitFor("main runs", func() bool { return count() == 1 })
48 c.push("other", map[string]string{"x.txt": "x\n"}, "")
49 settle()
50 assert.Equal(t, 1, count(), "branches: [main] does not run for other")
51
52 // A paths filter looks at the files of the push.
53 c.push("main", map[string]string{".github/workflows/paths.yml": workflowWith("paths: ['docs/**']")}, "")
54 c.waitFor("only-main runs again for the new commit", func() bool { return count() == 2 })
55 c.push("main", map[string]string{"docs/readme.md": "doc\n"}, "")
56 c.waitFor("both run for docs", func() bool { return count() == 4 })
57 c.push("main", map[string]string{"src/b.go": "package b\n"}, "")
58 c.waitFor("only-main runs for src", func() bool { return count() == 5 })
59 settle()
60 assert.Equal(t, 5, count(), "the paths filter kept paths.yml out")
61
62 // [skip ci] on the head commit.
63 c.push("main", map[string]string{"src/c.go": "package c\n"}, "wip [skip ci]")
64 settle()
65 assert.Equal(t, 5, count())
66
67 // An invalid workflow file is a failed run without a job.
68 c.push("main", map[string]string{".github/workflows/broken.yml": "on: push\njobs:\n j:\n steps: [\n"}, "")
69 c.waitFor("the invalid file gives a run", func() bool {
70 for _, r := range c.runs() {
71 if r.GetWorkflowPath() == ".github/workflows/broken.yml" {
72 return true
73 }
74 }
75 return false
76 })
77 var broken *pb.Run
78 for _, r := range c.runs() {
79 if r.GetWorkflowPath() == ".github/workflows/broken.yml" {
80 broken = r
81 }
82 }
83 assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, broken.GetStatus())
84 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, broken.GetConclusion())
85 assert.NotEmpty(t, broken.GetError())
86 assert.Empty(t, c.getRun(broken.GetId()).GetJobs(), "an invalid file has no job")
87 sha := strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
88 var failed []string
89 for _, ck := range c.checks(sha) {
90 if ck.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_FAILURE {
91 failed = append(failed, ck.GetName())
92 }
93 }
94 assert.Equal(t, []string{".github/workflows/broken.yml"}, failed, "and a failed check on the commit")
95
96 // A push never fails for CI: the push above succeeded (mustGit). A deletion
97 // runs nothing.
98 before := count()
99 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), ":other")
100 settle()
101 assert.Equal(t, before, count())
102}
103
104// AC3: tags run tag-filtered workflows; more than 3 tags in one push run none.
105func TestCITags(t *testing.T) {
106 c := newCIHarness(t, nil)
107 c.registerRunner("alpha", "ubuntu-latest")
108 c.push("main", map[string]string{".github/workflows/rel.yml": workflowWith("tags: ['v*']")}, "")
109 settle := func() { time.Sleep(300 * time.Millisecond) }
110 settle()
111 assert.Empty(t, c.runs(), "a tags-only workflow does not run for a branch")
112
113 c.mustGit(c.wt, "tag", "v1")
114 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "v1")
115 c.waitFor("a tag runs", func() bool { return len(c.runs()) == 1 })
116 assert.Equal(t, "refs/tags/v1", c.runs()[0].GetRef())
117
118 for _, tag := range []string{"v2", "v3", "v4", "v5"} {
119 c.mustGit(c.wt, "tag", tag)
120 }
121 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "v2", "v3", "v4", "v5")
122 settle()
123 assert.Len(t, c.runs(), 1, "more than 3 tags in one push run nothing")
124}
125
126// AC6: cancelling a run asks the runner to stop through WatchJob; a runner that
127// does not answer is ended by the server.
128func TestCICancelReachesTheRunnerAndIsEnforced(t *testing.T) {
129 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
130 cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 400*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
131 })
132 runner := c.registerRunner("alpha", "ubuntu-latest")
133 c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
134 job := runner.await()
135 api := runner.jobAPI(job)
136 _, err := api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
137 require.NoError(t, err)
138
139 watch, err := api.runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: job.GetJobId()}))
140 require.NoError(t, err)
141 run := c.runs()[0]
142
143 // A reader without ci:write cannot cancel.
144 reader := c.api(c.token(scopes(domain.ScopeRepoRead)))
145 _, err = reader.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
146 assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
147
148 _, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
149 require.NoError(t, err)
150 // Cancelling again changes nothing.
151 _, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
152 require.NoError(t, err)
153
154 var cancel *pb.CancelRequest
155 for watch.Receive() {
156 if m := watch.Msg().GetCancel(); m != nil {
157 cancel = m
158 break
159 }
160 }
161 require.NotNil(t, cancel, "the runner is told")
162 assert.Equal(t, pb.CancelReason_CANCEL_REASON_USER, cancel.GetReason())
163
164 // The queued job that needs it never ran: cancelled at once. The running
165 // one is ended by the server after the grace period.
166 c.waitFor("the server ends the job", func() bool {
167 return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
168 })
169 detail := c.getRun(run.GetId())
170 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, detail.GetRun().GetConclusion())
171 for _, j := range detail.GetJobs() {
172 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, j.GetConclusion(), j.GetKey())
173 }
174 assert.Contains(t, detail.GetJobs()[0].GetError(), "did not answer")
175
176 // The runner's late CompleteJob returns the server's result.
177 res, err := api.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
178 JobId: job.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
179 }))
180 require.NoError(t, err)
181 assert.False(t, res.Msg.GetApplied())
182
183 // A completed run cannot be cancelled.
184 _, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
185 assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
186}
187
188// A running job without heartbeats fails with GitHub's lost-runner message.
189func TestCILostRunnerFailsTheJob(t *testing.T) {
190 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
191 cfg.HeartbeatDeadline, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
192 })
193 runner := c.registerRunner("alpha", "ubuntu-latest")
194 c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
195 job := runner.await()
196 _, err := runner.jobAPI(job).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
197 require.NoError(t, err)
198
199 c.waitFor("the job is lost", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
200 j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
201 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
202 assert.Contains(t, j.GetError(), "The self-hosted runner: alpha lost communication with the server")
203}
204
205// AC6: a re-run is a new attempt of the same run; the failed-only variant keeps
206// the jobs that passed.
207func TestCIRerun(t *testing.T) {
208 c := newCIHarness(t, nil)
209 runner := c.registerRunner("alpha", "ubuntu-latest")
210 c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
211 run := func() *pb.Run { return c.runs()[0] }
212
213 build := runner.await()
214 runner.runJob(build, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
215 test := runner.await()
216 runner.runJob(test, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "red")
217 c.waitFor("the run fails", func() bool { return run().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
218 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, run().GetConclusion())
219 id, number := run().GetId(), run().GetNumber()
220
221 // A running run cannot be re-run; a completed one can, by who has ci:write.
222 reader := c.api(c.token(scopes(domain.ScopeRepoRead)))
223 _, err := reader.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id}))
224 assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
225
226 res, err := c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id, FailedOnly: true}))
227 require.NoError(t, err)
228 assert.Equal(t, id, res.Msg.GetRun().GetId())
229 assert.Equal(t, number, res.Msg.GetRun().GetNumber())
230 assert.EqualValues(t, 2, res.Msg.GetRun().GetAttempt())
231
232 // build keeps its attempt-1 result; only test runs again.
233 again := runner.await()
234 assert.Equal(t, "test", again.GetJobKey())
235 assert.EqualValues(t, 2, again.GetAttempt())
236 runner.runJob(again, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "green")
237 c.waitFor("attempt 2 passes", func() bool {
238 r := run()
239 return r.GetAttempt() == 2 && r.GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
240 })
241 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, run().GetConclusion())
242 detail := c.getRun(id)
243 require.Len(t, detail.GetJobs(), 2)
244 assert.EqualValues(t, 1, detail.GetJobs()[0].GetAttempt(), "build was kept from attempt 1")
245 assert.EqualValues(t, 2, detail.GetJobs()[1].GetAttempt())
246
247 // Attempt 1 is still readable, with its failure.
248 first, err := c.root.run.GetRun(c.ctx, connect.NewRequest(&pb.GetRunRequest{Repo: "proj", RunId: id, Attempt: 1}))
249 require.NoError(t, err)
250 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, first.Msg.GetRun().GetConclusion())
251 assert.EqualValues(t, 2, first.Msg.GetLatestAttempt())
252
253 // Re-running a run with nothing failed is refused with failed-only.
254 _, err = c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id, FailedOnly: true}))
255 assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
256}
257
258// A matrix is expanded when its job is ready, with fail-fast.
259func TestCIMatrixAndFailFast(t *testing.T) {
260 c := newCIHarness(t, nil)
261 runner := c.registerRunner("alpha", "ubuntu-latest")
262 c.push("main", map[string]string{".github/workflows/m.yml": `on: push
263jobs:
264 build:
265 runs-on: ubuntu-latest
266 strategy:
267 matrix:
268 go: ['1.26', '1.27']
269 steps:
270 - run: echo ${{ matrix.go }}
271 after:
272 needs: build
273 if: always()
274 runs-on: ubuntu-latest
275 steps:
276 - run: echo done
277`}, "")
278 a := runner.await()
279 b := runner.await()
280 assert.Equal(t, "build", a.GetJobKey())
281 assert.JSONEq(t, `{"go":"1.26"}`, a.GetMatrix())
282 assert.JSONEq(t, `{"go":"1.27"}`, b.GetMatrix())
283 assert.Contains(t, a.GetContexts(), `"job-total":2`)
284
285 jobs := c.getRun(c.runs()[0].GetId()).GetJobs()
286 var names []string
287 for _, j := range jobs {
288 names = append(names, j.GetName())
289 }
290 assert.Equal(t, []string{"build (1.26)", "build (1.27)", "after"}, names)
291
292 // The first fails: fail-fast asks the second to stop.
293 apiB := runner.jobAPI(b)
294 _, err := apiB.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: b.GetJobId()}))
295 require.NoError(t, err)
296 watch, err := apiB.runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: b.GetJobId()}))
297 require.NoError(t, err)
298 runner.runJob(a, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "x")
299 var reason pb.CancelReason
300 for watch.Receive() {
301 if m := watch.Msg().GetCancel(); m != nil {
302 reason = m.GetReason()
303 break
304 }
305 }
306 assert.Equal(t, pb.CancelReason_CANCEL_REASON_FAIL_FAST, reason)
307 _, err = apiB.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{JobId: b.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED}))
308 require.NoError(t, err)
309
310 // "after" has if: always(), so it runs once both ended.
311 after := runner.await()
312 assert.Equal(t, "after", after.GetJobKey())
313 assert.Contains(t, after.GetContexts(), `"result":"failure"`)
314 runner.runJob(after, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "done")
315 c.waitFor("the run ends", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
316 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, c.runs()[0].GetConclusion())
317}
318
319// A job that calls a reusable workflow which is not there fails clearly, never silently.
320func TestCICallOfAMissingWorkflowFailsTheJob(t *testing.T) {
321 c := newCIHarness(t, nil)
322 c.registerRunner("alpha", "ubuntu-latest")
323 c.push("main", map[string]string{".github/workflows/u.yml": `on: push
324jobs:
325 j:
326 uses: ./.github/workflows/callee.yml
327`}, "")
328 c.waitFor("the run fails", func() bool { return len(c.runs()) == 1 && c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
329 j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
330 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
331 assert.Contains(t, j.GetError(), "Unable to find reusable workflow '.github/workflows/callee.yml'")
332}
333
334// A job for a label no runner has is "waiting for a runner" and fails after the
335// queue timeout.
336func TestCIQueueTimeout(t *testing.T) {
337 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.QueueTimeout, cfg.Tick = 300*time.Millisecond, 20*time.Millisecond })
338 c.registerRunner("alpha", "ubuntu-latest")
339 c.push("main", map[string]string{".github/workflows/q.yml": strings.Replace(workflowWith("branches: [main]"), "ubuntu-latest", "gpu", 1)}, "")
340 c.waitFor("the job fails", func() bool { return len(c.runs()) == 1 && c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
341 j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
342 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
343 assert.Contains(t, j.GetError(), "was not acquired")
344}
345
346// The runner API validates what it is given.
347func TestCIRunnerRegistrationRules(t *testing.T) {
348 c := newCIHarness(t, nil)
349 reg := func(api *clients, name string, replace bool) error {
350 _, err := api.runner.RegisterRunner(c.ctx, connect.NewRequest(&pb.RegisterRunnerRequest{
351 Name: name, Version: "v1", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux", Replace: replace,
352 }))
353 return err
354 }
355 assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(reg(c.api(""), "a", false)))
356 assert.Equal(t, connect.CodePermissionDenied, ciErrCode(reg(c.api(c.token(scopes(domain.ScopeRepoRead))), "a", false)))
357 assert.Equal(t, connect.CodePermissionDenied, ciErrCode(reg(c.api(c.token(scopes(domain.ScopeRunnerRegister), "proj")), "a", false)), "a repository-restricted token cannot register a runner")
358 reg1 := c.api(c.token(scopes(domain.ScopeRunnerRegister)))
359 require.NoError(t, reg(reg1, "a", false))
360 assert.Equal(t, connect.CodeAlreadyExists, ciErrCode(reg(reg1, "a", false)))
361 require.NoError(t, reg(reg1, "a", true), "replace")
362 assert.Equal(t, connect.CodeInvalidArgument, ciErrCode(reg(reg1, "bad name!", false)))
363
364 list, err := reg1.run.ListRunners(c.ctx, connect.NewRequest(&pb.ListRunnersRequest{}))
365 require.NoError(t, err)
366 require.Len(t, list.Msg.GetRunners(), 1)
367 assert.Equal(t, []string{"linux", "self-hosted", "x64"}, list.Msg.GetRunners()[0].GetLabels(), "implicit labels, sorted")
368}
369
370// AC2: after a restart of the server, running jobs go on with their tokens,
371// queued jobs stay queued, and a job whose runner kept quiet during the outage
372// is not failed for it.
373func TestCIRestartKeepsRunningAndQueuedJobs(t *testing.T) {
374 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.HeartbeatDeadline = 2 * time.Second })
375 runner := c.registerRunner("alpha", "ubuntu-latest")
376 c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
377 job := runner.await() // "build"; "test" waits for it
378 api := runner.jobAPI(job)
379 _, err := api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
380 require.NoError(t, err)
381 _, err = api.runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
382 JobId: job.GetJobId(), Step: 2, Offset: 0, Lines: []*pb.LogLine{{Time: now(), Text: "before the restart"}},
383 }))
384 require.NoError(t, err)
385
386 // Stop the server and start another on the same data directory.
387 cfg, hookExe := c.forge.cfg, binary(t)
388 c.srv.Close()
389 c.forge.Close(c.ctx)
390 f2, err := newForge(c.ctx, cfg, options{hookExe: hookExe, ci: func(cc *usecase.CIConfig) { cc.HeartbeatDeadline = 2 * time.Second }})
391 require.NoError(t, err)
392 t.Cleanup(func() { f2.Close(context.WithoutCancel(c.ctx)) })
393 srv := httptest.NewServer(f2.server.Handler())
394 t.Cleanup(srv.Close)
395 api2 := newClients(srv.URL+"/-/api", job.GetJobToken())
396
397 // The job token still works, and the log goes on from where it was.
398 _, err = api2.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
399 require.NoError(t, err, "the job token survives a restart")
400 next, err := api2.runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
401 JobId: job.GetJobId(), Step: 2, Offset: 1, Lines: []*pb.LogLine{{Time: now(), Text: "after the restart"}},
402 }))
403 require.NoError(t, err)
404 assert.EqualValues(t, 2, next.Msg.GetNextOffset())
405
406 root2 := newClients(srv.URL+"/-/api", adminToken)
407 runs, err := root2.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: "proj"}))
408 require.NoError(t, err)
409 require.Len(t, runs.Msg.GetRuns(), 1)
410 assert.Equal(t, pb.CiStatus_CI_STATUS_IN_PROGRESS, runs.Msg.GetRuns()[0].GetStatus())
411
412 // Finish the job on the new server: its dependant is planned and ready.
413 _, err = api2.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
414 JobId: job.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_SUCCESS,
415 }))
416 require.NoError(t, err)
417 got := make(chan *pb.Job, 1)
418 go func() {
419 for range 50 {
420 res, err := newClients(srv.URL+"/-/api", c.runnerSecret(runner)).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
421 if err == nil && res.Msg.GetJob() != nil {
422 got <- res.Msg.GetJob()
423 return
424 }
425 }
426 got <- nil
427 }()
428 select {
429 case j := <-got:
430 require.NotNil(t, j, "the runner token and the queue survive a restart")
431 assert.Equal(t, "test", j.GetJobKey())
432 case <-time.After(30 * time.Second):
433 t.Fatal("no job after the restart")
434 }
435}
436
437// AC5: a head without any check merges, and so does one whose checks passed; a
438// merge runs the push workflows on the target.
439func TestCIMergeWithoutChecksAndWithPassingChecks(t *testing.T) {
440 c := newCIHarness(t, nil)
441 runner := c.registerRunner("alpha", "ubuntu-latest")
442 c.push("main", map[string]string{"README": "x\n"}, "")
443 headOf := func(branch string) string { return strings.TrimSpace(c.bareGit("proj", "rev-parse", branch)) }
444 merge := func(n int64, head string) error {
445 _, err := c.root.mr.MergeMergeRequest(c.ctx, connect.NewRequest(&pb.MergeMergeRequestRequest{Repo: "proj", Number: n, HeadId: head}))
446 return err
447 }
448
449 // No workflow on the branch: no check, the merge goes through.
450 c.push("plain", map[string]string{"a.txt": "a\n"}, "")
451 mr1, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "plain", SourceBranch: "plain"}))
452 require.NoError(t, err)
453 require.NoError(t, merge(mr1.Msg.GetMergeRequest().GetNumber(), headOf("plain")))
454
455 // A branch with a workflow: merge after its check passed. The merge itself
456 // triggers the workflow on main (GitHub runs push workflows on a merge).
457 c.mustGit(c.wt, "checkout", "-q", "main")
458 c.mustGit(c.wt, "pull", "-q", c.urlAuth(adminToken, "/proj.git"), "main")
459 c.mustGit(c.wt, "checkout", "-q", "-b", "ci")
460 c.writeFile(c.wt, ".github/workflows/ci.yml", []byte(strings.Replace(workflowWith("branches: [main, ci]"), "echo hi", "echo ci", 1)))
461 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "ci")
462 head := headOf("ci")
463 c.waitFor("the run on ci", func() bool { return len(c.checks(head)) == 1 })
464 mr2, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "ci", SourceBranch: "ci"}))
465 require.NoError(t, err)
466 job := runner.await()
467 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
468 c.waitFor("the check passes", func() bool {
469 cs := c.checks(head)
470 return len(cs) == 1 && cs[0].GetConclusion() == pb.CiConclusion_CI_CONCLUSION_SUCCESS
471 })
472 require.NoError(t, merge(mr2.Msg.GetMergeRequest().GetNumber(), head))
473
474 // The fast-forward merge moved main to the same commit: the workflow runs
475 // again for main, with the owner as the actor.
476 c.waitFor("the merge triggers a run on main", func() bool {
477 for _, r := range c.runs() {
478 if r.GetRef() == "refs/heads/main" && r.GetSha() == head {
479 return true
480 }
481 }
482 return false
483 })
484 for _, r := range c.runs() {
485 if r.GetRef() == "refs/heads/main" {
486 assert.Equal(t, "owner", r.GetActor())
487 }
488 }
489}
490
491// Repository deletion removes the runs, the checks and the logs; a repository
492// created again starts at run number 1.
493func TestCIRepositoryDeletionRemovesRunsAndLogs(t *testing.T) {
494 c := newCIHarness(t, nil)
495 runner := c.registerRunner("alpha", "ubuntu-latest")
496 sha := c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
497 job := runner.await()
498 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "line")
499 logDir := filepath.Join(c.data, "ci", "logs")
500 entries, err := os.ReadDir(logDir)
501 require.NoError(t, err)
502 require.NotEmpty(t, entries, "the log is a file under ci/logs")
503
504 _, err = c.root.repo.DeleteRepo(c.ctx, connect.NewRequest(&pb.DeleteRepoRequest{Repo: "proj"}))
505 require.NoError(t, err)
506 _, err = c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: "proj"}))
507 assert.Equal(t, connect.CodeNotFound, ciErrCode(err))
508 entries, err = os.ReadDir(logDir)
509 require.NoError(t, err)
510 assert.Empty(t, entries, "the logs are removed with the repository")
511
512 _, err = c.forge.Repos.Create(c.ctx, domain.Root(), ports.CreateRepoParams{Name: c.name("proj"), Visibility: domain.VisibilityPrivate})
513 require.NoError(t, err)
514 assert.Empty(t, c.checks(sha), "no check survives the repository")
515 c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0] + "# again\n"}, "again")
516 c.waitFor("a run", func() bool { return len(c.runs()) == 1 })
517 assert.EqualValues(t, 1, c.runs()[0].GetNumber(), "run numbers start again with the repository")
518 assert.Greater(t, c.runs()[0].GetId(), int64(1), "run ids are never reused")
519}
520
521// The events of a run and its jobs are published on the repository topic;
522// "home", which shows the CI state of each repository, gets the runs and the
523// jobs that ended only.
524func TestCIEventsAreStreamedOnTheRepositoryTopic(t *testing.T) {
525 c := newCIHarness(t, nil)
526 runner := c.registerRunner("alpha", "ubuntu-latest")
527 s := c.watch(c.root, "", "repo:proj")
528 c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
529 job := runner.await()
530 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "x")
531
532 var runUpdates, jobUpdates int
533 var lastRun *pb.Run
534 deadline := time.After(10 * time.Second)
535 for lastRun.GetStatus() != pb.CiStatus_CI_STATUS_COMPLETED {
536 select {
537 case ev := <-s.events:
538 if r := ev.GetRunUpdated(); r != nil {
539 runUpdates++
540 if r.GetRun().GetVersion() >= lastRun.GetVersion() {
541 lastRun = r.GetRun()
542 }
543 }
544 if ev.GetJobUpdated() != nil {
545 jobUpdates++
546 }
547 case <-deadline:
548 t.Fatalf("no completed run event: %d run and %d job events", runUpdates, jobUpdates)
549 }
550 }
551 assert.GreaterOrEqual(t, runUpdates, 2)
552 assert.GreaterOrEqual(t, jobUpdates, 2)
553
554 home := c.watch(c.root, "", "home")
555 _, err := c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: lastRun.GetId()}))
556 require.NoError(t, err)
557 ev := home.next(t, watchWait)
558 require.NotNil(t, ev.GetRunUpdated(), "home gets the run: %v", ev)
559 assert.Equal(t, "proj", ev.GetRunUpdated().GetRun().GetRepo())
560 // The rest of the rerun before the sentinel: runs, no job in progress.
561 _, err = c.forge.Repos.Create(c.ctx, domain.Root(), ports.CreateRepoParams{Name: c.name("sentinel"), Visibility: domain.VisibilityPublic})
562 require.NoError(t, err)
563 for {
564 ev = home.next(t, watchWait)
565 if ev.GetRepoCreated() != nil {
566 assert.Equal(t, "sentinel", ev.GetRepoCreated().GetRepo().GetName())
567 break
568 }
569 if j := ev.GetJobUpdated(); j != nil {
570 assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, j.GetJob().GetStatus(), "a job in progress does not reach home")
571 continue
572 }
573 require.NotNil(t, ev.GetRunUpdated(), "only CI state reaches home: %v", ev)
574 }
575}
576
577// A new branch at a commit that already exists, and a branch rewound to an
578// ancestor, bring no new commit: GitHub still runs the push workflows (those
579// without a paths filter, as no file changed).
580func TestCIPushWithoutNewCommitsStillRuns(t *testing.T) {
581 c := newCIHarness(t, nil)
582 c.registerRunner("alpha", "ubuntu-latest")
583 c.push("main", map[string]string{".github/workflows/any.yml": workflowWith("branches: ['**']")}, "")
584 c.waitFor("main runs", func() bool { return len(c.runs()) == 1 })
585 first := strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
586
587 // A new branch at the commit main already has.
588 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "HEAD:refs/heads/same")
589 c.waitFor("the new branch runs", func() bool { return len(c.runs()) == 2 })
590 assert.Equal(t, "refs/heads/same", c.runs()[0].GetRef())
591 assert.Equal(t, first, c.runs()[0].GetSha())
592
593 // A branch moved forward, then rewound to its ancestor (a forced push).
594 c.push("main", map[string]string{"a.txt": "a\n"}, "")
595 c.waitFor("the push runs", func() bool { return len(c.runs()) == 3 })
596 c.mustGit(c.wt, "push", "-q", "-f", c.urlAuth(adminToken, "/proj.git"), first+":refs/heads/main")
597 c.waitFor("the rewind runs", func() bool { return len(c.runs()) == 4 })
598 assert.Equal(t, first, c.runs()[0].GetSha())
599}
600
601// CI is on once a runner is registered: before that a push makes no run and no
602// check; after, it does.
603func TestCINoRunnerNoRunsNoChecks(t *testing.T) {
604 c := newCIHarness(t, nil)
605 sha := c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
606 time.Sleep(400 * time.Millisecond)
607 assert.Empty(t, c.runs())
608 assert.Empty(t, c.checks(sha))
609
610 c.registerRunner("alpha", "ubuntu-latest")
611 c.push("main", map[string]string{"a.txt": "a\n"}, "")
612 c.waitFor("a run once a runner exists", func() bool { return len(c.runs()) == 1 })
613}
614
615// A job that fail-fast cancelled while its assignment was not acknowledged is
616// over when the assignment is cancelled: it is not given to a runner again.
617func TestCICancelledUnacknowledgedAssignmentIsNotAssignedAgain(t *testing.T) {
618 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
619 cfg.AckTimeout, cfg.Tick, cfg.PollTimeout = 300*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
620 })
621 runner := c.registerRunner("alpha", "ubuntu-latest")
622 c.push("main", map[string]string{".github/workflows/m.yml": `on: push
623jobs:
624 build:
625 runs-on: ubuntu-latest
626 strategy:
627 matrix:
628 n: [1, 2]
629 steps:
630 - run: echo ${{ matrix.n }}
631`}, "")
632 a, b := runner.await(), runner.await() // both assigned, neither acknowledged yet
633 assert.NotEqual(t, a.GetJobId(), b.GetJobId())
634
635 runner.runJob(a, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "x") // fail-fast asks b to stop
636
637 c.waitFor("the run ends", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
638 for _, j := range c.getRun(c.runs()[0].GetId()).GetJobs() {
639 if j.GetId() == b.GetJobId() {
640 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, j.GetConclusion())
641 }
642 }
643 again, err := runner.fetch("")
644 require.NoError(t, err)
645 assert.Nil(t, again, "a cancelled job is not given again")
646}
647
648// Once the runner acknowledged a job, a retried poll with the same poll_id has
649// nothing to repeat: the plaintext token is no longer kept.
650func TestCIAcknowledgedAssignmentIsNotRepeated(t *testing.T) {
651 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PollTimeout = 150 * time.Millisecond })
652 runner := c.registerRunner("alpha", "ubuntu-latest")
653 c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
654 var job *pb.Job
655 c.waitFor("a job", func() bool {
656 j, err := runner.fetch("poll-9")
657 require.NoError(t, err)
658 job = j
659 return j != nil
660 })
661 _, err := runner.jobAPI(job).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
662 require.NoError(t, err)
663 again, err := runner.fetch("poll-9")
664 require.NoError(t, err)
665 assert.Nil(t, again, "the acknowledged assignment is not handed out again")
666}
667
668// The push payload follows GitHub's (checked against the recordings of
669// ermite-conformance): 12-character ids in "compare", the commit page for a
670// created ref and a compare range for an update, the default branch of the
671// repository, a tree id, and string counters.
672func TestCIGithubContextOfAPush(t *testing.T) {
673 c := newCIHarness(t, nil)
674 runner := c.registerRunner("alpha", "ubuntu-latest")
675 oneJobWF := strings.Split(ciWorkflow, " test:")[0]
676 first := c.push("main", map[string]string{".github/workflows/ci.yml": oneJobWF}, "")
677 ev := func() (map[string]any, map[string]any) {
678 job := runner.await()
679 var ctxs map[string]map[string]any
680 require.NoError(t, json.Unmarshal([]byte(job.GetContexts()), &ctxs))
681 gh := ctxs["github"]
682 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "x")
683 return gh, gh["event"].(map[string]any)
684 }
685
686 gh, event := ev()
687 assert.Equal(t, publicOrigin+"/owner/proj/commit/"+first[:12], event["compare"], "a created ref compares as its commit")
688 assert.Equal(t, true, event["created"])
689 assert.Equal(t, domain.ZeroOID, event["before"])
690 repo := event["repository"].(map[string]any)
691 assert.Equal(t, "main", repo["default_branch"])
692 assert.Equal(t, "main", repo["master_branch"])
693 assert.Equal(t, "90", gh["retention_days"], "a string, from the configured retention")
694 head := event["head_commit"].(map[string]any)
695 tree := strings.TrimSpace(c.bareGit("proj", "rev-parse", first+"^{tree}"))
696 assert.Equal(t, tree, head["tree_id"])
697
698 second := c.push("main", map[string]string{"a.txt": "a\n"}, "")
699 _, event = ev()
700 assert.Equal(t, publicOrigin+"/owner/proj/compare/"+first[:12]+"..."+second[:12], event["compare"])
701 assert.Equal(t, false, event["created"])
702}
703
704// ListRunners says by the server's clock whether a runner is online, and shows
705// the version the runner reported at its last poll, not the one it registered
706// with. A poll without a version keeps the stored one.
707func TestCIRunnerListOnlineAndVersion(t *testing.T) {
708 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PollTimeout = 50 * time.Millisecond })
709 r := c.registerRunner("alpha", "ubuntu-latest")
710 list := func() *pb.Runner {
711 res, err := c.root.run.ListRunners(c.ctx, connect.NewRequest(&pb.ListRunnersRequest{}))
712 require.NoError(t, err)
713 require.Len(t, res.Msg.GetRunners(), 1)
714 return res.Msg.GetRunners()[0]
715 }
716 poll := func(version string) error {
717 _, err := r.api.runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{Version: version}))
718 return err
719 }
720 got := list()
721 assert.False(t, got.GetOnline(), "never seen")
722 assert.Equal(t, "devel", got.GetVersion())
723
724 require.NoError(t, poll("v1.2.3"))
725 got = list()
726 assert.True(t, got.GetOnline())
727 assert.Equal(t, "v1.2.3", got.GetVersion(), "the running version replaces the registered one")
728
729 require.NoError(t, poll(""))
730 assert.Equal(t, "v1.2.3", list().GetVersion(), "an older runner sends none: unchanged")
731
732 assert.Equal(t, connect.CodeInvalidArgument, ciErrCode(poll(strings.Repeat("v", 65))))
733 assert.Equal(t, "v1.2.3", list().GetVersion())
734}