Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240package db
import ( "context" "errors" "path/filepath" "slices" "strings" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/notifier" "tangled.org/core/spindle/models" pipelinecodec "tangled.org/core/spindle/pipeline" "tangled.org/core/workflow")
const pushRepo = "did:plc:pushrepo"
func pushRecord(t *testing.T, id, repoDid, ref, oldSha, newSha string) (*models.PipelineRecord, workflow.PipelineSpec) { t.Helper() did := repoDid raw := workflow.PipelineSpec{ TriggerMetadata: &workflow.TriggerMetadata{ Kind: "push", Repo: &workflow.TriggerRepo{Knot: "knot.test", RepoDid: &did, Did: repoDid}, Push: &workflow.PushTriggerData{Ref: ref, OldSha: oldSha, NewSha: newSha}, }, Workflows: []*workflow.WorkflowSpec{{Name: "ci.yml"}}, } record, err := pipelinecodec.FromTangled(models.PipelineId(id), time.Now(), raw) if err != nil { t.Fatal(err) } return record, raw}
func seedPush(t *testing.T, d *DB, id, repoDid, ref, oldSha, newSha string) { t.Helper() record, _ := pushRecord(t, id, repoDid, ref, oldSha, newSha) if err := d.CreatePipeline(record); err != nil { t.Fatal(err) }}
func TestEnqueuePipelineStoresTheRowAndItsJobTogether(t *testing.T) { d := newTestDB(t) ctx := context.Background() record, raw := pushRecord(t, "p1", pushRepo, "refs/heads/main", "aaa", "bbb") n := notifier.New() if err := d.EnqueuePipeline(ctx, pushRepo, record, nil, raw, "", "", &n); err != nil { t.Fatal(err) } if id, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "aaa", "bbb"); err != nil || !found || id != "p1" { t.Fatalf("PushPipeline = %q, %v, %v", id, found, err) } if job, err := d.DequeueJob(ctx, nil); err != nil || job == nil || job.PipelineId != "p1" { t.Fatalf("dequeued %+v, %v, want the pipeline's job", job, err) }}
// a pipeline row alone would make the replayed push look handled foreverfunc TestEnqueuePipelineLeavesNoRowWhenTheJobCannotBeStored(t *testing.T) { d := newTestDB(t) if _, err := d.Exec(`create trigger no_jobs before insert on jobs begin select raise(abort, 'disk full'); end`); err != nil { t.Fatal(err) } record, raw := pushRecord(t, "p1", pushRepo, "refs/heads/main", "aaa", "bbb") if err := d.EnqueuePipeline(context.Background(), pushRepo, record, nil, raw, "", "", nil); err == nil { t.Fatal("enqueue succeeded although the job insert failed") } if _, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "aaa", "bbb"); err != nil || found { t.Fatalf("a failed enqueue left a pipeline row behind, found = %v, err = %v", found, err) }}
func TestEnqueuePipelineForABannedRepoLeavesNoRow(t *testing.T) { d := newTestDB(t) if _, err := d.Exec(`insert into bans (subject_did) values (?)`, pushRepo); err != nil { t.Fatal(err) } record, raw := pushRecord(t, "p1", pushRepo, "refs/heads/main", "aaa", "bbb") if err := d.EnqueuePipeline(context.Background(), pushRepo, record, nil, raw, "", "", nil); !errors.Is(err, ErrBannedSubject) { t.Fatalf("err = %v, want ErrBannedSubject", err) } if _, found, _ := d.PushPipeline(pushRepo, "refs/heads/main", "aaa", "bbb"); found { t.Fatal("a refused pipeline left a row a replay would be deduped against") }}
func TestPushPipelineMatchesOnRepoRefAndBothShas(t *testing.T) { d := newTestDB(t) seedPush(t, d, "p1", pushRepo, "refs/heads/main", "aaa", "bbb")
for name, tc := range map[string]struct { repo, ref, old, new string want bool }{ "same push": {pushRepo, "refs/heads/main", "aaa", "bbb", true}, "another repo": {"did:plc:other", "refs/heads/main", "aaa", "bbb", false}, "another ref": {pushRepo, "refs/heads/dev", "aaa", "bbb", false}, "another old sha": {pushRepo, "refs/heads/main", "ccc", "bbb", false}, "another new sha": {pushRepo, "refs/heads/main", "aaa", "ccc", false}, "the push in reverse (A-B then B-A)": {pushRepo, "refs/heads/main", "bbb", "aaa", false}, } { id, found, err := d.PushPipeline(tc.repo, tc.ref, tc.old, tc.new) if err != nil || found != tc.want { t.Errorf("%s: found = %v, err = %v, want found = %v", name, found, err, tc.want) } if found && id != "p1" { t.Errorf("%s: id = %q, want p1", name, id) } }}
func TestPushPipelineIgnoresOtherTriggerKinds(t *testing.T) { d := newTestDB(t) seedPipeline(t, d, "p-manual", pushRepo, "manual") if _, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "", "sha-p-manual"); err != nil || found { t.Fatalf("a manual pipeline matched a push: found = %v, err = %v", found, err) }}
func TestPushPipelineLookupUsesTheIndex(t *testing.T) { d := newTestDB(t) rows, err := d.Query(`explain query plan select pipeline_id from pipelines where kind = 'push' and repo_did = ? and commit_sha = ? and ref = ? and old_sha = ? limit 1`, pushRepo, "b", "refs/heads/main", "a") if err != nil { t.Fatal(err) } defer rows.Close() var plan []string for rows.Next() { var id, parent, unused int var detail string if err := rows.Scan(&id, &parent, &unused, &detail); err != nil { t.Fatal(err) } plan = append(plan, detail) } if got := strings.Join(plan, "; "); !strings.Contains(got, "idx_pipelines_push_dedupe") { t.Fatalf("plan = %q, want the push dedupe index", got) }}
func TestPushDedupeMigrationKeepsExistingPipelines(t *testing.T) { path := filepath.Join(t.TempDir(), "spindle.db") ctx := context.Background() d, err := Make(ctx, path) if err != nil { t.Fatal(err) } seedPush(t, d, "old-run", pushRepo, "refs/heads/main", "aaa", "bbb") if _, err := d.Exec(` drop index idx_pipelines_push_dedupe; alter table pipelines drop column ref; alter table pipelines drop column old_sha; delete from migrations where name = 'pipelines-push-dedupe'; `); err != nil { t.Fatal(err) } d.Close()
d, err = Make(ctx, path) if err != nil { t.Fatalf("reopening an old db: %v", err) } defer d.Close() if _, err := d.GetPipeline(ctx, "old-run"); err != nil { t.Fatalf("pipeline lost in the migration: %v", err) } seedPush(t, d, "new-run", pushRepo, "refs/heads/main", "bbb", "ccc") if id, found, err := d.PushPipeline(pushRepo, "refs/heads/main", "bbb", "ccc"); err != nil || !found || id != "new-run" { t.Fatalf("PushPipeline after migrating = %q, %v, %v", id, found, err) }}
func TestRepoRetriesSurviveReopen(t *testing.T) { path := filepath.Join(t.TempDir(), "spindle.db") d, err := Make(context.Background(), path) if err != nil { t.Fatal(err) } now := time.Unix(5000, 0) if err := d.PutRepoRetry("did:plc:a", 2, now); err != nil { t.Fatal(err) } if err := d.PutRepoRetry("did:plc:b", 1, now.Add(time.Hour)); err != nil { t.Fatal(err) } d.Close()
d, err = Make(context.Background(), path) if err != nil { t.Fatal(err) } defer d.Close() if due, err := d.DueRepoRetries(now); err != nil || !slices.Equal(due, []syntax.DID{"did:plc:a"}) { t.Fatalf("due = %v, %v, want only a", due, err) } if n, err := d.RepoRetryAttempts("did:plc:a"); err != nil || n != 2 { t.Fatalf("attempts = %d, %v, want 2", n, err) } if err := d.DeleteRepoRetry("did:plc:a"); err != nil { t.Fatal(err) } if n, _ := d.RepoRetryAttempts("did:plc:a"); n != 0 { t.Fatalf("attempts after delete = %d", n) }}
func TestPendingCatchUpKeepsTheOriginalTimeAndAgesOut(t *testing.T) { d := newTestDB(t) now := time.Unix(1_000_000, 0) if err := d.AddPendingCatchUp([]syntax.DID{"did:plc:a", "did:plc:b"}, now.Add(-10*24*time.Hour)); err != nil { t.Fatal(err) } // re-adding keeps the first timestamp, so did:plc:a stays stale if err := d.AddPendingCatchUp([]syntax.DID{"did:plc:a", "did:plc:c"}, now); err != nil { t.Fatal(err) } dropped, err := d.DropStalePendingCatchUp(now.Add(-7 * 24 * time.Hour)) slices.Sort(dropped) if err != nil || !slices.Equal(dropped, []syntax.DID{"did:plc:a", "did:plc:b"}) { t.Fatalf("dropped = %v, %v, want a and b", dropped, err) } if left, err := d.PendingCatchUp(); err != nil || !slices.Equal(left, []syntax.DID{"did:plc:c"}) { t.Fatalf("left = %v, %v, want c", left, err) } if err := d.DeletePendingCatchUp("did:plc:c", "did:plc:nobody"); err != nil { t.Fatal(err) } if left, _ := d.PendingCatchUp(); len(left) != 0 { t.Fatalf("left after delete = %v", left) }}