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