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