1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase_test
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "testing"
10 "time"
11
12 "github.com/bnema/zerowrap"
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/mock"
15 "github.com/stretchr/testify/require"
16
17 "bnema.dev/ermite/internal/domain"
18 "bnema.dev/ermite/internal/ports"
19 "bnema.dev/ermite/internal/ports/mocks"
20 "bnema.dev/ermite/internal/usecase"
21)
22
23// ciUnit is the CI use cases over mocks, with a clock the test moves.
24type ciUnit struct {
25 store *mocks.CIStore
26 logs *mocks.CILogStore
27 meta *mocks.RepoMetaStore
28 bus *mocks.EventBus
29 clock *fakeClock
30 ci *usecase.CI
31}
32
33type fakeClock struct{ t time.Time }
34
35func (c *fakeClock) Now() time.Time { return c.t }
36
37func newCIUnit(t *testing.T, tweak func(*usecase.CIConfig)) *ciUnit {
38 t.Helper()
39 u := &ciUnit{
40 store: mocks.NewCIStore(t), logs: mocks.NewCILogStore(t), meta: mocks.NewRepoMetaStore(t),
41 bus: mocks.NewEventBus(t), clock: &fakeClock{t: now},
42 }
43 cfg := usecase.CIConfig{
44 Store: u.store, Logs: u.logs, Meta: u.meta, Repos: mocks.NewGitRepos(t), Reader: mocks.NewRepoReader(t),
45 Git: mocks.NewMergeGit(t), Diff: mocks.NewCIGit(t), Bus: u.bus, Clock: u.clock, Owner: "owner", PublicURL: "https://forge.test",
46 }
47 if tweak != nil {
48 tweak(&cfg)
49 }
50 u.ci = usecase.NewCI(context.Background(), cfg)
51 t.Cleanup(func() { u.ci.Close(context.Background()) })
52 return u
53}
54
55const (
56 jobSecret = "ermitejob_abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMN-_0"
57 jobIDOne = "7-0-0"
58)
59
60func ciJob(t *testing.T, phase domain.CIJobPhase, secret string) domain.CIJob {
61 t.Helper()
62 return domain.CIJob{
63 ID: jobIDOne, RunID: 7, Attempt: 1, RunAttempt: 1, Repo: name(t, "proj"), Key: "build", Name: "build",
64 Phase: phase, TokenHash: domain.HashTokenSecret(secret), RunnerID: "r1", RunnerName: "alpha", Planned: true,
65 }
66}
67
68// expectToken makes the token lookup find j.
69func (u *ciUnit) expectToken(secret string, j domain.CIJob) {
70 u.store.EXPECT().FindJobByToken(mock.Anything, domain.HashTokenSecret(secret), mock.Anything).Return(j, nil)
71}
72
73// S4: a call whose token was replaced between the lookup and the lock (the
74// assignment was cancelled and given again) is unauthenticated, for every
75// call that writes the job.
76func TestStaleAssignmentIsUnauthenticatedUnderTheLock(t *testing.T) {
77 t.Parallel()
78 replaced := func(j domain.CIJob) domain.CIJob {
79 j.TokenHash = domain.HashTokenSecret("ermitejob_replaced")
80 return j
81 }
82 calls := map[string]func(u *ciUnit) error{
83 "UpdateJob": func(u *ciUnit) error {
84 return u.ci.UpdateJob(context.Background(), jobSecret, jobIDOne, []ports.StepUpdate{{
85 Number: 1, State: domain.CIStepInProgress, StartedAt: now,
86 }})
87 },
88 "AppendLog": func(u *ciUnit) error {
89 _, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 0, []domain.CILogLine{{Time: now, Text: "x"}})
90 return err
91 },
92 }
93 for label, call := range calls {
94 t.Run(label, func(t *testing.T) {
95 t.Parallel()
96 u := newCIUnit(t, nil)
97 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
98 u.expectToken(jobSecret, j)
99 u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(replaced(j), nil)
100 // No write: the stores and the log expect nothing else.
101 assert.ErrorIs(t, call(u), domain.ErrUnauthenticated)
102 })
103 }
104
105 t.Run("CompleteJob", func(t *testing.T) {
106 t.Parallel()
107 u := newCIUnit(t, nil)
108 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
109 u.expectToken(jobSecret, j)
110 u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo}, nil)
111 u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
112 u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return([]domain.CIJob{replaced(j)}, nil)
113 _, err := u.ci.CompleteJob(context.Background(), jobSecret, ports.CompleteJobParams{JobID: j.ID, Conclusion: domain.CISuccess})
114 assert.ErrorIs(t, err, domain.ErrUnauthenticated)
115 })
116
117 t.Run("an ended job answers with its end, not unauthenticated", func(t *testing.T) {
118 t.Parallel()
119 u := newCIUnit(t, nil)
120 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
121 ended := replaced(j)
122 ended.Phase, ended.Conclusion = domain.CIPhaseEnded, domain.CICancelled
123 u.expectToken(jobSecret, j)
124 u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(ended, nil)
125 err := u.ci.UpdateJob(context.Background(), jobSecret, jobIDOne, []ports.StepUpdate{{Number: 1, State: domain.CIStepInProgress, StartedAt: now}})
126 got, ok := domain.IsJobEnded(err)
127 require.True(t, ok, "%v", err)
128 assert.Equal(t, domain.CICancelled, got.Conclusion)
129 })
130}
131
132// S5: the job vanished while the log limit was handled: a plain not found, never
133// a typed nil error.
134func TestLogLimitOnAVanishedJobIsNotFound(t *testing.T) {
135 t.Parallel()
136 u := newCIUnit(t, nil)
137 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
138 u.expectToken(jobSecret, j)
139 // UpdateJob-style lookup under lock, then a log at the limit.
140 u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
141 u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(0, nil)
142 u.logs.EXPECT().Append(mock.Anything, mock.Anything, int64(0), mock.Anything, int64(domain.CIMaxLogBytes)).Return(0, 0, ports.ErrCILogLimit)
143 // stopForLogLimit loads the run, and the job is no longer in it.
144 u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo}, nil)
145 u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
146 u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return(nil, nil)
147
148 _, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 0, []domain.CILogLine{{Time: now, Text: "x"}})
149 require.Error(t, err)
150 assert.ErrorIs(t, err, domain.ErrNotFound)
151 _, isEnded := domain.IsJobEnded(err)
152 assert.False(t, isEnded)
153}
154
155// expectRunJob makes GetJobLog find the job of the repository (a public one).
156func (u *ciUnit) expectLogJob(j domain.CIJob) {
157 u.meta.EXPECT().Get(mock.Anything, j.Repo).Return(domain.Repo{Name: j.Repo, Visibility: domain.VisibilityPublic}, nil)
158 u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1}, nil)
159 u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
160}
161
162// S7: without follow a missing log is not found and an offset past the end is
163// out of range; with follow a running job's reader waits for the lines, and a
164// job that ended answers as without follow (follow changes nothing then).
165func TestGetJobLogFollowAndOutOfRange(t *testing.T) {
166 t.Parallel()
167 p := domain.Anonymous()
168 params := func(follow bool, offset int64) ports.GetJobLogParams {
169 return ports.GetJobLogParams{JobID: jobIDOne, Offset: offset, Follow: follow}
170 }
171 send := func(ports.LogBatch) error { return nil }
172
173 t.Run("no follow: out of range", func(t *testing.T) {
174 t.Parallel()
175 u := newCIUnit(t, nil)
176 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
177 u.expectLogJob(j)
178 u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{}))
179 u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(5), mock.Anything, mock.Anything).Return(nil, domain.ErrOutOfRange)
180 err := u.ci.GetJobLog(context.Background(), p, j.Repo, params(false, 5), send)
181 assert.ErrorIs(t, err, domain.ErrOutOfRange)
182 })
183 t.Run("no follow: no log", func(t *testing.T) {
184 t.Parallel()
185 u := newCIUnit(t, nil)
186 j := ciJob(t, domain.CIPhaseReady, jobSecret)
187 u.expectLogJob(j)
188 u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{}))
189 u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(0), mock.Anything, mock.Anything).Return(nil, domain.ErrNotFound)
190 assert.ErrorIs(t, u.ci.GetJobLog(context.Background(), p, j.Repo, params(false, 0), send), domain.ErrNotFound)
191 })
192 t.Run("follow, ended: out of range like without follow", func(t *testing.T) {
193 t.Parallel()
194 u := newCIUnit(t, nil)
195 j := ciJob(t, domain.CIPhaseEnded, jobSecret)
196 u.expectLogJob(j)
197 u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{}))
198 u.logs.EXPECT().Read(mock.Anything, mock.Anything, int64(5), mock.Anything, mock.Anything).Return(nil, domain.ErrOutOfRange)
199 err := u.ci.GetJobLog(context.Background(), p, j.Repo, params(true, 5), send)
200 assert.ErrorIs(t, err, domain.ErrOutOfRange)
201 })
202 t.Run("follow, running: waits for the line, then sends it", func(t *testing.T) {
203 t.Parallel()
204 u := newCIUnit(t, nil)
205 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
206 u.expectLogJob(j)
207 wake := make(chan struct{})
208 u.logs.EXPECT().Changed(mock.Anything).Return((<-chan struct{})(wake))
209 reads := 0
210 u.logs.EXPECT().Read(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
211 func(_ context.Context, _ ports.CILogRef, from int64, _, _ int) ([]domain.CILogLine, error) {
212 reads++
213 if from == 0 && reads == 1 {
214 return nil, domain.ErrOutOfRange // the runner has not written yet
215 }
216 if from == 0 {
217 return []domain.CILogLine{{Time: now, Text: "late line"}}, nil
218 }
219 return nil, nil
220 })
221 ctx, cancel := context.WithCancel(context.Background())
222 got := make(chan ports.LogBatch, 1)
223 errc := make(chan error, 1)
224 go func() {
225 errc <- u.ci.GetJobLog(ctx, p, j.Repo, params(true, 0), func(b ports.LogBatch) error {
226 got <- b
227 cancel() // enough
228 return nil
229 })
230 }()
231 select {
232 case <-got:
233 t.Fatal("a line was sent before it existed")
234 case <-time.After(150 * time.Millisecond):
235 }
236 close(wake) // the append happens
237 select {
238 case b := <-got:
239 require.Len(t, b.Lines, 1)
240 assert.Equal(t, "late line", b.Lines[0].Text)
241 case <-time.After(5 * time.Second):
242 t.Fatal("the reader did not wake up")
243 }
244 require.NoError(t, <-errc)
245 })
246}
247
248// S8: a stream that never idles revalidates the token too.
249func TestGetJobLogRevalidatesABusyStream(t *testing.T) {
250 t.Parallel()
251 auth := mocks.NewRevalidator(t)
252 u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.Auth = auth })
253 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
254 u.expectLogJob(j)
255 u.logs.EXPECT().Changed(mock.Anything).Return(make(chan struct{}))
256 // A token principal; the log keeps having lines to send.
257 tok := domain.TokenPrincipal(domain.Token{ID: "t1", Scopes: []domain.Scope{domain.ScopeRepoRead}})
258 u.logs.EXPECT().Read(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]domain.CILogLine{{Time: now, Text: "x"}}, nil)
259 revoked := domain.ErrUnauthenticated
260 auth.EXPECT().Revalidate(mock.Anything, tok).Return(revoked).Once()
261
262 sent := 0
263 start := time.Now()
264 err := u.ci.GetJobLog(context.Background(), tok, j.Repo, ports.GetJobLogParams{JobID: jobIDOne, Follow: true}, func(ports.LogBatch) error {
265 sent++
266 time.Sleep(usecase.DefaultRevalidateEvery / 2)
267 if time.Since(start) > 30*time.Second {
268 return context.DeadlineExceeded
269 }
270 return nil
271 })
272 assert.ErrorIs(t, err, domain.ErrUnauthenticated, "a revoked token stops a busy stream")
273 assert.GreaterOrEqual(t, sent, 2)
274}
275
276// S9: more than 1,000 checks on a commit is an error for the merge rule, not a
277// list cut at 1,000.
278func TestCommitChecksRefusesATruncatedList(t *testing.T) {
279 t.Parallel()
280 u := newCIUnit(t, nil)
281 repo := name(t, "proj")
282 u.store.EXPECT().ListChecks(mock.Anything, repo, "abc", usecase.MaxCIChecks).Return(make([]domain.CICheck, usecase.MaxCIChecks), true, nil).Once()
283 _, err := u.ci.CommitChecks(context.Background(), repo, "abc")
284 assert.ErrorIs(t, err, domain.ErrResourceExhausted)
285
286 u.store.EXPECT().ListChecks(mock.Anything, repo, "def", usecase.MaxCIChecks).Return([]domain.CICheck{{Name: "a"}}, false, nil).Once()
287 checks, err := u.ci.CommitChecks(context.Background(), repo, "def")
288 require.NoError(t, err)
289 assert.Len(t, checks, 1)
290}
291
292// logCapture is a context whose logger writes into a buffer the test reads.
293func logCapture(t *testing.T) (context.Context, *bytes.Buffer) {
294 t.Helper()
295 var buf bytes.Buffer
296 log := zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &buf})
297 return zerowrap.WithCtx(context.Background(), log), &buf
298}
299
300// M5, M6: the sweep removes the runs older than the retention (with the clock
301// moved, not waited for), the logs of runs no record has any more, and says so
302// when it cannot.
303func TestSweep(t *testing.T) {
304 t.Parallel()
305 retention := 30 * 24 * time.Hour
306
307 t.Run("old runs go with their logs, orphan logs too", func(t *testing.T) {
308 t.Parallel()
309 u := newCIUnit(t, func(cfg *usecase.CIConfig) { cfg.Retention = retention })
310 u.clock.t = now.Add(100 * 24 * time.Hour)
311 ctx, out := logCapture(t)
312 u.store.EXPECT().Sweep(mock.Anything, u.clock.t.Add(-retention)).Return([]int64{3, 4}, nil)
313 u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{3, 4}).Return(nil).Once()
314 u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{5, 6, 7}, nil)
315 u.store.EXPECT().GetRunSource(mock.Anything, int64(5)).Return(domain.CIRunSource{}, nil)
316 u.store.EXPECT().GetRunSource(mock.Anything, int64(6)).Return(domain.CIRunSource{}, domain.ErrNotFound)
317 u.store.EXPECT().GetRunSource(mock.Anything, int64(7)).Return(domain.CIRunSource{}, domain.ErrNotFound)
318 u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{6, 7}).Return(nil).Once()
319 u.ci.Sweep(ctx)
320 assert.NotContains(t, out.String(), `"level":"error"`)
321 assert.Contains(t, out.String(), "old runs removed")
322 })
323
324 t.Run("a store that fails ends the sweep, logged", func(t *testing.T) {
325 t.Parallel()
326 u := newCIUnit(t, nil)
327 ctx, out := logCapture(t)
328 u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, errors.New("kv closed"))
329 u.ci.Sweep(ctx) // the log store expects no call
330 assert.Contains(t, out.String(), "sweep the old runs")
331 assert.Contains(t, out.String(), "kv closed")
332 })
333
334 t.Run("a log directory that cannot be removed is logged and the orphans are still looked for", func(t *testing.T) {
335 t.Parallel()
336 u := newCIUnit(t, nil)
337 ctx, out := logCapture(t)
338 u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return([]int64{3}, nil)
339 u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{3}).Return(errors.New("permission denied")).Once()
340 u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{9}, nil)
341 u.store.EXPECT().GetRunSource(mock.Anything, int64(9)).Return(domain.CIRunSource{}, domain.ErrNotFound)
342 u.logs.EXPECT().RemoveRuns(mock.Anything, []int64{9}).Return(errors.New("disk on fire")).Once()
343 u.ci.Sweep(ctx)
344 assert.Contains(t, out.String(), "remove the logs of the old runs")
345 assert.Contains(t, out.String(), "permission denied")
346 assert.Contains(t, out.String(), "remove the logs of runs that no longer exist")
347 assert.Contains(t, out.String(), "disk on fire")
348 })
349
350 t.Run("a listing of the log directories that fails is logged", func(t *testing.T) {
351 t.Parallel()
352 u := newCIUnit(t, nil)
353 ctx, out := logCapture(t)
354 u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, nil)
355 u.logs.EXPECT().RunIDs(mock.Anything).Return(nil, errors.New("io error"))
356 u.ci.Sweep(ctx)
357 assert.Contains(t, out.String(), "list the log directories")
358 })
359
360 t.Run("a run with a lookup error keeps its logs", func(t *testing.T) {
361 t.Parallel()
362 u := newCIUnit(t, nil)
363 ctx, out := logCapture(t)
364 u.store.EXPECT().Sweep(mock.Anything, mock.Anything).Return(nil, nil)
365 u.logs.EXPECT().RunIDs(mock.Anything).Return([]int64{9}, nil)
366 u.store.EXPECT().GetRunSource(mock.Anything, int64(9)).Return(domain.CIRunSource{}, errors.New("kv closed"))
367 u.ci.Sweep(ctx) // RemoveRuns is not expected: an unknown state removes nothing
368 assert.Contains(t, out.String(), "look up a run")
369 })
370}
371
372// M5: a log that would pass its limit appends nothing; the server adds a last
373// line saying so, fails the job and answers failed_precondition with the end.
374func TestLogLimitEndsTheJob(t *testing.T) {
375 t.Parallel()
376 u := newCIUnit(t, nil)
377 j := ciJob(t, domain.CIPhaseRunning, jobSecret)
378 j.Steps = []domain.CIStep{{Number: 1, State: domain.CIStepInProgress, StartedAt: now, HasLines: true}}
379 u.expectToken(jobSecret, j)
380 u.store.EXPECT().GetJob(mock.Anything, j.RunID, j.RunAttempt, j.ID).Return(j, nil)
381 u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(10, nil)
382 u.logs.EXPECT().Append(mock.Anything, mock.Anything, int64(10), mock.Anything, int64(domain.CIMaxLogBytes)).Return(0, 0, ports.ErrCILogLimit)
383
384 // stopForLogLimit: the run is loaded, the last line is forced in, the job ends.
385 u.store.EXPECT().GetRunSource(mock.Anything, j.RunID).Return(domain.CIRunSource{RunID: j.RunID, Repo: j.Repo, WorkflowYAML: "on: push\njobs:\n build:\n runs-on: x\n steps:\n - run: echo\n"}, nil)
386 u.store.EXPECT().GetRun(mock.Anything, j.Repo, j.RunID).Return(domain.CIRun{Repo: j.Repo, ID: j.RunID, Attempt: 1, WorkflowPath: "w.yml"}, nil)
387 u.store.EXPECT().ListJobs(mock.Anything, j.RunID, int32(1)).Return([]domain.CIJob{j}, nil)
388 var forced []domain.CILogLine
389 u.logs.EXPECT().AppendForced(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
390 func(_ context.Context, _ ports.CILogRef, lines []domain.CILogLine) (int64, error) {
391 forced = lines
392 return 11, nil
393 })
394 var written ports.CIWrite
395 u.store.EXPECT().Apply(mock.Anything, mock.Anything).RunAndReturn(
396 func(_ context.Context, w ports.CIWrite) (uint64, error) { written = w; return 5, nil })
397 u.meta.EXPECT().Get(mock.Anything, j.Repo).Return(domain.Repo{Name: j.Repo, Visibility: domain.VisibilityPrivate}, nil)
398 u.logs.EXPECT().Count(mock.Anything, mock.Anything).Return(11, nil).Maybe()
399 u.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return().Maybe()
400
401 _, err := u.ci.AppendLog(context.Background(), jobSecret, jobIDOne, 1, 10, []domain.CILogLine{{Time: now, Text: "one more"}})
402 ended, ok := domain.IsJobEnded(err)
403 require.True(t, ok, "%v", err)
404 assert.Equal(t, domain.CIFailure, ended.Conclusion)
405 assert.Contains(t, ended.Message, "512 MiB")
406
407 require.Len(t, forced, 1)
408 assert.Contains(t, forced[0].Text, "##[error]")
409 assert.Contains(t, forced[0].Text, "512 MiB")
410 require.Len(t, written.Jobs, 1)
411 assert.Equal(t, domain.CIPhaseEnded, written.Jobs[0].Phase)
412 assert.Equal(t, domain.CIFailure, written.Jobs[0].Conclusion)
413 assert.Equal(t, domain.CIStepCompleted, written.Jobs[0].Steps[0].State, "an open step is completed with the job")
414}