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 TestCISweepRemovesOldCompletedRunsOnly(t *testing.T) {
255 ctx := context.Background()
256 _, ci, n := ciRepo(t)
257 oldRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
258 require.NoError(t, err)
259 newRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
260 require.NoError(t, err)
261 openRun, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
262 require.NoError(t, err)
263
264 for _, tc := range []struct {
265 d ports.CIRunDraft
266 at time.Time
267 }{{oldRun, created}, {newRun, created.Add(100 * 24 * time.Hour)}} {
268 run, _ := ci.GetRun(ctx, n, tc.d.Run.ID)
269 run.Status, run.Conclusion, run.CompletedAt = domain.CICompleted, domain.CISuccess, tc.at
270 job, _ := ci.GetJob(ctx, run.ID, 1, tc.d.Jobs[0].ID)
271 job.Phase, job.Conclusion = domain.CIPhaseEnded, domain.CISuccess
272 _, err := ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{run}, Jobs: []domain.CIJob{job}})
273 require.NoError(t, err)
274 }
275
276 ids, err := ci.Sweep(ctx, created.Add(90*24*time.Hour))
277 require.NoError(t, err)
278 assert.Equal(t, []int64{oldRun.Run.ID}, ids)
279
280 _, err = ci.GetRun(ctx, n, oldRun.Run.ID)
281 assert.ErrorIs(t, err, domain.ErrNotFound)
282 _, err = ci.GetRunSource(ctx, oldRun.Run.ID)
283 assert.ErrorIs(t, err, domain.ErrNotFound)
284 jobs, _ := ci.ListJobs(ctx, oldRun.Run.ID, 1)
285 assert.Empty(t, jobs)
286 _, err = ci.GetRun(ctx, n, newRun.Run.ID)
287 assert.NoError(t, err)
288 _, err = ci.GetRun(ctx, n, openRun.Run.ID)
289 assert.NoError(t, err, "an open run is never swept")
290
291 // Numbers are never reused.
292 next, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
293 require.NoError(t, err)
294 assert.Equal(t, int64(4), next.Run.Number)
295}
296
297func TestCIRepositoryDeletionRemovesEverythingAndRestartsNumbers(t *testing.T) {
298 ctx := context.Background()
299 s, ci, n := ciRepo(t)
300 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 2))
301 require.NoError(t, err)
302 _, err = ci.CreateRunner(ctx, domain.CIRunner{ID: "r1", Name: "a", TokenHash: domain.HashTokenSecret("x")}, false)
303 require.NoError(t, err)
304
305 require.NoError(t, s.Delete(ctx, n))
306
307 _, err = ci.GetRun(ctx, n, d.Run.ID)
308 assert.ErrorIs(t, err, domain.ErrNotFound)
309 _, err = ci.GetRunSource(ctx, d.Run.ID)
310 assert.ErrorIs(t, err, domain.ErrNotFound)
311 jobs, _ := ci.ListJobs(ctx, d.Run.ID, 1)
312 assert.Empty(t, jobs)
313 checks, _, _ := ci.ListChecks(ctx, n, d.Run.Sha, 10)
314 assert.Empty(t, checks)
315 ready, _ := ci.ListReadyJobs(ctx, 10)
316 assert.Empty(t, ready)
317 open, _ := ci.ListOpenRuns(ctx)
318 assert.Empty(t, open)
319 runners, _ := ci.ListRunners(ctx)
320 assert.Len(t, runners, 1, "runners do not belong to a repository")
321
322 // A repository created again under the name starts its run numbers at 1,
323 // while ids are never reused.
324 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
325 again, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 0))
326 require.NoError(t, err)
327 assert.Equal(t, int64(1), again.Run.Number)
328 assert.Equal(t, int64(2), again.Run.ID)
329}
330
331func TestCIStateSurvivesARestart(t *testing.T) {
332 ctx := context.Background()
333 dir := t.TempDir()
334 s, err := kvstore.Open(ctx, dir)
335 require.NoError(t, err)
336 n := name(t, "proj")
337 require.NoError(t, s.Create(ctx, domain.Repo{Name: n, Visibility: domain.VisibilityPrivate, CreatedAt: created}))
338 ci := s.CI()
339 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
340 require.NoError(t, err)
341 j := d.Jobs[0]
342 hash := domain.HashTokenSecret("ermitejob_z")
343 j.Phase, j.TokenHash, j.RunnerID, j.RunnerName = domain.CIPhaseRunning, hash, "r1", "alpha"
344 j.Steps = []domain.CIStep{{Number: 1, Name: "Set up job", State: domain.CIStepCompleted, Conclusion: domain.CISuccess, HasLines: true, LogStart: 0}}
345 j.Outputs = map[string]string{"a": "b"}
346 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
347 require.NoError(t, err)
348 require.NoError(t, s.Close(ctx))
349
350 s2, err := kvstore.Open(ctx, dir)
351 require.NoError(t, err)
352 defer func() { _ = s2.Close(ctx) }()
353 ci2 := s2.CI()
354 got, err := ci2.FindJobByToken(ctx, hash, time.Time{})
355 require.NoError(t, err)
356 assert.Equal(t, domain.CIPhaseRunning, got.Phase)
357 assert.Equal(t, "alpha", got.RunnerName)
358 assert.Equal(t, map[string]string{"a": "b"}, got.Outputs)
359 require.Len(t, got.Steps, 1)
360 assert.True(t, got.Steps[0].HasLines)
361 active, _ := ci2.ListActiveJobs(ctx)
362 assert.Len(t, active, 1)
363}
364
365func TestCICancelReasonsRoundTrip(t *testing.T) {
366 ctx := context.Background()
367 _, ci, n := ciRepo(t)
368 d, err := ci.CreateRun(ctx, n, "w.yml", draftOf(n, "w.yml", 1))
369 require.NoError(t, err)
370 for _, reason := range []domain.CICancelReason{domain.CICancelUser, domain.CICancelFailFast, domain.CICancelTimeout, domain.CICancelNone} {
371 j, err := ci.GetJob(ctx, d.Run.ID, 1, d.Jobs[0].ID)
372 require.NoError(t, err)
373 j.CancelReason, j.CancelMessage, j.CancelRequestedAt = reason, "why", created
374 _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}})
375 require.NoError(t, err)
376 got, err := ci.GetJob(ctx, d.Run.ID, 1, d.Jobs[0].ID)
377 require.NoError(t, err)
378 assert.Equal(t, reason, got.CancelReason)
379 }
380}
381
382// Run.Name round-trips, and a run stored without one (before the field existed)
383// reads as its workflow name.
384func TestCIRunNameRoundTripsAndDefaultsToTheWorkflowName(t *testing.T) {
385 ctx := context.Background()
386 _, ci, n := ciRepo(t)
387
388 old, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 0))
389 require.NoError(t, err)
390 got, err := ci.GetRun(ctx, n, old.Run.ID)
391 require.NoError(t, err)
392 assert.Equal(t, "ci", got.WorkflowName)
393 assert.Equal(t, "ci", got.Name, "no stored name: the workflow name")
394
395 named, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", func(id, number int64) ports.CIRunDraft {
396 d := draftOf(n, ".github/workflows/ci.yml", 0)(id, number)
397 d.Run.Name = "run push on branch by someone"
398 return d
399 })
400 require.NoError(t, err)
401 got, err = ci.GetRun(ctx, n, named.Run.ID)
402 require.NoError(t, err)
403 assert.Equal(t, "run push on branch by someone", got.Name)
404 assert.Equal(t, "ci", got.WorkflowName, "the workflow name is unchanged")
405}
406
407func TestCIRunTrustedRoundTripsAndDefaultsToFalse(t *testing.T) {
408 ctx := context.Background()
409 _, ci, n := ciRepo(t)
410
411 old, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", draftOf(n, ".github/workflows/ci.yml", 0))
412 require.NoError(t, err)
413 got, err := ci.GetRun(ctx, n, old.Run.ID)
414 require.NoError(t, err)
415 assert.False(t, got.Trusted, "a run without the flag is not trusted")
416
417 trusted, err := ci.CreateRun(ctx, n, ".github/workflows/ci.yml", func(id, number int64) ports.CIRunDraft {
418 d := draftOf(n, ".github/workflows/ci.yml", 0)(id, number)
419 d.Run.Trusted = true
420 return d
421 })
422 require.NoError(t, err)
423 got, err = ci.GetRun(ctx, n, trusted.Run.ID)
424 require.NoError(t, err)
425 assert.True(t, got.Trusted)
426}