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