1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package app
4
5import (
6 "errors"
7 "fmt"
8 "strings"
9 "sync"
10 "testing"
11 "time"
12
13 "connectrpc.com/connect"
14 "github.com/stretchr/testify/assert"
15 "github.com/stretchr/testify/require"
16 "google.golang.org/protobuf/types/known/fieldmaskpb"
17 "google.golang.org/protobuf/types/known/timestamppb"
18
19 pb "bnema.dev/ermite-api/gen/ermite"
20 "bnema.dev/ermite/internal/domain"
21 "bnema.dev/ermite/internal/ports"
22 "bnema.dev/ermite/internal/usecase"
23)
24
25// ciHarness is a server with a repository whose pushes run workflows, and a
26// fake runner that talks to the real Connect handlers.
27type ciHarness struct {
28 *harness
29 root *clients
30 wt string
31 repo string
32}
33
34const ciWorkflow = `name: ci
35on:
36 push:
37 branches: [main, feat]
38jobs:
39 build:
40 runs-on: ubuntu-latest
41 steps:
42 - run: echo hello
43 test:
44 needs: build
45 runs-on: ubuntu-latest
46 steps:
47 - run: echo test
48`
49
50func newCIHarness(t *testing.T, tweak func(*usecase.CIConfig)) *ciHarness {
51 t.Helper()
52 h := newHarnessWith(t, harnessOpts{}, func(o *options) { o.ci = tweak })
53 c := &ciHarness{harness: h, root: h.api(adminToken), repo: "proj"}
54 _, err := h.forge.Repos.Create(h.ctx, domain.Root(), ports.CreateRepoParams{Name: h.name("proj"), Visibility: domain.VisibilityPrivate})
55 require.NoError(t, err)
56 c.wt = h.workTree("main")
57 return c
58}
59
60// push commits the files to branch and pushes it.
61func (c *ciHarness) push(branch string, files map[string]string, msg string) string {
62 c.t.Helper()
63 if cur, err := c.git(c.wt, "rev-parse", "--abbrev-ref", "HEAD"); err != nil || strings.TrimSpace(cur) != branch {
64 if _, err := c.git(c.wt, "checkout", "-q", branch); err != nil {
65 c.mustGit(c.wt, "checkout", "-q", "-b", branch)
66 }
67 }
68 for name, content := range files {
69 c.writeFile(c.wt, name, []byte(content))
70 }
71 if msg != "" {
72 c.mustGit(c.wt, "commit", "-q", "--amend", "-m", msg)
73 }
74 c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/"+c.repo+".git"), branch)
75 return strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
76}
77
78// registerRunner registers a runner and returns a fake that polls with it.
79func (c *ciHarness) registerRunner(name string, labels ...string) *fakeRunner {
80 c.t.Helper()
81 res, err := c.root.runner.RegisterRunner(c.ctx, connect.NewRequest(&pb.RegisterRunnerRequest{
82 Name: name, Labels: labels, Version: "devel", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux",
83 }))
84 require.NoError(c.t, err)
85 return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), secret: res.Msg.GetRunnerToken(), api: c.api(res.Msg.GetRunnerToken())}
86}
87
88// fakeRunner is the runner side of the protocol, as a test.
89type fakeRunner struct {
90 c *ciHarness
91 id string
92 secret string
93 api *clients
94}
95
96// runnerSecret is the runner token of a fake runner.
97func (c *ciHarness) runnerSecret(r *fakeRunner) string { return r.secret }
98
99// fetch polls once for a job.
100func (r *fakeRunner) fetch(pollID string) (*pb.Job, error) {
101 res, err := r.api.runner.FetchJob(r.c.ctx, connect.NewRequest(&pb.FetchJobRequest{PollId: pollID}))
102 if err != nil {
103 return nil, err
104 }
105 return res.Msg.GetJob(), nil
106}
107
108// await polls until a job is given.
109func (r *fakeRunner) await() *pb.Job {
110 r.c.t.Helper()
111 deadline := time.Now().Add(15 * time.Second)
112 for time.Now().Before(deadline) {
113 j, err := r.fetch("")
114 require.NoError(r.c.t, err)
115 if j != nil {
116 return j
117 }
118 }
119 r.c.t.Fatal("no job was given to the runner")
120 return nil
121}
122
123// jobAPI is the client of a job: its token is the bearer secret.
124func (r *fakeRunner) jobAPI(j *pb.Job) *clients { return r.c.api(j.GetJobToken()) }
125
126func now() *timestamppb.Timestamp { return timestamppb.Now() }
127
128// runJob executes a job as a runner does: heartbeat, steps, log, completion.
129func (r *fakeRunner) runJob(j *pb.Job, conclusion pb.CiConclusion, outputs map[string]string, text ...string) {
130 r.c.t.Helper()
131 ctx, api := r.c.ctx, r.jobAPI(j)
132 _, err := api.runner.Heartbeat(ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j.GetJobId()}))
133 require.NoError(r.c.t, err)
134 start := now()
135 _, err = api.runner.UpdateJob(ctx, connect.NewRequest(&pb.UpdateJobRequest{JobId: j.GetJobId(), Steps: []*pb.StepUpdate{
136 {Number: 1, Name: "Set up job", State: pb.StepState_STEP_STATE_IN_PROGRESS, StartedAt: start},
137 }}))
138 require.NoError(r.c.t, err)
139 var lines []*pb.LogLine
140 for _, t := range text {
141 lines = append(lines, &pb.LogLine{Time: now(), Text: t})
142 }
143 if len(lines) > 0 {
144 _, err = api.runner.AppendLog(ctx, connect.NewRequest(&pb.AppendLogRequest{JobId: j.GetJobId(), Step: 2, Offset: 0, Lines: lines}))
145 require.NoError(r.c.t, err)
146 }
147 step := pb.CiConclusion_CI_CONCLUSION_SUCCESS
148 if conclusion != pb.CiConclusion_CI_CONCLUSION_SUCCESS {
149 step = conclusion
150 }
151 res, err := api.runner.CompleteJob(ctx, connect.NewRequest(&pb.CompleteJobRequest{
152 JobId: j.GetJobId(), Conclusion: conclusion, Outputs: outputs,
153 Steps: []*pb.StepUpdate{
154 {Number: 1, Name: "Set up job", State: pb.StepState_STEP_STATE_COMPLETED, Conclusion: pb.CiConclusion_CI_CONCLUSION_SUCCESS, StartedAt: start, CompletedAt: now()},
155 {Number: 2, Name: "Run echo", State: pb.StepState_STEP_STATE_COMPLETED, Conclusion: step, StartedAt: start, CompletedAt: now()},
156 },
157 }))
158 require.NoError(r.c.t, err)
159 assert.True(r.c.t, res.Msg.GetApplied())
160}
161
162// waitFor polls cond until it holds.
163func (c *ciHarness) waitFor(what string, cond func() bool) {
164 c.t.Helper()
165 require.Eventually(c.t, cond, 15*time.Second, 20*time.Millisecond, what)
166}
167
168func (c *ciHarness) runs() []*pb.Run {
169 c.t.Helper()
170 res, err := c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: c.repo}))
171 require.NoError(c.t, err)
172 return res.Msg.GetRuns()
173}
174
175func (c *ciHarness) checks(sha string) []*pb.Check {
176 c.t.Helper()
177 res, err := c.root.run.ListChecks(c.ctx, connect.NewRequest(&pb.ListChecksRequest{Repo: c.repo, Sha: sha}))
178 require.NoError(c.t, err)
179 return res.Msg.GetChecks()
180}
181
182func (c *ciHarness) getRun(id int64) *pb.GetRunResponse {
183 c.t.Helper()
184 res, err := c.root.run.GetRun(c.ctx, connect.NewRequest(&pb.GetRunRequest{Repo: c.repo, RunId: id}))
185 require.NoError(c.t, err)
186 return res.Msg
187}
188
189func ciErrCode(err error) connect.Code {
190 var ce *connect.Error
191 if errors.As(err, &ce) {
192 return ce.Code()
193 }
194 return connect.Code(0)
195}
196
197// AC3, AC5: a push runs the workflow on a runner; its jobs follow their needs;
198// the checks appear on the commit; the log is readable.
199func TestCIPushRunsAWorkflowOnARunner(t *testing.T) {
200 c := newCIHarness(t, nil)
201 runner := c.registerRunner("alpha", "ubuntu-latest")
202 sha := c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
203
204 c.waitFor("a run is created", func() bool { return len(c.runs()) == 1 })
205 run := c.runs()[0]
206 assert.Equal(t, "ci", run.GetWorkflowName())
207 assert.Equal(t, "refs/heads/main", run.GetRef())
208 assert.Equal(t, sha, run.GetSha())
209 assert.Equal(t, "push", run.GetEvent())
210 assert.EqualValues(t, 1, run.GetNumber())
211 assert.Equal(t, "root", run.GetActor())
212
213 // Only the first job is ready: the second needs it.
214 job := runner.await()
215 assert.Equal(t, "build", job.GetJobKey())
216 assert.Equal(t, run.GetId(), job.GetRunId())
217 assert.Equal(t, sha, job.GetSha())
218 assert.Equal(t, "refs/heads/main", job.GetRef())
219 assert.Equal(t, "proj", job.GetRepo())
220 assert.Equal(t, ciWorkflow, job.GetWorkflowYaml())
221 assert.Equal(t, []string{"ubuntu-latest"}, job.GetLabels())
222 assert.EqualValues(t, 360, job.GetTimeoutMinutes())
223 assert.Equal(t, "{}", job.GetMatrix())
224 assert.True(t, strings.HasPrefix(job.GetJobToken(), "ermitejob_"))
225 assert.Contains(t, job.GetContexts(), `"repository":"owner/proj"`)
226 assert.Contains(t, job.GetContexts(), `"needs":{}`)
227
228 // Assigned but not acknowledged: still queued for the owner.
229 assert.Equal(t, pb.CiStatus_CI_STATUS_QUEUED, c.getRun(run.GetId()).GetJobs()[0].GetStatus())
230
231 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, map[string]string{"artifact": "out.bin"}, "hello", "world")
232
233 next := runner.await()
234 assert.Equal(t, "test", next.GetJobKey())
235 assert.Contains(t, next.GetContexts(), `"needs":{"build":{"result":"success","outputs":{"artifact":"out.bin"}}}`)
236 runner.runJob(next, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "tested")
237
238 c.waitFor("the run completes", func() bool {
239 return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
240 })
241 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, c.runs()[0].GetConclusion())
242
243 checks := c.checks(sha)
244 require.Len(t, checks, 2)
245 assert.Equal(t, "build", checks[0].GetName())
246 assert.Equal(t, "test", checks[1].GetName())
247 for _, ck := range checks {
248 assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, ck.GetStatus())
249 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, ck.GetConclusion())
250 }
251
252 detail := c.getRun(run.GetId())
253 require.Len(t, detail.GetJobs(), 2)
254 steps := detail.GetJobs()[0].GetSteps()
255 require.Len(t, steps, 2)
256 assert.EqualValues(t, 0, steps[1].GetLogStart(), "step 2 has the lines")
257 assert.EqualValues(t, 2, steps[1].GetLogEnd())
258
259 // The log is streamed back with its steps.
260 stream, err := c.root.run.GetJobLog(c.ctx, connect.NewRequest(&pb.GetJobLogRequest{Repo: "proj", JobId: job.GetJobId()}))
261 require.NoError(t, err)
262 var got []string
263 for stream.Receive() {
264 for _, l := range stream.Msg().GetLines() {
265 got = append(got, l.GetText())
266 assert.EqualValues(t, 2, l.GetStep())
267 }
268 }
269 require.NoError(t, stream.Err())
270 assert.Equal(t, []string{"hello", "world"}, got)
271}
272
273// AC5: merge refused while checks are pending or failing, allowed when they
274// pass, and when failing_checks_block is off.
275func TestCIMergeIsBlockedByChecks(t *testing.T) {
276 c := newCIHarness(t, nil)
277 runner := c.registerRunner("alpha", "ubuntu-latest")
278 c.push("main", map[string]string{"README": "x\n"}, "")
279
280 feat := c.push("feat", map[string]string{".github/workflows/ci.yml": strings.ReplaceAll(ciWorkflow, " test:\n needs: build\n runs-on: ubuntu-latest\n steps:\n - run: echo test\n", "")}, "")
281 mr, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "Add CI", SourceBranch: "feat"}))
282 require.NoError(t, err)
283 merge := func() error {
284 _, err := c.root.mr.MergeMergeRequest(c.ctx, connect.NewRequest(&pb.MergeMergeRequestRequest{Repo: "proj", Number: mr.Msg.GetMergeRequest().GetNumber(), HeadId: feat}))
285 return err
286 }
287
288 c.waitFor("the feature run is queued", func() bool { return len(c.checks(feat)) == 1 })
289 err = merge()
290 require.Error(t, err)
291 assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err), "a pending check blocks the merge")
292 assert.Contains(t, err.Error(), "have not finished")
293
294 job := runner.await()
295 runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "boom")
296 c.waitFor("the check fails", func() bool {
297 cs := c.checks(feat)
298 return len(cs) == 1 && cs[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
299 })
300 err = merge()
301 require.Error(t, err)
302 assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
303 assert.Contains(t, err.Error(), "failing checks block the merge")
304
305 // Off: the merge goes through.
306 _, err = c.root.repo.UpdateRepo(c.ctx, connect.NewRequest(&pb.UpdateRepoRequest{
307 Repo: "proj",
308 Settings: &pb.RepoSettings{FailingChecksBlock: false},
309 UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"settings.failing_checks_block"}},
310 }))
311 require.NoError(t, err)
312 require.NoError(t, merge())
313}
314
315// AC2: a job token works for its job only, and not once the job has ended.
316func TestCIJobTokensAreScopedToTheirJob(t *testing.T) {
317 c := newCIHarness(t, nil)
318 runner := c.registerRunner("alpha", "ubuntu-latest")
319 c.push("main", map[string]string{".github/workflows/a.yml": ciWorkflow, ".github/workflows/b.yml": ciWorkflow}, "")
320 c.waitFor("two runs", func() bool { return len(c.runs()) == 2 })
321 j1, j2 := runner.await(), runner.await()
322 require.NotEqual(t, j1.GetJobId(), j2.GetJobId())
323
324 // Another job's id with this job's token: permission_denied.
325 _, err := runner.jobAPI(j1).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j2.GetJobId()}))
326 assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
327 // A runner token is not a job token, and a stored token is neither.
328 _, err = runner.api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j1.GetJobId()}))
329 assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
330 _, err = c.root.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: j1.GetJobId()}))
331 assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
332 // A job token does not poll.
333 _, err = runner.jobAPI(j1).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
334 assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err))
335
336 runner.runJob(j1, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "a")
337 // After the end: failed_precondition with the JobEnded detail; CompleteJob
338 // still answers with the stored result.
339 _, err = runner.jobAPI(j1).runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
340 JobId: j1.GetJobId(), Step: 2, Offset: 1, Lines: []*pb.LogLine{{Time: now(), Text: "late"}},
341 }))
342 require.Error(t, err)
343 assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
344 var ce *connect.Error
345 require.True(t, errors.As(err, &ce))
346 require.Len(t, ce.Details(), 1)
347 msg, derr := ce.Details()[0].Value()
348 require.NoError(t, derr)
349 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, msg.(*pb.JobEnded).GetConclusion())
350
351 res, err := runner.jobAPI(j1).runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
352 JobId: j1.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_FAILURE,
353 }))
354 require.NoError(t, err)
355 assert.False(t, res.Msg.GetApplied(), "a second CompleteJob changes nothing")
356 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, res.Msg.GetConclusion(), "and answers the stored conclusion")
357}
358
359// AC2: a job is given to exactly one runner, however many poll at once.
360func TestCIConcurrentFetchJobAssignsOnce(t *testing.T) {
361 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PollTimeout = 300 * time.Millisecond })
362 var runners []*fakeRunner
363 for i := range 8 {
364 runners = append(runners, c.registerRunner(fmt.Sprintf("r%d", i), "ubuntu-latest"))
365 }
366 var wf strings.Builder
367 wf.WriteString("on: push\njobs:\n")
368 for i := range 3 {
369 fmt.Fprintf(&wf, " j%d:\n runs-on: ubuntu-latest\n steps:\n - run: echo %d\n", i, i)
370 }
371 c.push("main", map[string]string{".github/workflows/many.yml": wf.String()}, "")
372 c.waitFor("the run exists", func() bool { return len(c.runs()) == 1 })
373
374 var mu sync.Mutex
375 got := map[string]int{}
376 var wg sync.WaitGroup
377 for _, r := range runners {
378 wg.Add(1)
379 go func() {
380 defer wg.Done()
381 for range 20 {
382 j, err := r.fetch("")
383 if err != nil {
384 return
385 }
386 if j != nil {
387 mu.Lock()
388 got[j.GetJobId()]++
389 mu.Unlock()
390 }
391 }
392 }()
393 }
394 wg.Wait()
395 assert.Len(t, got, 3, "every job was given")
396 for id, n := range got {
397 assert.Equal(t, 1, n, "job %s was given %d times", id, n)
398 }
399}
400
401// AC2: the first heartbeat acknowledges; without it the assignment is
402// cancelled and the job is queued again, three times in a row at most.
403func TestCIUnacknowledgedAssignmentGoesBackToTheQueue(t *testing.T) {
404 c := newCIHarness(t, func(cfg *usecase.CIConfig) {
405 cfg.AckTimeout, cfg.Tick, cfg.PollTimeout = 150*time.Millisecond, 20*time.Millisecond, 100*time.Millisecond
406 })
407 runner := c.registerRunner("alpha", "ubuntu-latest")
408 c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
409
410 first := runner.await()
411 // No heartbeat: the token is revoked and the job comes back with a new one.
412 second := runner.await()
413 assert.Equal(t, first.GetJobId(), second.GetJobId())
414 assert.NotEqual(t, first.GetJobToken(), second.GetJobToken())
415 _, err := runner.jobAPI(first).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: first.GetJobId()}))
416 assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(err), "the replaced token is rejected")
417
418 // The third assignment is the last: ignoring it fails the job.
419 runner.await()
420 c.waitFor("the job fails", func() bool {
421 return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
422 })
423 detail := c.getRun(c.runs()[0].GetId())
424 assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, detail.GetJobs()[0].GetConclusion())
425 assert.Contains(t, detail.GetJobs()[0].GetError(), "acknowledged")
426}
427
428// A retried poll with the same poll_id returns the same assignment.
429func TestCIRetriedPollReturnsTheSameAssignment(t *testing.T) {
430 c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PollTimeout = 200 * time.Millisecond })
431 runner := c.registerRunner("alpha", "ubuntu-latest")
432 c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
433 c.waitFor("a run", func() bool { return len(c.runs()) == 1 })
434
435 var first *pb.Job
436 c.waitFor("a job", func() bool {
437 j, err := runner.fetch("poll-1")
438 require.NoError(t, err)
439 first = j
440 return j != nil
441 })
442 again, err := runner.fetch("poll-1")
443 require.NoError(t, err)
444 require.NotNil(t, again)
445 assert.Equal(t, first.GetJobId(), again.GetJobId())
446 assert.Equal(t, first.GetJobToken(), again.GetJobToken())
447}