Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485package spindle
import ( "context" "errors" "slices" "strings" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/xrpc" "tangled.org/core/workflow")
const ( prWorkflow = "engine: dummy\nwhen:\n - event: pull_request\n branch: main\nsteps:\n - name: hi\n command: echo hi\n" doneWorkflow = "engine: dummy\nwhen:\n - event: pull_request\n branch: main\n types: [closed, merged]\nsteps:\n - name: bye\n command: echo bye\n" otherPull = "3kpull0000002")
type prPipelineEnv struct { *pullEnv repo *fakeRepo}
func newPRPipelineEnv(t *testing.T) *prPipelineEnv { t.Helper() w := newWorld(t) repo := w.serveGitRepo(testRepoDid) n := w.boot() w.seed(n, testBob) e := &pullEnv{world: w, n: n, sink: newSink(t)} w.subscribe(n, e.sink, pullEvents...) return &prPipelineEnv{pullEnv: e, repo: repo}}
func (e *prPipelineEnv) head(t *testing.T) string { t.Helper() return e.repo.commitWorkflows(t, map[string]string{"pr.yaml": prWorkflow, "done.yaml": doneWorkflow})}
func (e *prPipelineEnv) ticketAs(rkey, state string, editor syntax.DID, op string) { e.putTicket(testRepoDid, rkey, state, editor, true) e.n.consumed(e.emitTicket(testRepoDid, rkey, op))}
func (e *prPipelineEnv) patchAs(rkey, ticket, head string, editor syntax.DID) { e.putPatch(testRepoDid, rkey, ticket, head, editor) e.n.consumed(e.emitPatch(testRepoDid, rkey, opCreate))}
func prPipelines(t *testing.T, n *node) []*models.PipelineRecord { t.Helper() records, _, _, err := n.s.db.QueryPipelines(context.Background(), testRepoDid.String(), nil, "", []string{"pull_request"}, 100) if err != nil { t.Fatal(err) } slices.SortFunc(records, func(a, b *models.PipelineRecord) int { return strings.Compare(string(a.ID), string(b.ID)) }) return records}
// like the sink, waits for the count and then sees that nothing else followsfunc expectPRPipelines(t *testing.T, n *node, want int) []*models.PipelineRecord { t.Helper() if want > 0 { eventually(t, 30*time.Second, "the pull request pipelines", func() bool { return len(prPipelines(t, n)) >= want }) } time.Sleep(settleTime) got := prPipelines(t, n) if len(got) != want { t.Fatalf("got %d pull request pipelines, want %d: %s", len(got), want, describePipelines(got)) } return got}
func describePipelines(records []*models.PipelineRecord) string { var out []string for _, r := range records { pr := r.Trigger.PullRequest action := "" if pr.Action != nil { action = *pr.Action } out = append(out, action+"@"+r.Commit) } return "[" + strings.Join(out, " ") + "]"}
func wantPRPipeline(t *testing.T, r *models.PipelineRecord, action, sha, ticket string, workflows ...string) { t.Helper() pr := r.Trigger.PullRequest if pr == nil || pr.Action == nil || *pr.Action != action { t.Fatalf("pipeline %s: trigger %+v, want action %s", r.ID, pr, action) } if r.Commit != sha || pr.SourceCommit != sha { t.Errorf("%s: commit %s source %s, want %s", action, r.Commit, pr.SourceCommit, sha) } if pr.TargetBranch != "main" { t.Errorf("%s: target branch %q, want main", action, pr.TargetBranch) } if want := "at://" + testRepoDid.String() + "/" + org_tangled.TrackTicketNSID + "/" + ticket; pr.Pull == nil || *pr.Pull != want { t.Errorf("%s: pull %v, want %s", action, pr.Pull, want) } var names []string for _, w := range r.Workflows { names = append(names, w.Name) } if !slices.Equal(names, workflows) { t.Errorf("%s: workflows %v, want %v", action, names, workflows) }}
func TestPullRequestActionsTriggerPipelines(t *testing.T) { e := newPRPipelineEnv(t) sha1, sha2 := e.head(t), e.head(t)
e.ticketAs(pullRkey, "open", testBob, opCreate) e.patchAs(patchOne, pullRkey, sha1, testBob) got := expectPRPipelines(t, e.n, 1) wantPRPipeline(t, got[0], "opened", sha1, pullRkey, "pr.yaml")
e.patchAs(patchTwo, pullRkey, sha2, testBob) got = expectPRPipelines(t, e.n, 2) wantPRPipeline(t, got[1], "synchronize", sha2, pullRkey, "pr.yaml")
// only the types: [closed, merged] workflow runs on a close or a merge e.ticketAs(pullRkey, "closed", testOwner, "update") got = expectPRPipelines(t, e.n, 3) wantPRPipeline(t, got[2], "closed", sha2, pullRkey, "done.yaml")
e.ticketAs(pullRkey, "open", testBob, "update") got = expectPRPipelines(t, e.n, 4) wantPRPipeline(t, got[3], "reopened", sha2, pullRkey, "pr.yaml")
e.ticketAs(pullRkey, "completed", testOwner, "update") got = expectPRPipelines(t, e.n, 5) wantPRPipeline(t, got[4], "merged", sha2, pullRkey, "done.yaml") e.sink.expect(t, 5)}
// a blocked pipeline has blocked workflows and no job, a running one has neitherfunc wantBlocked(t *testing.T, n *node, r *models.PipelineRecord, blocked bool) { t.Helper() for _, w := range r.Workflows { if (w.Status == string(models.StatusKindBlocked)) != blocked { t.Errorf("pipeline %s workflow %s is %s, want blocked %v", r.ID, w.Name, w.Status, blocked) } } if !blocked { return } var jobs int if err := n.s.db.QueryRow(`select count(*) from jobs where pipeline_id = ?`, r.ID).Scan(&jobs); err != nil { t.Fatal(err) } if jobs != 0 { t.Errorf("blocked pipeline %s has %d jobs, want none", r.ID, jobs) }}
func TestPullRequestPipelineNeedsPushAccess(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t)
// dave isn't a collaborator, so his pull's pipelines wait for approval e.ticketAs(otherPull, "open", testDave, opCreate) e.patchAs("bafydavepatch", otherPull, sha, testDave) e.ticketAs(otherPull, "closed", testDave, "update") e.ticketAs(otherPull, "open", testDave, "update") e.sink.expect(t, 3) for _, r := range expectPRPipelines(t, e.n, 3) { wantBlocked(t, e.n, r, true) }
e.ticketAs(pullRkey, "open", testBob, opCreate) e.patchAs(patchOne, pullRkey, sha, testBob) got := expectPRPipelines(t, e.n, 4) wantBlocked(t, e.n, got[3], false)
// bob may run pipelines, but carol acting on his pull may not e.ticketAs(pullRkey, "closed", testCarol, "update") e.patchAs(patchTwo, pullRkey, e.head(t), testCarol) e.sink.expect(t, 6) got = expectPRPipelines(t, e.n, 6) wantBlocked(t, e.n, got[4], true) wantBlocked(t, e.n, got[5], true)
e.ticketAs(pullRkey, "open", testOwner, "update") got = expectPRPipelines(t, e.n, 7) if a := got[6].Trigger.PullRequest.Action; a == nil || *a != "reopened" { t.Errorf("the owner's reopen ran %v", a) } wantBlocked(t, e.n, got[6], false)}
// every round's pipelines link back to the patch they ran, even a close that came laterfunc TestPullRequestPipelineKeepsItsPatch(t *testing.T) { e := newPRPipelineEnv(t) e.ticketAs(pullRkey, "open", testBob, opCreate) e.patchAs(patchOne, pullRkey, e.head(t), testBob) expectPRPipelines(t, e.n, 1) e.patchAs(patchTwo, pullRkey, e.head(t), testBob) expectPRPipelines(t, e.n, 2) e.ticketAs(pullRkey, "closed", testOwner, "update") got := expectPRPipelines(t, e.n, 3) for i, rkey := range []string{patchOne, patchTwo, patchTwo} { var patch *string if err := e.n.s.db.QueryRow(`select patch_uri from pipelines where pipeline_id = ?`, got[i].ID).Scan(&patch); err != nil { t.Fatal(err) } if want := "at://" + testRepoDid.String() + "/" + org_tangled.ReviewPatchNSID + "/" + rkey; patch == nil || *patch != want { t.Errorf("pipeline %d patch_uri %v, want %s", i, patch, want) } }}
func TestApprovedPullRequestPipelineIsQueued(t *testing.T) { e := newPRPipelineEnv(t) e.ticketAs(otherPull, "open", testDave, opCreate) e.patchAs("bafydavepatch", otherPull, e.head(t), testDave) got := expectPRPipelines(t, e.n, 1) wantBlocked(t, e.n, got[0], true)
if err := e.n.s.ApprovePipeline(context.Background(), testRepoDid, got[0].ID); err != nil { t.Fatal(err) } // the test world runs no job workers, so the job waits in the queue wid := models.WorkflowId{PipelineId: got[0].ID, Name: "pr.yaml"} if st, err := e.n.s.db.GetStatus(wid); err != nil || st != models.StatusKindPending { t.Fatalf("approved workflow is %s (%v), want pending", st, err) } var jobs int if err := e.n.s.db.QueryRow(`select count(*) from jobs where pipeline_id = ?`, got[0].ID).Scan(&jobs); err != nil || jobs != 1 { t.Fatalf("approved pipeline has %d jobs (%v), want 1", jobs, err) } if err := e.n.s.ApprovePipeline(context.Background(), testRepoDid, got[0].ID); !errors.Is(err, db.ErrNotBlocked) { t.Fatalf("second approve = %v, want ErrNotBlocked", err) }}
func TestRetryRerunsTheStoredSpec(t *testing.T) { e := newPRPipelineEnv(t) ctx := context.Background() sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate) e.patchAs(patchOne, pullRkey, sha, testBob) got := expectPRPipelines(t, e.n, 1)
retryId, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[0].ID, nil) if err != nil { t.Fatal(err) } got = expectPRPipelines(t, e.n, 2) retried := got[1] if retried.ID != retryId || retried.ID == got[0].ID { t.Fatalf("retry made %s, want a new pipeline %s", retried.ID, retryId) } wantPRPipeline(t, retried, "opened", sha, pullRkey, "pr.yaml") wantBlocked(t, e.n, retried, false) stored, err := e.n.s.db.GetStoredPipeline(retryId) if err != nil { t.Fatal(err) } if a := stored.Spec.TriggerMetadata.Actor; a == nil || *a != testOwner.String() { t.Errorf("retry ran as %v, want the retrier %s", a, testOwner) } var patches int if err := e.n.s.db.QueryRow(`select count(distinct patch_uri) from pipelines where pipeline_id in (?, ?)`, got[0].ID, retryId).Scan(&patches); err != nil || patches != 1 { t.Errorf("the retry links %d patches (%v), want the original's", patches, err) }
if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[0].ID, []string{"nope.yaml"}); !errors.Is(err, xrpc.ErrNoMatchingWorkflows) { t.Errorf("retrying only an unknown workflow = %v, want ErrNoMatchingWorkflows", err) }
// dave's pipeline has to be approved, a retry would skip that e.ticketAs(otherPull, "open", testDave, opCreate) e.patchAs("bafydavepatch", otherPull, sha, testDave) got = expectPRPipelines(t, e.n, 3) if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[2].ID, nil); !errors.Is(err, xrpc.ErrNotRetryable) { t.Fatalf("retrying a blocked pipeline = %v, want ErrNotRetryable", err) } if err := e.n.s.ApprovePipeline(ctx, testRepoDid, got[2].ID); err != nil { t.Fatal(err) } if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, got[2].ID, nil); err != nil { t.Fatalf("retrying an approved pipeline: %v", err) }
// pipelines from before specs were stored can't be rerun if err := e.n.s.db.CreatePipeline(&models.PipelineRecord{ID: "legacy", RepoDID: testRepoDid.String()}); err != nil { t.Fatal(err) } if _, err := e.n.s.RetryPipeline(ctx, testOwner, testRepoDid, "legacy", nil); !errors.Is(err, xrpc.ErrNotRetryable) { t.Fatalf("retrying a pipeline without a spec = %v, want ErrNotRetryable", err) }}
func TestPullRequestPipelineSkipsBannedRepo(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) if err := e.n.s.db.PutBan(db.BanEntry{SubjectDid: testRepoDid}); err != nil { t.Fatal(err) } e.ticketAs(pullRkey, "open", testBob, opCreate) e.patchAs(patchOne, pullRkey, sha, testBob) e.sink.expect(t, 1) expectPRPipelines(t, e.n, 0)}
// the retry loop only looks every tick, so move the clock past the backoff and wait for itfunc (e *prPipelineEnv) skipBackoff() { e.skew.Add(int64(2 * time.Hour))}
func TestPullRequestEventSurvivesAKnotOutage(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate)
e.setKnotDown(true) e.patchAs(patchOne, pullRkey, sha, testBob) e.sink.expect(t, 0) expectPRPipelines(t, e.n, 0)
e.setKnotDown(false) e.skipBackoff() got := expectPRPipelines(t, e.n, 1) wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml") e.sink.expect(t, 1)
// another round of the retry loop finds nothing left to do e.skipBackoff() time.Sleep(5 * testRetryTick) expectPRPipelines(t, e.n, 1) e.sink.expect(t, 1)}
func TestPullRequestEventRunsOnceAfterARestart(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate)
// the webhook goes out, then spindle dies while it fetches the head release := e.holdGit() e.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob) at := e.emitPatch(testRepoDid, patchOne, opCreate) eventually(t, 15*time.Second, "the head fetch to start", func() bool { return e.gitFetches.Load() > 0 }) e.sink.expect(t, 1) e.n.stop() release(false)
n2 := e.boot() got := expectPRPipelines(t, n2, 1) wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml") n2.consumed(at) e.sink.expect(t, 1)}
func TestPullRequestEventHangingKnotDoesNotStallIngest(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate)
e.serve(faultOtherRepoDid, "other", testBob) e.emitDeclaration("other", faultOtherRepoDid) e.n.consumed(e.emitManifest(faultOtherRepoDid)) eventually(t, 15*time.Second, "the other repo to be served", func() bool { return len(repoRows(t, e.n.s, faultOtherRepoDid)) == 1 })
release := e.holdGit() defer release(false) e.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob) e.emitPatch(testRepoDid, patchOne, opCreate) eventually(t, 15*time.Second, "the head fetch to start", func() bool { return e.gitFetches.Load() > 0 })
// the next event is for another repo, so the stuck fetch can't be what it waits on e.putTicket(faultOtherRepoDid, pullRkey, "open", testBob, true) e.n.consumed(e.emitTicket(faultOtherRepoDid, pullRkey, opCreate)) if _, err := e.n.s.db.GetPullTicket(faultOtherRepoDid, pullRkey); err != nil { t.Fatalf("the other repo's ticket wasn't stored: %v", err) }
release(false) got := expectPRPipelines(t, e.n, 1) wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml")}
func TestPullRequestPipelineRetriesAFailedFetch(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate)
e.gitFailing.Store(true) e.patchAs(patchOne, pullRkey, sha, testBob) e.sink.expect(t, 1) expectPRPipelines(t, e.n, 0)
e.gitFailing.Store(false) e.skipBackoff() got := expectPRPipelines(t, e.n, 1) wantPRPipeline(t, got[0], "opened", sha, pullRkey, "pr.yaml") e.sink.expect(t, 1)}
func TestPullRequestReplaysRunNothingTwice(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate) e.patchAs(patchOne, pullRkey, sha, testBob) e.ticketAs(pullRkey, "closed", testOwner, "update") expectPRPipelines(t, e.n, 2) e.sink.expect(t, 2)
e.emitTicket(testRepoDid, pullRkey, opCreate) e.emitPatch(testRepoDid, patchOne, opCreate) e.n.consumed(e.emitTicket(testRepoDid, pullRkey, "update")) expectPRPipelines(t, e.n, 2) e.sink.expect(t, 2)}
func TestPullTriggerCarriesTheActorAndOnlyATargetRepoBranch(t *testing.T) { repo := &workflow.TriggerRepo{Did: testOwner.String()} for _, c := range []struct { name string head db.PullHead branch string }{ {"an old patch", db.PullHead{Sha: patchSha2}, ""}, {"a branch of the repo", db.PullHead{Sha: patchSha2, SourceRepo: testRepoDid.String(), SourceBranch: "feature"}, "feature"}, {"a fork's branch", db.PullHead{Sha: patchSha2, SourceRepo: faultOtherRepoDid.String(), SourceBranch: "feature"}, ""}, } { ev := db.PullEvent{RepoDid: testRepoDid, Ticket: pullRkey, Event: models.WebhookEventPullRequestMerged, Author: testBob, Sender: testOwner, Head: c.head} got := pullTrigger(repo, ev, "main") pr := got.PullRequest if got.Kind != "pull_request" || got.Repo != repo || got.Actor == nil || *got.Actor != testOwner.String() { t.Errorf("%s: trigger kind %q repo %v actor %v, want the merger %s as actor", c.name, got.Kind, got.Repo, got.Actor, testOwner) } if pr == nil || pr.Action == nil || *pr.Action != "merged" || pr.SourceSha != patchSha2 || pr.TargetBranch != "main" || pr.SourceBranch != c.branch { t.Errorf("%s: pull request trigger %+v, want source branch %q", c.name, pr, c.branch) } }}
func TestPullRequestSourceBranchComesFromThePatch(t *testing.T) { e := newPRPipelineEnv(t) sha := e.head(t) e.ticketAs(pullRkey, "open", testBob, opCreate) record := e.putPatch(testRepoDid, patchOne, pullRkey, sha, testBob) record["source"] = map[string]any{"repo": testRepoDid.String(), "branch": "feature"} e.knot.putVersion(testRepoDid, org_tangled.ReviewPatchNSID, patchOne, testBob, record) e.n.consumed(e.emitPatch(testRepoDid, patchOne, opCreate))
got := expectPRPipelines(t, e.n, 1) if b := got[0].Trigger.PullRequest.SourceBranch; b == nil || *b != "feature" { t.Errorf("pipeline source branch %v, want feature", b) } p := pullPayload(t, e.sink.expect(t, 1)[0]) if p.PullRequest.Source == nil || p.PullRequest.Source.Branch != "feature" || p.PullRequest.Source.Repo != testRepoDid.String() || p.PullRequest.Source.Sha != sha { t.Errorf("webhook source %+v, want feature of %s at %s", p.PullRequest.Source, testRepoDid, sha) }}
func TestRefusedByKnotOnlyCountsAnswersARetryWouldGetAgain(t *testing.T) { for stderr, want := range map[string]bool{ "error: RPC failed; HTTP 400 curl 22 The requested URL returned error: 400\nfatal: expected 'packfile'": true, "fatal: unable to access 'https://knot.test/did:plc:repo/': The requested URL returned error: 404": true, "fatal: unable to access 'https://knot.test/did:plc:repo/': The requested URL returned error: 429": false, "error: RPC failed; HTTP 502 curl 22 The requested URL returned error: 502": false, "fatal: unable to access 'https://knot.test/did:plc:repo/': Could not resolve host: knot.test": false, "fatal: remote error: upload-pack: not our ref 0123456789012345678901234567890123456789": true, "error: Server does not allow request for unadvertised object 0123456789012345678901234567890123456789": true, "remote: no git\nfatal: repository 'http://127.0.0.1:42617/did:plc:repo/' not found": true, "fatal: unable to access 'https://knot.test/did:plc:repo/': Could not resolve host: not found": false, } { if got := refusedByKnot(errors.New("git fetch: exit status 128: " + stderr)); got != want { t.Errorf("%q: refused = %v, want %v", stderr, got, want) } }}