package 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.applied(testRepoDid, 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.applied(testRepoDid, 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 follows
func 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 neither
func 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 later
func 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 it
func (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.applied(faultOtherRepoDid, 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)
}
}
}