| @@ -1,111 +0,0 @@ |
| 1 | | - | // SPDX-License-Identifier: AGPL-3.0-or-later |
| 2 | | - | |
| 3 | | - | package kvstore |
| 4 | | - | |
| 5 | | - | import ( |
| 6 | | - | "context" |
| 7 | | - | "strings" |
| 8 | | - | "testing" |
| 9 | | - | "time" |
| 10 | | - | |
| 11 | | - | kv "bnema.dev/ermite-kv" |
| 12 | | - | storagepb "bnema.dev/ermite/gen/ermite/storage" |
| 13 | | - | "bnema.dev/ermite/internal/domain" |
| 14 | | - | "bnema.dev/ermite/internal/ports" |
| 15 | | - | "github.com/stretchr/testify/require" |
| 16 | | - | "google.golang.org/protobuf/proto" |
| 17 | | - | ) |
| 18 | | - | |
| 19 | | - | func TestCIJobContextsStorage(t *testing.T) { |
| 20 | | - | ctx := context.Background() |
| 21 | | - | s, err := Open(ctx, t.TempDir()) |
| 22 | | - | require.NoError(t, err) |
| 23 | | - | t.Cleanup(func() { require.NoError(t, s.Close(ctx)) }) |
| 24 | | - | repo, err := domain.ParseRepoName("proj") |
| 25 | | - | require.NoError(t, err) |
| 26 | | - | require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate})) |
| 27 | | - | ci := s.CI() |
| 28 | | - | payload := `{"event":"` + strings.Repeat("payload", 3500) + `"}` |
| 29 | | - | j := domain.CIJob{ID: "build", Repo: repo, RunID: 1, RunAttempt: 1, Attempt: 1, Phase: domain.CIPhaseReady, ContextsJSON: payload} |
| 30 | | - | save := func() { |
| 31 | | - | v, err := ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}}) |
| 32 | | - | require.NoError(t, err) |
| 33 | | - | j.Version = v |
| 34 | | - | } |
| 35 | | - | save() |
| 36 | | - | contextsKey := jobContextsKey(1, 1, j.ID) |
| 37 | | - | initial, ok := ci.db.Get(ctx, contextsKey) |
| 38 | | - | require.True(t, ok) |
| 39 | | - | for i := range 5 { |
| 40 | | - | j.Steps = []domain.CIStep{{Number: int32(i + 1), Name: "step", State: domain.CIStepInProgress}} |
| 41 | | - | save() |
| 42 | | - | stored, ok := ci.db.Get(ctx, jobKey(1, 1, j.ID)) |
| 43 | | - | require.True(t, ok) |
| 44 | | - | require.NotContains(t, string(stored.Value), payload) |
| 45 | | - | var pb storagepb.CIJob |
| 46 | | - | require.NoError(t, proto.Unmarshal(stored.Value, &pb)) |
| 47 | | - | require.Empty(t, pb.GetContextsJson()) |
| 48 | | - | contexts, ok := ci.db.Get(ctx, contextsKey) |
| 49 | | - | require.True(t, ok) |
| 50 | | - | require.Equal(t, initial.Version, contexts.Version, "unchanged contexts must not be rewritten") |
| 51 | | - | got, err := ci.GetJob(ctx, 1, 1, j.ID) |
| 52 | | - | require.NoError(t, err) |
| 53 | | - | require.Equal(t, payload, got.ContextsJSON) |
| 54 | | - | jobs, err := ci.ListJobs(ctx, 1, 1) |
| 55 | | - | require.NoError(t, err) |
| 56 | | - | require.Equal(t, payload, jobs[0].ContextsJSON) |
| 57 | | - | ready, err := ci.ListReadyJobs(ctx, 0) |
| 58 | | - | require.NoError(t, err) |
| 59 | | - | require.Equal(t, payload, ready[0].ContextsJSON) |
| 60 | | - | } |
| 61 | | - | // A kept job in a re-run gets an independent contexts key. |
| 62 | | - | j.RunAttempt = 2 |
| 63 | | - | j.Version = 0 |
| 64 | | - | save() |
| 65 | | - | copied, ok := ci.db.Get(ctx, jobContextsKey(1, 2, j.ID)) |
| 66 | | - | require.True(t, ok) |
| 67 | | - | require.Equal(t, payload, string(copied.Value)) |
| 68 | | - | j.ContextsJSON = "" |
| 69 | | - | save() |
| 70 | | - | got, err := ci.GetJob(ctx, 1, 2, j.ID) |
| 71 | | - | require.NoError(t, err) |
| 72 | | - | require.Empty(t, got.ContextsJSON) |
| 73 | | - | // Sweep must remove contexts from every attempt as well as job records. |
| 74 | | - | _, err = ci.Apply(ctx, ports.CIWrite{Runs: []domain.CIRun{{Repo: repo, ID: 1, Status: domain.CICompleted, CompletedAt: time.Now().Add(-time.Hour)}}}) |
| 75 | | - | require.NoError(t, err) |
| 76 | | - | ids, err := ci.Sweep(ctx, time.Now()) |
| 77 | | - | require.NoError(t, err) |
| 78 | | - | require.Equal(t, []int64{1}, ids) |
| 79 | | - | for range ci.db.Snapshot(ctx).Prefix(ctx, []byte(jobCtxPrefix)) { |
| 80 | | - | t.Fatal("orphan contexts key") |
| 81 | | - | } |
| 82 | | - | } |
| 83 | | - | |
| 84 | | - | func TestCIJobLegacyContexts(t *testing.T) { |
| 85 | | - | ctx := context.Background() |
| 86 | | - | s, err := Open(ctx, t.TempDir()) |
| 87 | | - | require.NoError(t, err) |
| 88 | | - | t.Cleanup(func() { require.NoError(t, s.Close(ctx)) }) |
| 89 | | - | ci := s.CI() |
| 90 | | - | var b kv.Batch |
| 91 | | - | b.Put(jobKey(1, 1, "build"), mustMarshal(&storagepb.CIJob{Id: "build", Repo: "proj", RunId: 1, RunAttempt: 1, ContextsJson: `{"legacy":true}`})) |
| 92 | | - | _, err = ci.db.Apply(ctx, &b) |
| 93 | | - | require.NoError(t, err) |
| 94 | | - | j, err := ci.GetJob(ctx, 1, 1, "build") |
| 95 | | - | require.NoError(t, err) |
| 96 | | - | require.Equal(t, `{"legacy":true}`, j.ContextsJSON) |
| 97 | | - | jobs, err := ci.ListJobs(ctx, 1, 1) |
| 98 | | - | require.NoError(t, err) |
| 99 | | - | require.Equal(t, j.ContextsJSON, jobs[0].ContextsJSON) |
| 100 | | - | // Updating the legacy job moves its contexts out of the protobuf. |
| 101 | | - | _, err = ci.Apply(ctx, ports.CIWrite{Jobs: []domain.CIJob{j}}) |
| 102 | | - | require.NoError(t, err) |
| 103 | | - | got, err := ci.GetJob(ctx, 1, 1, "build") |
| 104 | | - | require.NoError(t, err) |
| 105 | | - | require.Equal(t, j.ContextsJSON, got.ContextsJSON) |
| 106 | | - | item, ok := ci.db.Get(ctx, jobKey(1, 1, "build")) |
| 107 | | - | require.True(t, ok) |
| 108 | | - | pbJob, err := decodeJob(item) |
| 109 | | - | require.NoError(t, err) |
| 110 | | - | require.Empty(t, pbJob.ContextsJSON) |
| 111 | | - | } |