1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore_test
4
5import (
6 "context"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/assert"
12 "github.com/stretchr/testify/require"
13
14 "bnema.dev/ermite/internal/adapters/kvstore"
15 "bnema.dev/ermite/internal/domain"
16 "bnema.dev/ermite/internal/ports"
17)
18
19// ciRepo opens a store holding repository "proj".
20func ciRepo(t *testing.T) (*kvstore.Store, *kvstore.CI, domain.RepoName) {
21 t.Helper()
22 s := openStore(t, t.TempDir())
23 n := name(t, "proj")
24 require.NoError(t, s.Create(context.Background(), domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
25 return s, s.CI(), n
26}
27
28func draftOf(repo domain.RepoName, workflow string, jobs int) func(id, number int64) ports.CIRunDraft {
29 return func(id, number int64) ports.CIRunDraft {
30 d := ports.CIRunDraft{
31 Run: domain.CIRun{
32 Repo: repo, ID: id, Number: number, Attempt: 1, WorkflowPath: workflow, WorkflowName: "ci", Event: "push",
33 Ref: "refs/heads/main", Sha: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Status: domain.CIQueued, CreatedAt: created,
34 },
35 Source: domain.CIRunSource{WorkflowYAML: "on: push", GithubJSON: "{}"},
36 }
37 for i := range jobs {
38 j := domain.CIJob{
39 ID: domain.NewJobID(id, i, 0), RunID: id, Attempt: 1, RunAttempt: 1, Repo: repo, Key: "build", Name: "build",
40 Phase: domain.CIPhaseReady, ReadyAt: created.Add(time.Duration(i) * time.Second), CreatedAt: created,
41 Labels: []string{"self-hosted"}, Index: i, TimeoutMinutes: 360, Planned: true,
42 }
43 d.Jobs = append(d.Jobs, j)
44 d.Checks = append(d.Checks, domain.CICheck{
45 ID: j.ID, Repo: repo, Sha: d.Run.Sha, Name: j.Name, Status: domain.CIQueued, RunID: id, JobID: j.ID, Attempt: 1,
46 WorkflowName: "ci", CreatedAt: created,
47 })
48 }
49 return d
50 }
51}
52
53func TestCICreateRunTakesIncreasingIdsAndPerWorkflowNumbers(t *testing.T) {
54 ctx := context.Background()
55 _, ci, n := ciRepo(t)
56
57 a, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 1))
58 require.NoError(t, err)
59 b, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 1))
60 require.NoError(t, err)
61 c, err := ci.CreateRun(ctx, n, ".github/workflows/lint.yml", draftOf(n, ".github/workflows/lint.yml", 0))
62 require.NoError(t, err)
63
64 assert.Equal(t, []int64{1, 2, 3}, []int64{a.Run.ID, b.Run.ID, c.Run.ID})
65 assert.Equal(t, []int64{1, 2, 1}, []int64{a.Run.Number, b.Run.Number, c.Run.Number})
66 assert.NotZero(t, a.Run.Version)
67}
68
69func TestCIConcurrentCreateRunNeverSharesAnId(t *testing.T) {
70 ctx := context.Background()
71 _, ci, n := ciRepo(t)
72 var mu sync.Mutex
73 ids := map[int64]bool{}
74 var wg sync.WaitGroup
75 for range 12 {
76 wg.Add(1)
77 go func() {
78 defer wg.Done()
79 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
80 if assert.NoError(t, err) {
81 mu.Lock()
82 defer mu.Unlock()
83 assert.False(t, ids[d.Run.ID], "id %d handed out twice", d.Run.ID)
84 ids[d.Run.ID] = true
85 }
86 }()
87 }
88 wg.Wait()
89 assert.Len(t, ids, 12)
90}
91
92func TestCICreateRunFailsOnAMissingRepository(t *testing.T) {
93 _, ci, _ := ciRepo(t)
94 ghost := name(t, "ghost")
95 _, err := ci.CreateRun(context.Background(), ghost, "w.yml", draftOf(ghost, "w.yml", 1))
96 assert.ErrorIs(t, err, domain.ErrNotFound)
97}
98
99func TestCIJobIndexesFollowThePhase(t *testing.T) {
100 ctx := context.Background()
101 _, ci, n := ciRepo(t)
102 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 2))
103 require.NoError(t, err)
104
105 ready, err := ci.ListReadyJobs(ctx, 10)
106 require.NoError(t, err)
107 require.Len(t, ready, 2)
108 assert.Equal(t, d.Jobs[0].ID, ready[0].ID, "oldest first")
109
110 open, err := ci.ListOpenRuns(ctx)
111 require.NoError(t, err)
112 assert.Len(t, open, 1)
113
114 // Assign the first job: it leaves the queue, enters the active set and its
115 // token becomes findable.
116 j := ready[0]
117 hash := domain.HashTokenSecret("ermitejob_x")
118 j.Phase, j.TokenHash, j.AssignedAt = domain.CIPhaseAssigned, hash, created.Add(time.Minute)
119 j.TokenValidUntil = created.Add(48 * time.Hour)
120 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
121 require.NoError(t, err)
122
123 ready, _ = ci.ListReadyJobs(ctx, 10)
124 assert.Len(t, ready, 1)
125 active, err := ci.ListActiveJobs(ctx)
126 require.NoError(t, err)
127 require.Len(t, active, 1)
128 assert.Equal(t, j.ID, active[0].ID)
129 found, err := ci.FindJobByToken(ctx, hash, created.Add(2*time.Minute))
130 require.NoError(t, err)
131 assert.Equal(t, j.ID, found.ID)
132 _, err = ci.FindJobByToken(ctx, hash, created.Add(72*time.Hour))
133 assert.ErrorIs(t, err, domain.ErrNotFound, "an expired token is not found")
134
135 // A new token replaces the old one.
136 j2, err := ci.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
137 require.NoError(t, err)
138 j2.TokenHash = domain.HashTokenSecret("ermitejob_y")
139 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j2}})
140 require.NoError(t, err)
141 _, err = ci.FindJobByToken(ctx, hash, created.Add(2*time.Minute))
142 assert.ErrorIs(t, err, domain.ErrNotFound, "the replaced token is gone")
143
144 // End it.
145 j3, _ := ci.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)
146 j3.Phase, j3.Conclusion = domain.CIPhaseEnded, domain.CISuccess
147 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j3}})
148 require.NoError(t, err)
149 active, _ = ci.ListActiveJobs(ctx)
150 assert.Empty(t, active)
151}
152
153func TestCIApplyChecksVersions(t *testing.T) {
154 ctx := context.Background()
155 _, ci, n := ciRepo(t)
156 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
157 require.NoError(t, err)
158
159 a, err := ci.GetRun(ctx, n, d.Run.ID)
160 require.NoError(t, err)
161 b := a
162 a.Status = domain.CIInProgress
163 _, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{a}})
164 require.NoError(t, err)
165 b.Status = domain.CICompleted
166 _, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{b}})
167 assert.ErrorIs(t, err, domain.ErrConflict, "a write on a stale copy loses")
168}
169
170func TestCIChecksAreOrderedAndLimited(t *testing.T) {
171 ctx := context.Background()
172 _, ci, n := ciRepo(t)
173 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 3))
174 require.NoError(t, err)
175
176 checks, more, err := ci.ListChecks(ctx, n, d.Run.Sha, 10)
177 require.NoError(t, err)
178 assert.Len(t, checks, 3)
179 assert.False(t, more)
180 checks, more, err = ci.ListChecks(ctx, n, d.Run.Sha, 2)
181 require.NoError(t, err)
182 assert.Len(t, checks, 2)
183 assert.True(t, more)
184 none, _, err := ci.ListChecks(ctx, n, "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", 10)
185 require.NoError(t, err)
186 assert.Empty(t, none)
187}
188
189func TestCIListRunsNewestFirstAndBounded(t *testing.T) {
190 ctx := context.Background()
191 s, ci, n := ciRepo(t)
192 other := name(t, "other")
193 require.NoError(t, s.Create(ctx, domain.Repo{Name: other, Visibility: domain.VisibilityPublic, CreatedAt: created}))
194 _, err := ci.CreateRun(ctx, other, "w.yml", draftOf(other, "w.yml", 0))
195 require.NoError(t, err)
196 for range 5 {
197 _, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
198 require.NoError(t, err)
199 }
200 runs, err := ci.ListRuns(ctx, n, 0, 3)
201 require.NoError(t, err)
202 require.Len(t, runs, 3)
203 assert.Equal(t, []int64{6, 5, 4}, []int64{runs[0].ID, runs[1].ID, runs[2].ID})
204 runs, err = ci.ListRuns(ctx, n, 4, 10)
205 require.NoError(t, err)
206 assert.Equal(t, []int64{3, 2}, []int64{runs[0].ID, runs[1].ID})
207 runs, err = ci.ListRuns(ctx, n, 0, 0)
208 require.NoError(t, err)
209 require.Len(t, runs, 5)
210 assert.Equal(t, int64(6), runs[0].ID)
211 assert.Equal(t, int64(2), runs[4].ID)
212 ids, err := ci.RunIDs(ctx, n)
213 require.NoError(t, err)
214 assert.Len(t, ids, 5)
215}
216
217func TestCIRunnersCreateReplaceFindRemove(t *testing.T) {
218 ctx := context.Background()
219 _, ci, _ := ciRepo(t)
220 h1, h2 := domain.HashTokenSecret("ermiterun_1"), domain.HashTokenSecret("ermiterun_2")
221 r1 := domain.CIRunner{ID: "r1", Name: "Alpha", Labels: []string{"linux"}, Isolation: domain.CIIsolationVM, TokenHash: h1, RegisteredAt: created}
222 _, err := ci.CreateRunner(ctx, r1, false)
223 require.NoError(t, err)
224
225 r2 := domain.CIRunner{ID: "r2", Name: "alpha", TokenHash: h2, RegisteredAt: created}
226 _, err = ci.CreateRunner(ctx, r2, false)
227 assert.ErrorIs(t, err, domain.ErrAlreadyExists, "names are case-insensitive")
228
229 removed, err := ci.CreateRunner(ctx, r2, true)
230 require.NoError(t, err)
231 require.Len(t, removed, 1)
232 assert.Equal(t, "r1", removed[0].ID)
233 _, err = ci.FindRunnerByToken(ctx, h1)
234 assert.ErrorIs(t, err, domain.ErrNotFound, "the replaced runner's token stops working")
235 got, err := ci.FindRunnerByToken(ctx, h2)
236 require.NoError(t, err)
237 assert.Equal(t, "r2", got.ID)
238
239 list, err := ci.ListRunners(ctx)
240 require.NoError(t, err)
241 assert.Len(t, list, 1)
242
243 _, err = ci.RemoveRunner(ctx, "r2")
244 require.NoError(t, err)
245 _, err = ci.RemoveRunner(ctx, "r2")
246 assert.ErrorIs(t, err, domain.ErrNotFound)
247 _, err = ci.FindRunnerByToken(ctx, h2)
248 assert.ErrorIs(t, err, domain.ErrNotFound)
249 // The name is free again.
250 _, err = ci.CreateRunner(ctx, domain.CIRunner{ID: "r3", Name: "alpha", TokenHash: domain.HashTokenSecret("ermiterun_3")}, false)
251 assert.NoError(t, err)
252}
253
254func TestCISetRunnerVersion(t *testing.T) {
255 ctx := context.Background()
256 _, ci, _ := ciRepo(t)
257 h := domain.HashTokenSecret("ermiterun_v")
258 _, err := ci.CreateRunner(ctx, domain.CIRunner{ID: "r1", Name: "alpha", Version: "v1", TokenHash: h, RegisteredAt: created}, false)
259 require.NoError(t, err)
260
261 require.NoError(t, ci.SetRunnerVersion(ctx, "r1", "v2"))
262 got, err := ci.GetRunner(ctx, "r1")
263 require.NoError(t, err)
264 assert.Equal(t, "v2", got.Version)
265 assert.Equal(t, h, got.TokenHash, "the rest of the record is kept")
266 assert.Equal(t, "alpha", got.Name)
267
268 // An unchanged version is not written.
269 before, ok := ci.RawVersionForTest(ctx, "r1")
270 require.True(t, ok)
271 require.NoError(t, ci.SetRunnerVersion(ctx, "r1", "v2"))
272 after, _ := ci.RawVersionForTest(ctx, "r1")
273 assert.Equal(t, before, after)
274
275 assert.ErrorIs(t, ci.SetRunnerVersion(ctx, "gone", "v1"), domain.ErrNotFound)
276}
277
278func TestCISweepRemovesOldCompletedRunsOnly(t *testing.T) {
279 ctx := context.Background()
280 _, ci, n := ciRepo(t)
281 oldRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
282 require.NoError(t, err)
283 newRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
284 require.NoError(t, err)
285 openRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
286 require.NoError(t, err)
287
288 for _, tc := range []struct {
289 d ports.CIRunDraft
290 at time.Time
291 }{{oldRun, created}, {newRun, created.Add(100 * 24 * time.Hour)}} {
292 run, _ := ci.GetRun(ctx, n, tc.d.Run.ID)
293 run.Status, run.Conclusion, run.CompletedAt = domain.CICompleted, domain.CISuccess, tc.at
294 job, _ := ci.GetJob(ctx, run.ID, 1, tc.d.Jobs[0].ID)
295 job.Phase, job.Conclusion = domain.CIPhaseEnded, domain.CISuccess
296 _, err := ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{run}, Jobs: []domain.CIJob{job}})
297 require.NoError(t, err)
298 }
299
300 ids, err := ci.Sweep(ctx, created.Add(90*24*time.Hour))
301 require.NoError(t, err)
302 assert.Equal(t, []int64{oldRun.Run.ID}, ids)
303
304 _, err = ci.GetRun(ctx, n, oldRun.Run.ID)
305 assert.ErrorIs(t, err, domain.ErrNotFound)
306 _, err = ci.GetRunSource(ctx, oldRun.Run.ID)
307 assert.ErrorIs(t, err, domain.ErrNotFound)
308 jobs, _ := ci.ListJobs(ctx, oldRun.Run.ID, 1)
309 assert.Empty(t, jobs)
310 _, err = ci.GetRun(ctx, n, newRun.Run.ID)
311 assert.NoError(t, err)
312 _, err = ci.GetRun(ctx, n, openRun.Run.ID)
313 assert.NoError(t, err, "an open run is never swept")
314
315 // Numbers are never reused.
316 next, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
317 require.NoError(t, err)
318 assert.Equal(t, int64(4), next.Run.Number)
319}
320
321func TestCIRepositoryDeletionRemovesEverythingAndRestartsNumbers(t *testing.T) {
322 ctx := context.Background()
323 s, ci, n := ciRepo(t)
324 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 2))
325 require.NoError(t, err)
326 _, err = ci.CreateRunner(ctx, domain.CIRunner{ID: "r1", Name: "a", TokenHash: domain.HashTokenSecret("x")}, false)
327 require.NoError(t, err)
328
329 require.NoError(t, s.Delete(ctx, n))
330
331 _, err = ci.GetRun(ctx, n, d.Run.ID)
332 assert.ErrorIs(t, err, domain.ErrNotFound)
333 _, err = ci.GetRunSource(ctx, d.Run.ID)
334 assert.ErrorIs(t, err, domain.ErrNotFound)
335 jobs, _ := ci.ListJobs(ctx, d.Run.ID, 1)
336 assert.Empty(t, jobs)
337 checks, _, _ := ci.ListChecks(ctx, n, d.Run.Sha, 10)
338 assert.Empty(t, checks)
339 ready, _ := ci.ListReadyJobs(ctx, 10)
340 assert.Empty(t, ready)
341 open, _ := ci.ListOpenRuns(ctx)
342 assert.Empty(t, open)
343 runners, _ := ci.ListRunners(ctx)
344 assert.Len(t, runners, 1, "runners do not belong to a repository")
345
346 // A repository created again under the name starts its run numbers at 1,
347 // while ids are never reused.
348 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
349 again, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
350 require.NoError(t, err)
351 assert.Equal(t, int64(1), again.Run.Number)
352 assert.Equal(t, int64(2), again.Run.ID)
353}
354
355func TestCIStateSurvivesARestart(t *testing.T) {
356 ctx := context.Background()
357 dir := t.TempDir()
358 s, err := kvstore.Open(ctx, dir)
359 require.NoError(t, err)
360 n := name(t, "proj")
361 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
362 ci := s.CI()
363 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
364 require.NoError(t, err)
365 j := d.Jobs[0]
366 hash := domain.HashTokenSecret("ermitejob_z")
367 j.Phase, j.TokenHash, j.RunnerID, j.RunnerName = domain.CIPhaseRunning, hash, "r1", "alpha"
368 j.Steps = []domain.CIStep{{Number: 1, Name: "Set up job", State: domain.CIStepCompleted, Conclusion: domain.CISuccess, HasLines: true, LogStart: 0}}
369 j.Outputs = map[string]string{"a": "b"}
370 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
371 require.NoError(t, err)
372 require.NoError(t, s.Close(ctx))
373
374 s2, err := kvstore.Open(ctx, dir)
375 require.NoError(t, err)
376 defer func() { _ = s2.Close(ctx) }()
377 ci2 := s2.CI()
378 got, err := ci2.FindJobByToken(ctx, hash, time.Time{})
379 require.NoError(t, err)
380 assert.Equal(t, domain.CIPhaseRunning, got.Phase)
381 assert.Equal(t, "alpha", got.RunnerName)
382 assert.Equal(t, map[string]string{"a": "b"}, got.Outputs)
383 require.Len(t, got.Steps, 1)
384 assert.True(t, got.Steps[0].HasLines)
385 active, _ := ci2.ListActiveJobs(ctx)
386 assert.Len(t, active, 1)
387}
388
389func TestCICancelReasonsRoundTrip(t *testing.T) {
390 ctx := context.Background()
391 _, ci, n := ciRepo(t)
392 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
393 require.NoError(t, err)
394 for _, reason := range []domain.CICancelReason{domain.CICancelUser, domain.CICancelFailFast, domain.CICancelTimeout, domain.CICancelNone} {
395 j, err := ci.GetJob(ctx, d.Run.ID, 1, d.Jobs[0].ID)
396 require.NoError(t, err)
397 j.CancelReason, j.CancelMessage, j.CancelRequestedAt = reason, "why", created
398 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
399 require.NoError(t, err)
400 got, err := ci.GetJob(ctx, d.Run.ID, 1, d.Jobs[0].ID)
401 require.NoError(t, err)
402 assert.Equal(t, reason, got.CancelReason)
403 }
404}
405
406// Run.Name round-trips, and a run stored without one (before the field existed)
407// reads as its workflow name.
408func TestCIRunNameRoundTripsAndDefaultsToTheWorkflowName(t *testing.T) {
409 ctx := context.Background()
410 _, ci, n := ciRepo(t)
411
412 old, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 0))
413 require.NoError(t, err)
414 got, err := ci.GetRun(ctx, n, old.Run.ID)
415 require.NoError(t, err)
416 assert.Equal(t, "ci", got.WorkflowName)
417 assert.Equal(t, "ci", got.Name, "no stored name: the workflow name")
418
419 named, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", func(id, number int64) ports.CIRunDraft {
420 d := draftOf(n, ".github/workflows/ci.yml", 0)(id, number)
421 d.Run.Name = "run push on branch by someone"
422 return d
423 })
424 require.NoError(t, err)
425 got, err = ci.GetRun(ctx, n, named.Run.ID)
426 require.NoError(t, err)
427 assert.Equal(t, "run push on branch by someone", got.Name)
428 assert.Equal(t, "ci", got.WorkflowName, "the workflow name is unchanged")
429}
430
431func TestCIRunTrustedRoundTripsAndDefaultsToFalse(t *testing.T) {
432 ctx := context.Background()
433 _, ci, n := ciRepo(t)
434
435 old, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 0))
436 require.NoError(t, err)
437 got, err := ci.GetRun(ctx, n, old.Run.ID)
438 require.NoError(t, err)
439 assert.False(t, got.Trusted, "a run without the flag is not trusted")
440
441 trusted, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", func(id, number int64) ports.CIRunDraft {
442 d := draftOf(n, ".github/workflows/ci.yml", 0)(id, number)
443 d.Run.Trusted = true
444 return d
445 })
446 require.NoError(t, err)
447 got, err = ci.GetRun(ctx, n, trusted.Run.ID)
448 require.NoError(t, err)
449 assert.True(t, got.Trusted)
450}