package spindle
import (
"context"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"net/http/httptest"
"path/filepath"
"slices"
"strings"
"testing"
"time"
"tangled.org/core/api/org_tangled"
"tangled.org/core/gitutil"
"tangled.org/core/knotfeed"
"tangled.org/core/spindle/models"
pipelinecodec "tangled.org/core/spindle/pipeline"
"tangled.org/core/spindle/webhook"
"tangled.org/core/spindle/xrpc"
"tangled.org/core/workflow"
)
const (
testHexOld = "1111111111111111111111111111111111111111"
testHexNew = "2222222222222222222222222222222222222222"
)
func mustObjectID(hex string) knotfeed.ObjectID {
sha, ok := knotfeed.ParseObjectID(hex)
if !ok {
panic("test fixture " + hex + " is not an object id")
}
return sha
}
func pushEvent(ref string, options ...string) *org_tangled.EventNotifyEvent_Input_Event {
actor := testBob.String()
return &org_tangled.EventNotifyEvent_Input_Event{EventPush: &org_tangled.EventPush{
Repo: testRepoDid.String(),
Ref: ref,
OldSha: testHexOld,
NewSha: testHexNew,
Actor: &actor,
PushOptions: options,
}}
}
type hookEnv struct {
*ensureEnv
pushes chan string
}
func newHookEnv(t *testing.T) *hookEnv {
t.Helper()
env := newEnsureEnv(t)
env.s.wh = webhook.New(env.s.db, true, pauseGate(env.s.db, &env.s.cfg.Server))
env.serve("repo")
if env.ensure(t) == nil {
t.Fatal("fixture repo not served")
}
pushes := make(chan string, 4)
receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
pushes <- string(body)
}))
t.Cleanup(receiver.Close)
hook := &models.Webhook{RepoDid: testRepoDid, Url: receiver.URL, Active: true, Events: []string{"push"}}
if created, err := env.s.db.AddWebhook(context.Background(), hook, 0, 0); err != nil || !created {
t.Fatalf("add webhook: created=%v err=%v", created, err)
}
return &hookEnv{ensureEnv: env, pushes: pushes}
}
func (h *hookEnv) notifyErr(event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) {
return h.s.NotifyEvent(context.Background(), h.k, testRepoDid, event)
}
func (h *hookEnv) notify(t *testing.T, event *org_tangled.EventNotifyEvent_Input_Event) string {
t.Helper()
pipeline, err := h.notifyErr(event)
if err != nil {
t.Fatalf("NotifyEvent: %v", err)
}
return string(pipeline)
}
func TestNotifyEventSkipCIFiresWebhookButSchedulesNothing(t *testing.T) {
h := newHookEnv(t)
if pipeline := h.notify(t, pushEvent("refs/heads/main", "skip-ci")); pipeline != "" {
t.Fatalf("pipeline = %q, want none", pipeline)
}
select {
case body := <-h.pushes:
if !strings.Contains(body, testHexNew) {
t.Errorf("webhook body %s lacks the new sha", body)
}
case <-time.After(5 * time.Second):
t.Fatal("skip-ci push never reached the webhook")
}
}
func TestNotifyEventIgnoresRefsThatRunNothing(t *testing.T) {
h := newHookEnv(t)
deleted := pushEvent("refs/heads/main")
deleted.EventPush.NewSha = strings.Repeat("0", 40)
for name, event := range map[string]*org_tangled.EventNotifyEvent_Input_Event{
"ref outside heads and tags": pushEvent("refs/notes/wip"),
"deleted ref": deleted,
} {
if pipeline := h.notify(t, event); pipeline != "" {
t.Errorf("%s: pipeline = %q, want none", name, pipeline)
}
}
h.expectNoPush(t, "webhook fired for an ignored event")
}
func (h *hookEnv) expectPush(t *testing.T) {
t.Helper()
select {
case <-h.pushes:
case <-time.After(5 * time.Second):
t.Fatal("push never reached the webhook")
}
}
func (h *hookEnv) seedPushPipeline(t *testing.T, id, ref, oldSha, newSha string) {
t.Helper()
repoDid := testRepoDid.String()
actor := testBob.String()
raw := workflow.PipelineSpec{
TriggerMetadata: &workflow.TriggerMetadata{
Kind: string(workflow.TriggerKindPush),
Repo: &workflow.TriggerRepo{Knot: h.k.Host, RepoDid: &repoDid, Did: testOwner.String()},
Push: &workflow.PushTriggerData{Ref: ref, OldSha: oldSha, NewSha: newSha},
Actor: &actor,
},
Workflows: []*workflow.WorkflowSpec{{Name: "ci.yml"}},
}
record, err := pipelinecodec.FromTangled(models.PipelineId(id), time.Now(), raw)
if err != nil {
t.Fatal(err)
}
if err := h.s.db.CreatePipeline(record); err != nil {
t.Fatal(err)
}
if _, err := h.s.db.RecordPush(testRepoDid, ref, oldSha, newSha, time.Now()); err != nil {
t.Fatal(err)
}
}
func (h *hookEnv) expectNoPush(t *testing.T, why string) {
t.Helper()
select {
case body := <-h.pushes:
t.Fatalf("%s: %s", why, body)
case <-time.After(300 * time.Millisecond):
}
}
type notifyResult struct {
pipeline models.PipelineId
err error
}
func (h *hookEnv) notifyAsync(event *org_tangled.EventNotifyEvent_Input_Event) <-chan notifyResult {
out := make(chan notifyResult, 1)
go func() {
pipeline, err := h.notifyErr(event)
out <- notifyResult{pipeline, err}
}()
return out
}
func TestNotifyEventAnswersOnlyOnceThePushIsHandled(t *testing.T) {
h := newHookEnv(t)
unlock := h.s.lockRepo(testRepoDid).Unlock
answer := h.notifyAsync(pushEvent("refs/heads/main", "skip-ci"))
select {
case res := <-answer:
t.Fatalf("answered %+v while the push was still waiting on its work", res)
case <-time.After(2500 * time.Millisecond):
}
h.expectNoPush(t, "work ran while the repo was locked")
unlock()
if res := <-answer; res.err != nil {
t.Fatal(res.err)
}
h.expectPush(t)
}
func TestNotifyEventReplayOfAPipelinePushAnswersWithItAndRunsNothing(t *testing.T) {
h := newHookEnv(t)
h.seedPushPipeline(t, "first", "refs/heads/main", testHexOld, testHexNew)
// skip-ci makes no new pipeline, the id can only come from the replay lookup
for range 2 {
got := h.notify(t, pushEvent("refs/heads/main", "skip-ci"))
if got != "first" {
t.Fatalf("replay answered %q, want the first delivery's pipeline id", got)
}
if _, err := h.s.db.GetPipeline(context.Background(), models.PipelineId(got)); err != nil {
t.Fatalf("answer %q is not a stored pipeline id: %v", got, err)
}
}
h.expectNoPush(t, "a replayed push fired a webhook")
}
func TestNotifyEventDedupeKeyIsRepoRefOldAndNew(t *testing.T) {
h := newHookEnv(t)
h.seedPushPipeline(t, "first", "refs/heads/main", testHexOld, testHexNew)
other := strings.Repeat("3", 40)
for name, mutate := range map[string]func(*org_tangled.EventPush){
"another ref": func(p *org_tangled.EventPush) { p.Ref = "refs/heads/dev" },
"another old sha": func(p *org_tangled.EventPush) { p.OldSha = other },
"another new sha": func(p *org_tangled.EventPush) { p.NewSha = other },
// old and new are ordered, so A to B then B to A is two pushes
"old and new swapped": func(p *org_tangled.EventPush) { p.OldSha, p.NewSha = p.NewSha, p.OldSha },
} {
event := pushEvent("refs/heads/main", "skip-ci")
mutate(event.EventPush)
if got := h.notify(t, event); got != "" {
t.Errorf("%s: answered %q, want the push treated as new", name, got)
}
h.expectPush(t)
}
}
func TestNotifyEventReplayWithoutAPipelineFiresTheWebhookOnce(t *testing.T) {
h := newHookEnv(t)
event := pushEvent("refs/heads/main", "skip-ci")
h.notify(t, event)
h.expectPush(t)
h.notify(t, event)
h.expectNoPush(t, "a replay of a push that made no pipeline fired the webhook again")
}
func TestNotifyEventDuplicateWhileRunningJoinsTheWork(t *testing.T) {
h := newHookEnv(t)
event := pushEvent("refs/heads/main", "skip-ci")
unlock := h.s.lockRepo(testRepoDid).Unlock
first := h.notifyAsync(event)
time.Sleep(100 * time.Millisecond)
second := h.notifyAsync(event)
select {
case res := <-second:
t.Fatalf("duplicate answered %+v before the first finished", res)
case <-time.After(300 * time.Millisecond):
}
unlock()
a, b := <-first, <-second
if a.err != nil || b.err != nil || a.pipeline != b.pipeline {
t.Fatalf("answers = %+v and %+v, want the same one twice", a, b)
}
h.expectPush(t)
h.expectNoPush(t, "the duplicate ran its own work")
}
func TestNotifyEventWorkSurvivesTheFirstCallerGivingUp(t *testing.T) {
h := newHookEnv(t)
event := pushEvent("refs/heads/main", "skip-ci")
unlock := h.s.lockRepo(testRepoDid).Unlock
ctx, hangUp := context.WithCancel(context.Background())
first := make(chan error, 1)
go func() {
_, err := h.s.NotifyEvent(ctx, h.k, testRepoDid, event)
first <- err
}()
time.Sleep(100 * time.Millisecond)
second := h.notifyAsync(event)
time.Sleep(100 * time.Millisecond)
hangUp()
if err := <-first; !errors.Is(err, context.Canceled) {
t.Fatalf("first caller err = %v, want its own cancellation", err)
}
unlock()
if res := <-second; res.err != nil {
t.Fatalf("second caller inherited the first one's hangup: %v", res.err)
}
h.expectPush(t)
h.expectNoPush(t, "the push ran twice")
}
func TestNotifyEventAnswersAnErrorWhenTheWorkFails(t *testing.T) {
h := newHookEnv(t)
h.knot.failing = true
h.s.db.DeleteReposByDid(testRepoDid)
event := pushEvent("refs/heads/main", "skip-ci")
// an error here is what makes the knot retry
if _, err := h.notifyErr(event); err == nil {
t.Fatal("answered success for a push that was not handled")
}
h.expectNoPush(t, "a push that failed still fired")
h.knot.failing = false
h.notify(t, event)
h.expectPush(t)
}
func TestNotifyEventRefusesPastThePool(t *testing.T) {
h := newHookEnv(t)
h.s.cfg.Server.NotifySlots = 3
others := []string{"one.test", "two.test", "three.test"}
for _, host := range others {
if !h.s.notifySlots.acquire(host, h.s.notifyLimits()) {
t.Fatal("slot refused below the pool")
}
}
_, err := h.notifyErr(pushEvent("refs/heads/main", "skip-ci"))
if !errors.Is(err, xrpc.ErrNotifyBusy) {
t.Fatalf("err = %v, want ErrNotifyBusy", err)
}
for _, host := range others {
h.s.notifySlots.release(host)
}
h.notify(t, pushEvent("refs/heads/main", "skip-ci"))
h.expectPush(t)
}
func TestNotifyEventRejectsMalformedShas(t *testing.T) {
h := newHookEnv(t)
event := pushEvent("refs/heads/main")
event.EventPush.NewSha = "not-a-sha"
if _, err := h.notifyErr(event); err == nil {
t.Fatal("malformed sha accepted")
}
}
func TestNotifyEventIgnoresUnservedRepos(t *testing.T) {
env := newEnsureEnv(t)
env.s.wh = webhook.New(env.s.db, true)
env.manifest("repo", "did:web:other.example.com")
env.knot.declare(testOwner, "repo", testRepoDid)
pipeline, err := env.s.NotifyEvent(context.Background(), env.k, testRepoDid, pushEvent("refs/heads/main"))
if err != nil || pipeline != "" {
t.Fatalf("got pipeline %q, err %v, want an empty answer", pipeline, err)
}
if rows := env.rows(t); len(rows) != 0 {
t.Errorf("unserved repo was recorded: %+v", rows)
}
}
func TestPushTriggerNamesTheActor(t *testing.T) {
event := pushEvent("refs/heads/main").EventPush
trigger := pushTrigger(nil, event, mustObjectID(testHexOld), mustObjectID(testHexNew))
if trigger.Actor == nil || *trigger.Actor != testBob.String() {
t.Errorf("actor = %v, want %s", trigger.Actor, testBob)
}
if trigger.Kind != string(workflow.TriggerKindPush) || trigger.Push.Ref != "refs/heads/main" || trigger.Push.OldSha != testHexOld || trigger.Push.NewSha != testHexNew {
t.Errorf("trigger = %+v", trigger)
}
event.Actor = nil
if trigger := pushTrigger(nil, event, mustObjectID(testHexOld), mustObjectID(testHexNew)); trigger.Actor != nil {
t.Errorf("actor = %v, want nil when the knot sent none", *trigger.Actor)
}
}
func repeatedPaths(n int, path string) []string {
paths := make([]string, n)
for i := range paths {
paths[i] = path
}
return paths
}
func TestAdmitChangedFilesUnderBudget(t *testing.T) {
if got, ok := admitChangedFiles([]string{"a.go", "b/c.go"}); !ok || !slices.Equal(got, []string{"a.go", "b/c.go"}) {
t.Errorf("small listing = %v, %v, want both paths", got, ok)
}
if got, ok := admitChangedFiles(repeatedPaths(11000, "src/f.go")); ok || got != nil {
t.Errorf("listing over the entry cap = %v, %v, want nothing admitted", got, ok)
}
wide := repeatedPaths(8192, "src/f.go")
if got, ok := admitChangedFiles(wide); !ok || !slices.Equal(got, wide) {
t.Errorf("listing at the entry cap = %d paths, %v, want all %d", len(got), ok, len(wide))
}
if got, ok := admitChangedFiles(nil); !ok || got == nil || len(got) != 0 {
t.Errorf("an empty listing = %v, %v, want an admitted list that is empty and not nil", got, ok)
}
big := repeatedPaths(11000, strings.Repeat("x", 64))
if got, ok := admitChangedFiles(big); ok || got != nil {
t.Errorf("listing over the byte budget = %d paths, %v, want nothing admitted", len(got), ok)
}
}
func TestChangedFilesUnderBudgetGuardsFirstSight(t *testing.T) {
quiet := slog.New(slog.NewTextHandler(io.Discard, nil))
if got, known := changedFilesUnderBudget(context.Background(), quiet, "/nonexistent/repo", knotfeed.ObjectID{}, mustObjectID(testHexNew)); known || got != nil {
t.Errorf("first sight of a ref = %v, %v, want unknown without touching the repo", got, known)
}
}
func TestChangedFilesUnderBudgetWithRealRepo(t *testing.T) {
for _, format := range []string{"sha1", "sha256"} {
t.Run(format, func(t *testing.T) { changedFilesInRealRepo(t, format) })
}
}
func commitHead(t *testing.T, repoPath string) knotfeed.ObjectID {
t.Helper()
commitEverything(t, repoPath)
return mustObjectID(gitInDir(t, repoPath, "rev-parse", "HEAD"))
}
func changedFilesInRealRepo(t *testing.T, format string) {
repoPath := newSourceRepo(t, format)
commit := func() knotfeed.ObjectID { return commitHead(t, repoPath) }
writeTestFile(t, filepath.Join(repoPath, "README.md"), "one\n")
shaFirst := commit()
writeTestFile(t, filepath.Join(repoPath, ".tangled/workflows/ci.yaml"), "when:\n event: push\n")
writeTestFile(t, filepath.Join(repoPath, "src/app.go"), "package main\n")
shaSecond := commit()
gitInDir(t, repoPath, "mv", "src/app.go", "src/main.go")
shaThird := commit()
quiet := slog.New(slog.NewTextHandler(io.Discard, nil))
synced := filepath.Join(t.TempDir(), "synced")
if err := gitutil.SparseSync(context.Background(), repoPath, synced, shaSecond.String(), "/.tangled/workflows"); err != nil {
t.Fatalf("sync: %v", err)
}
got, known := changedFilesUnderBudget(context.Background(), quiet, synced, shaFirst, shaSecond)
if !known {
t.Error("shallow checkout changed files are unknown")
}
slices.Sort(got)
if want := []string{".tangled/workflows/ci.yaml", "src/app.go"}; !slices.Equal(got, want) {
t.Errorf("shallow checkout changed files = %v, want %v", got, want)
}
for _, tc := range []struct {
name string
oldSha knotfeed.ObjectID
newSha knotfeed.ObjectID
want []string
}{
{"a push commit lists its added and modified files", shaFirst, shaSecond, []string{".tangled/workflows/ci.yaml", "src/app.go"}},
{"a rename lists both the old and new path", shaSecond, shaThird, []string{"src/app.go", "src/main.go"}},
{"a no-op revision range lists nothing", shaThird, shaThird, nil},
} {
t.Run(tc.name, func(t *testing.T) {
got, known := changedFilesUnderBudget(context.Background(), quiet, repoPath, tc.oldSha, tc.newSha)
slices.Sort(got)
if !slices.Equal(got, tc.want) {
t.Errorf("changed files = %v, want %v", got, tc.want)
}
// a nil list means unknown, and an empty change set still has to come back non-nil
if !known || got == nil {
t.Errorf("known = %v, list nil = %v, want a known list that is not nil", known, got == nil)
}
})
}
}
func TestNotifyEventDoesNotLetGitNearAnInternalKnot(t *testing.T) {
h := newHookEnv(t)
h.s.cfg.Server.Dev = false
before := h.knot.count()
_, err := h.notifyErr(pushEvent("refs/heads/main"))
if err == nil || err.Error() != "couldn't fetch the push from the knot" {
t.Fatalf("err = %v, want the generic fetch error", err)
}
if got := h.knot.count(); got != before {
t.Errorf("the knot saw %d requests, git should not have dialed a loopback host", got-before)
}
}
func TestPathsGatedWorkflowsRunWhenTheChangedFilesCantBeTold(t *testing.T) {
gated := workflow.Workflow{
Name: ".tangled/workflows/ci.yml",
Engine: "nixery",
When: []workflow.Constraint{{Event: []string{"push"}, Branch: []string{"main"}, Paths: []string{"src/**"}}},
}
compile := func(changed []string) int {
c := workflow.Compiler{
Trigger: workflow.TriggerMetadata{
Kind: string(workflow.TriggerKindPush),
Push: &workflow.PushTriggerData{Ref: "refs/heads/main"},
},
ChangedFiles: changed,
}
return len(c.Compile([]workflow.Workflow{gated}).Workflows)
}
small, ok := admitChangedFiles([]string{"docs/guide.md"})
if !ok || compile(small) != 0 {
t.Errorf("a push that only touched docs ran the src workflow (known %v)", ok)
}
over := repeatedPaths(changedFilesMaxCount+1, "docs/f.md")
tooMany, ok := admitChangedFiles(over)
if ok || compile(tooMany) != 1 {
t.Errorf("a push with %d changed files didn't run the paths-gated workflow (known %v)", len(over), ok)
}
}
func TestOverBudgetPushIsLoggedWithItsFileCount(t *testing.T) {
repoPath := newSourceRepo(t, "sha1")
writeTestFile(t, filepath.Join(repoPath, "README.md"), "one\n")
before := commitHead(t, repoPath)
for i := range changedFilesMaxCount + 1 {
writeTestFile(t, filepath.Join(repoPath, "gen", fmt.Sprintf("f%d.txt", i)), "x\n")
}
after := commitHead(t, repoPath)
var logged strings.Builder
l := slog.New(slog.NewTextHandler(&logged, nil))
got, known := changedFilesUnderBudget(context.Background(), l, repoPath, before, after)
if known || got != nil {
t.Fatalf("changed files = %d, known %v, want unknown", len(got), known)
}
for _, want := range []string{before.String(), after.String(), fmt.Sprintf("files=%d", changedFilesMaxCount+1)} {
if !strings.Contains(logged.String(), want) {
t.Errorf("log %q lacks %q", logged.String(), want)
}
}
}