Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416package spindle
import ( "context" "database/sql" "encoding/json" "errors" "io" "net/http" "net/http/httptest" "sync" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled" "tangled.org/core/spindle/models")
const ( pullRkey = "3kpull0000001" patchOne = "bafypatchone" patchTwo = "bafypatchtwo" patchSha1 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" patchSha2 = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" settleTime = 300 * time.Millisecond)
type delivery struct { event string signature string body []byte}
// stands in for the user's endpointtype sink struct { srv *httptest.Server
mu sync.Mutex got []delivery}
func newSink(t *testing.T) *sink { t.Helper() s := &sink{} s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { body, _ := io.ReadAll(r.Body) s.mu.Lock() s.got = append(s.got, delivery{event: r.Header.Get("X-Tangled-Event"), signature: r.Header.Get("X-Tangled-Signature-256"), body: body}) s.mu.Unlock() })) t.Cleanup(s.srv.Close) return s}
func (s *sink) all() []delivery { s.mu.Lock() defer s.mu.Unlock() return append([]delivery(nil), s.got...)}
// deliveries are async, so wait for the count and then see that nothing else followsfunc (s *sink) expect(t *testing.T, want int) []delivery { t.Helper() if want > 0 { eventually(t, 10*time.Second, "the webhook deliveries", func() bool { return len(s.all()) >= want }) } time.Sleep(settleTime) got := s.all() if len(got) != want { t.Fatalf("got %d webhook deliveries, want %d: %s", len(got), want, describeDeliveries(got)) } return got}
func describeDeliveries(got []delivery) string { out := "" for _, d := range got { out += "\n" + d.event + " " + string(d.body) } return out}
func (w *world) subscribe(n *node, s *sink, events ...string) { t := w.t t.Helper() hook := &models.Webhook{RepoDid: testRepoDid, Url: s.srv.URL, Secret: "shh", Active: true, Events: events} if created, err := n.s.db.AddWebhook(context.Background(), hook, 0, 0); err != nil || !created { t.Fatalf("add webhook: created=%v err=%v", created, err) }}
var pullEvents = []string{ string(models.WebhookEventPullRequestCreated), string(models.WebhookEventPullRequestResubmitted), string(models.WebhookEventPullRequestClosed), string(models.WebhookEventPullRequestMerged), string(models.WebhookEventPullRequestReopened),}
func (w *world) putTicket(repo syntax.DID, rkey, state string, editor syntax.DID, branch bool) map[string]any { record := map[string]any{ "$type": org_tangled.TrackTicketNSID, "state": state, "title": map[string]any{"$type": "org.tangled.markup.markdown#inline", "text": "fix the thing"}, "body": map[string]any{"$type": "org.tangled.markup.markdown", "text": "it was broken"}, "createdAt": "2026-02-01T00:00:00Z", } if branch { record["targetBranch"] = "main" } w.knot.putVersion(repo, org_tangled.TrackTicketNSID, rkey, editor, record) return record}
func (w *world) putPatch(repo syntax.DID, rkey, ticket, head string, editor syntax.DID) map[string]any { record := map[string]any{ "$type": org_tangled.ReviewPatchNSID, "base": map[string]any{"oid": patchSha1}, "head": map[string]any{"oid": head}, "ticket": ticket, "createdAt": "2026-02-01T00:00:00Z", } w.knot.putVersion(repo, org_tangled.ReviewPatchNSID, rkey, editor, record) return record}
// the knot already holds the record, the event only says it changedfunc (w *world) emitTicket(repo syntax.DID, rkey, op string) int64 { return w.js.commit(repo, org_tangled.TrackTicketNSID, rkey, op, map[string]any{"$type": org_tangled.TrackTicketNSID})}
func (w *world) emitPatch(repo syntax.DID, rkey, op string) int64 { return w.js.commit(repo, org_tangled.ReviewPatchNSID, rkey, op, map[string]any{"$type": org_tangled.ReviewPatchNSID})}
type pullEnv struct { *world n *node sink *sink}
func newPullEnv(t *testing.T) *pullEnv { t.Helper() w := newWorld(t) n := w.boot() w.seed(n, testBob) e := &pullEnv{world: w, n: n, sink: newSink(t)} w.subscribe(n, e.sink, pullEvents...) return e}
func (e *pullEnv) ticket(state string, editor syntax.DID, op string) { e.putTicket(testRepoDid, pullRkey, state, editor, true) e.n.applied(testRepoDid, e.emitTicket(testRepoDid, pullRkey, op))}
func (e *pullEnv) patch(rkey, head string, editor syntax.DID) { e.putPatch(testRepoDid, rkey, pullRkey, head, editor) e.n.applied(testRepoDid, e.emitPatch(testRepoDid, rkey, opCreate))}
func pullPayload(t *testing.T, d delivery) models.WebhookPullRequestPayload { t.Helper() var p models.WebhookPullRequestPayload if err := json.Unmarshal(d.body, &p); err != nil { t.Fatalf("payload %s: %v", d.body, err) } return p}
func wantPull(t *testing.T, d delivery, event models.WebhookEvent, action, state string, round int, author, sender syntax.DID, head string) { t.Helper() p := pullPayload(t, d) if d.event != string(event) || p.Action != action || p.PullRequest.State != state { t.Errorf("delivered %q action %q state %q, want %q %q %q", d.event, p.Action, p.PullRequest.State, event, action, state) } if p.PullRequest.RoundNumber != round { t.Errorf("%s: round %d, want %d", action, p.PullRequest.RoundNumber, round) } if p.PullRequest.Owner.Did != author.String() { t.Errorf("%s: pull author %s, want %s", action, p.PullRequest.Owner.Did, author) } if p.Sender.Did != sender.String() { t.Errorf("%s: sender %s, want %s", action, p.Sender.Did, sender) } if p.PullRequest.Title != "fix the thing" || p.PullRequest.Body != "it was broken" || p.PullRequest.TargetBranch != "main" { t.Errorf("%s: pull request %+v lost its record fields", action, p.PullRequest) } if p.PullRequest.Source == nil || p.PullRequest.Source.Sha != head || p.PullRequest.Source.Branch != "" { t.Errorf("%s: source %+v, want only the sha %s", action, p.PullRequest.Source, head) } if p.Repository.FullName != testOwner.String()+"/repo" { t.Errorf("%s: repository %q", action, p.Repository.FullName) } if d.signature == "" { t.Errorf("%s: delivery isn't signed", action) }}
func TestPullRequestWebhooksFireOncePerTransition(t *testing.T) { e := newPullEnv(t)
// the ticket alone is no pull request event, the first patch is e.ticket("open", testBob, opCreate) e.sink.expect(t, 0)
e.patch(patchOne, patchSha1, testBob) got := e.sink.expect(t, 1) wantPull(t, got[0], models.WebhookEventPullRequestCreated, "created", "open", 1, testBob, testBob, patchSha1)
e.patch(patchTwo, patchSha2, testBob) got = e.sink.expect(t, 2) wantPull(t, got[1], models.WebhookEventPullRequestResubmitted, "resubmitted", "open", 2, testBob, testBob, patchSha2)
// the maintainer closes it, so sender and author differ e.ticket("closed", testCarol, "update") got = e.sink.expect(t, 3) wantPull(t, got[2], models.WebhookEventPullRequestClosed, "closed", "closed", 2, testBob, testCarol, patchSha2)
e.ticket("open", testBob, "update") got = e.sink.expect(t, 4) wantPull(t, got[3], models.WebhookEventPullRequestReopened, "reopened", "open", 2, testBob, testBob, patchSha2)
e.ticket("completed", testCarol, "update") got = e.sink.expect(t, 5) wantPull(t, got[4], models.WebhookEventPullRequestMerged, "merged", "merged", 2, testBob, testCarol, patchSha2)}
func TestPullRequestReplaysFireNothing(t *testing.T) { e := newPullEnv(t) e.ticket("open", testBob, opCreate) e.patch(patchOne, patchSha1, testBob) e.patch(patchTwo, patchSha2, testBob) e.ticket("completed", testCarol, "update") e.sink.expect(t, 3)
// a stale event reads the knot's current state, which spindle already has e.emitTicket(testRepoDid, pullRkey, opCreate) e.emitTicket(testRepoDid, pullRkey, "update") e.emitPatch(testRepoDid, patchOne, opCreate) e.n.consumed(e.emitPatch(testRepoDid, patchTwo, opCreate)) e.sink.expect(t, 3)}
func TestPullRequestPatchBeforeItsTicketEvent(t *testing.T) { e := newPullEnv(t) e.putTicket(testRepoDid, pullRkey, "open", testBob, true) e.putPatch(testRepoDid, patchOne, pullRkey, patchSha1, testBob)
e.emitPatch(testRepoDid, patchOne, opCreate) e.n.consumed(e.emitTicket(testRepoDid, pullRkey, opCreate))
got := e.sink.expect(t, 1) wantPull(t, got[0], models.WebhookEventPullRequestCreated, "created", "open", 1, testBob, testBob, patchSha1)}
func TestPullRequestTransitionSurvivesARestart(t *testing.T) { e := newPullEnv(t) e.ticket("open", testBob, opCreate) e.patch(patchOne, patchSha1, testBob) e.sink.expect(t, 1)
e.pushCursorPast(e.n) e.putTicket(testRepoDid, pullRkey, "closed", testCarol, true) e.emitTicket(testRepoDid, pullRkey, "update") n2 := e.boot() eventually(t, 15*time.Second, "the close to be seen after the restart", func() bool { ticket, err := n2.s.db.GetPullTicket(testRepoDid, pullRkey) return err == nil && ticket.State == "closed" }) got := e.sink.expect(t, 2) wantPull(t, got[1], models.WebhookEventPullRequestClosed, "closed", "closed", 1, testBob, testCarol, patchSha1)}
func TestPullRequestIgnoresIssuesAndForeignRepos(t *testing.T) { e := newPullEnv(t)
e.putTicket(testRepoDid, "3kissue0000001", "open", testBob, false) e.putPatch(testRepoDid, "bafyissuepatch", "3kissue0000001", patchSha1, testBob) e.emitTicket(testRepoDid, "3kissue0000001", opCreate) e.n.consumed(e.emitPatch(testRepoDid, "bafyissuepatch", opCreate))
e.putTicket(faultOtherRepoDid, pullRkey, "open", testBob, true) e.putPatch(faultOtherRepoDid, patchOne, pullRkey, patchSha1, testBob) e.emitTicket(faultOtherRepoDid, pullRkey, opCreate) e.n.consumed(e.emitPatch(faultOtherRepoDid, patchOne, opCreate))
e.sink.expect(t, 0) if _, err := e.n.s.db.GetPullTicket(testRepoDid, "3kissue0000001"); !errors.Is(err, sql.ErrNoRows) { t.Errorf("an issue got a pull row: %v", err) } if _, err := e.n.s.db.GetPullTicket(faultOtherRepoDid, pullRkey); !errors.Is(err, sql.ErrNoRows) { t.Errorf("a repo this spindle doesn't serve got a pull row: %v", err) }}
func TestPullRequestEventsForRecordsTheKnotLacksFireNothing(t *testing.T) { e := newPullEnv(t)
// nothing was ever put on the knot, so these are forged or already gone e.emitTicket(testRepoDid, pullRkey, opCreate) e.emitTicket(testRepoDid, pullRkey, "update") e.n.consumed(e.emitPatch(testRepoDid, patchOne, opCreate)) e.sink.expect(t, 0)
// a patch whose ticket the knot lacks has no pull request to belong to e.putPatch(testRepoDid, patchTwo, pullRkey, patchSha1, testBob) e.n.consumed(e.emitPatch(testRepoDid, patchTwo, opCreate)) e.sink.expect(t, 0)}
func TestPullRequestDeletesDropTheRowsWithoutAWebhook(t *testing.T) { e := newPullEnv(t) e.ticket("open", testBob, opCreate) e.patch(patchOne, patchSha1, testBob) e.sink.expect(t, 1)
e.knot.erase(testRepoDid, org_tangled.ReviewPatchNSID, patchOne) e.emitPatch(testRepoDid, patchOne, opDelete) e.knot.erase(testRepoDid, org_tangled.TrackTicketNSID, pullRkey) e.n.consumed(e.emitTicket(testRepoDid, pullRkey, opDelete))
e.sink.expect(t, 1) if _, err := e.n.s.db.GetPullTicket(testRepoDid, pullRkey); !errors.Is(err, sql.ErrNoRows) { t.Errorf("deleted ticket still has a row: %v", err) } if rounds, _, err := e.n.s.db.PullRounds(testRepoDid, pullRkey); err != nil || rounds != 0 { t.Errorf("deleted ticket still has %d rounds, err %v", rounds, err) }}
func TestPushWebhookIsDelivered(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n) hook := newSink(t) w.subscribe(n, hook, string(models.WebhookEventPush))
// skip-ci keeps git out of it, the webhook fires either way if _, err := n.s.NotifyEvent(context.Background(), w.knotRef(), testRepoDid, pushEvent("refs/heads/main", "skip-ci")); err != nil { t.Fatalf("NotifyEvent: %v", err) } got := hook.expect(t, 1) if got[0].event != "push" || got[0].signature == "" { t.Errorf("delivered %q signed=%v, want a signed push", got[0].event, got[0].signature != "") } var p models.WebhookPayload if err := json.Unmarshal(got[0].body, &p); err != nil { t.Fatal(err) } if p.Ref != "refs/heads/main" || p.Before != testHexOld || p.After != testHexNew || p.Pusher.Did != testBob.String() { t.Errorf("push payload %+v", p) } if p.Repository.FullName != testOwner.String()+"/repo" || p.Repository.Owner.Did != testOwner.String() { t.Errorf("push repository %+v", p.Repository) }}
// web puts the manifest with the new name, then creates the new declaration and deletes the old onefunc TestRenameWebhookFiresOnceWhateverTheEventOrder(t *testing.T) { steps := map[string]func(w *world) int64{ "manifest": func(w *world) int64 { w.setManifest(testRepoDid, "Renamed", w.spindleDid()) return w.emitManifest(testRepoDid) }, "declaration created": func(w *world) int64 { w.declare("Renamed", testRepoDid) return w.emitDeclaration("Renamed", testRepoDid) }, "declaration deleted": func(w *world) int64 { w.undeclare("repo") return w.js.commit(testOwner, org_tangled.RepoDeclarationNSID, "repo", opDelete, nil) }, } orders := [][]string{ {"manifest", "declaration created", "declaration deleted"}, {"manifest", "declaration deleted", "declaration created"}, {"declaration created", "manifest", "declaration deleted"}, {"declaration created", "declaration deleted", "manifest"}, {"declaration deleted", "manifest", "declaration created"}, {"declaration deleted", "declaration created", "manifest"}, } for _, order := range orders { t.Run(order[0]+" then "+order[1]+" then "+order[2], func(t *testing.T) { w := newWorld(t) n := w.boot() w.seed(n, testBob) addTestSecret(t, n.s, testRepoDid) hook := newSink(t) w.subscribe(n, hook, string(models.WebhookEventRepoRenamed))
for _, step := range order { n.consumed(steps[step](w)) }
got := hook.expect(t, 1) var p models.WebhookRenamePayload if err := json.Unmarshal(got[0].body, &p); err != nil { t.Fatal(err) } if got[0].event != "repository:renamed" || p.OldName != "repo" || p.NewName != "Renamed" || p.Sender.Did != testOwner.String() { t.Errorf("delivered %q %+v", got[0].event, p) } rows := repoRows(t, n.s, testRepoDid) if len(rows) != 1 || rows[0].Name != "Renamed" || rows[0].Rkey != "renamed" { t.Errorf("repo rows after the rename: %s", describeRows(rows)) } if secretCount(t, n.s, testRepoDid) != 1 || !isCollaborator(t, n.s, testRepoDid, testBob) { t.Error("the rename wiped the repo's secrets or collaborators") } }) }}