Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411package sitefeed
import ( "bytes" "context" "errors" "path/filepath" "sync" "testing" "time"
"tangled.org/core/appview/cloudflare" "tangled.org/core/appview/config" "tangled.org/core/appview/db" "tangled.org/core/appview/models" "tangled.org/core/knotfeed" "tangled.org/core/log" "tangled.org/core/tapc"
cbg "github.com/whyrusleeping/cbor-gen")
const ( feedTestHost = "knot.invalid" feedTestOtherHost = "other.knot.invalid" feedTestRepoDid = "did:plc:limpet" feedTestOwner = "did:plc:akshay" feedTestSha = "bcfa6359b037bc411e9626260cc3dd9a28408d2c" feedTestOtherSha = "9eab6c4f29bc9e609ec52888cd6a4ef716b567c9")
var errBoom = errors.New("boom")
func feedTestDB(t *testing.T) *db.DB { t.Helper() d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "appview.db")) if err != nil { t.Fatalf("db.Make: %v", err) } t.Cleanup(func() { d.Close() }) return d}
func seedFeedRepo(t *testing.T, d *db.DB) { t.Helper() tx, err := d.Begin() if err != nil { t.Fatalf("begin: %v", err) } if err := db.AddRepo(tx, &models.Repo{ Did: feedTestOwner, Knot: feedTestHost, RepoDid: feedTestRepoDid, Name: "anemone", }); err != nil { t.Fatalf("AddRepo: %v", err) } if err := tx.Commit(); err != nil { t.Fatalf("commit: %v", err) }}
func setSiteConfig(t *testing.T, d *db.DB) { t.Helper() if err := db.SetRepoSiteConfig(d, feedTestRepoDid, "main", "/", false); err != nil { t.Fatalf("SetRepoSiteConfig: %v", err) }}
func enabledCf(t *testing.T) *cloudflare.Client { t.Helper() cf, err := cloudflare.New(&config.Config{Cloudflare: config.Cloudflare{ AccountId: "acct", KV: config.KVConfig{NamespaceId: "ns"}, R2: config.R2Config{Bucket: "bucket"}, }}) if err != nil { t.Fatalf("cloudflare.New: %v", err) } return cf}
type deployCall struct { repoDid string branch string dir string}
type deployRecorder struct { mu sync.Mutex calls []deployCall}
func (r *deployRecorder) hook(err error) func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error { return func(_ context.Context, _ *cloudflare.Client, _ *config.Config, repo *models.Repo, branch, dir string) error { r.mu.Lock() r.calls = append(r.calls, deployCall{repoDid: repo.RepoDid, branch: branch, dir: dir}) r.mu.Unlock() return err }}
func (r *deployRecorder) got() []deployCall { r.mu.Lock() defer r.mu.Unlock() return append([]deployCall(nil), r.calls...)}
func writeText(w *bytes.Buffer, s string) { if err := cbg.CborWriteHeader(w, cbg.MajTextString, uint64(len(s))); err != nil { w.Reset() } w.WriteString(s)}
func encodeRefRecord(t *testing.T, sha string) []byte { t.Helper() var out bytes.Buffer if err := cbg.CborWriteHeader(&out, cbg.MajMap, 1); err != nil { t.Fatalf("map header: %v", err) } writeText(&out, "sha") writeText(&out, sha) return out.Bytes()}
func refOpFor(t *testing.T, refname string, action tapc.RecordAction, sha string) knotfeed.RecordOp { t.Helper() rkey, _ := knotfeed.EscapeRefname(refname) op := knotfeed.RecordOp{Action: action, Collection: knotfeed.GitRefCollection, Rkey: rkey} if sha != "" { op.Bytes = encodeRefRecord(t, sha) } return op}
func object(t *testing.T, hex string) knotfeed.ObjectID { t.Helper() sha, ok := knotfeed.ParseObjectID(hex) if !ok { t.Fatalf("ParseObjectID(%q)", hex) } return sha}
func pollFor(t *testing.T, what string, ready func() bool) { t.Helper() deadline := time.Now().Add(2 * time.Second) for time.Now().Before(deadline) { if ready() { return } time.Sleep(5 * time.Millisecond) } t.Fatalf("%s never appeared", what)}
func feedFor(t *testing.T, d *db.DB, hook func(context.Context, *cloudflare.Client, *config.Config, *models.Repo, string, string) error) *Feed { t.Helper() f := New(d, nil, &config.Config{}, enabledCf(t), log.New("test")) f.deploy = hook t.Cleanup(func() { f.Unsubscribe(feedTestHost) f.Unsubscribe(feedTestOtherHost) }) return f}
func TestFeedRefOp_BranchMatchTriggersDeploy(t *testing.T) { d := feedTestDB(t) seedFeedRepo(t, d) setSiteConfig(t, d) r := &deployRecorder{} f := feedFor(t, d, r.hook(nil))
f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha)) pollFor(t, "the deploy call", func() bool { return len(r.got()) == 1 }) calls := r.got() if calls[0] != (deployCall{repoDid: feedTestRepoDid, branch: "main", dir: "/"}) { t.Fatalf("deploys = %+v, want one push on main", calls) }
var deploy *models.SiteDeploy pollFor(t, "the deploy row", func() bool { rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10) if err == nil && len(rows) > 0 { deploy = &rows[0] return true } return false }) if deploy.Status != models.SiteDeployStatusSuccess { t.Errorf("status = %q, want success", deploy.Status) } if deploy.Trigger != models.SiteDeployTriggerPush { t.Errorf("trigger = %q, want push", deploy.Trigger) } if deploy.CommitSHA != feedTestSha || deploy.Branch != "main" || deploy.Dir != "/" { t.Errorf("deploy row = %+v, want the pushed sha, branch and dir", deploy) }}
func TestFeedSkipsNonDeployingOps(t *testing.T) { seedSiteRepo := func(t *testing.T, d *db.DB) { seedFeedRepo(t, d) setSiteConfig(t, d) } for _, tc := range []struct { name string seed func(*testing.T, *db.DB) op func(*testing.T, *Feed) }{ { name: "push to a branch other than the site branch", seed: seedSiteRepo, op: func(_ *testing.T, f *Feed) { f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "other", object(t, feedTestSha)) }, }, { name: "push without a site config", seed: seedFeedRepo, op: func(_ *testing.T, f *Feed) { f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid}, "main", object(t, feedTestSha)) }, }, { name: "tag push", seed: seedSiteRepo, op: func(t *testing.T, f *Feed) { f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/tags/v1", tapc.RecordCreateAction, feedTestSha)) }, }, { name: "deleted ref", seed: seedSiteRepo, op: func(t *testing.T, f *Feed) { f.refOp(context.Background(), feedTestHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordDeleteAction, "")) }, }, { name: "record from a knot that doesn't host the repo", seed: seedSiteRepo, op: func(t *testing.T, f *Feed) { f.refOp(context.Background(), feedTestOtherHost, feedTestRepoDid, refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha)) }, }, { name: "repo missing from the index", op: func(t *testing.T, f *Feed) { f.refOp(context.Background(), feedTestHost, "did:plc:unknown", refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha)) }, }, { name: "garbage rkey", seed: seedSiteRepo, op: func(_ *testing.T, f *Feed) { f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: "~zz", Bytes: []byte{0xff}}) }, }, { name: "garbage record bytes", seed: seedSiteRepo, op: func(_ *testing.T, f *Feed) { f.refOp(context.Background(), feedTestHost, feedTestRepoDid, knotfeed.RecordOp{Action: tapc.RecordCreateAction, Collection: knotfeed.GitRefCollection, Rkey: "refs~2fheads~2fmain", Bytes: []byte{0xff}}) }, }, { name: "commit with only foreign collections", seed: seedFeedRepo, op: func(t *testing.T, f *Feed) { msg := knotfeed.Message{Type: knotfeed.TypeCommit, Commit: &knotfeed.Commit{ Repo: feedTestRepoDid, Seq: 7, Records: []knotfeed.RecordOp{ {Action: tapc.RecordCreateAction, Collection: "sh.tangled.repo", Rkey: "abc", Bytes: []byte("x")}, }, }} if err := f.handle(feedTestHost)(context.Background(), msg); err != nil { t.Errorf("handle: %v", err) } }, }, } { t.Run(tc.name, func(t *testing.T) { d := feedTestDB(t) if tc.seed != nil { tc.seed(t, d) } r := &deployRecorder{} f := feedFor(t, d, r.hook(nil)) tc.op(t, f) if calls := r.got(); len(calls) != 0 { t.Errorf("deploys = %v, want none", calls) } }) }}
func TestFeedTrigger_RecordsDeployFailure(t *testing.T) { d := feedTestDB(t) seedFeedRepo(t, d) setSiteConfig(t, d) r := &deployRecorder{} f := feedFor(t, d, r.hook(errBoom))
f.maybeDeploy(context.Background(), &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost}, "main", object(t, feedTestSha))
pollFor(t, "the deploy call", func() bool { return len(r.got()) == 1 }) var deploy *models.SiteDeploy pollFor(t, "the deploy row", func() bool { rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10) if err == nil && len(rows) > 0 { deploy = &rows[0] return true } return false }) if deploy.Status != models.SiteDeployStatusFailure { t.Errorf("status = %q, want failure", deploy.Status) } if deploy.Error != "boom" { t.Errorf("error = %q, want the sync failure text", deploy.Error) }}
func TestFeedHandleFailsTheCommitWhenTheDatabaseIsDown(t *testing.T) { d := feedTestDB(t) seedFeedRepo(t, d) setSiteConfig(t, d) r := &deployRecorder{} f := feedFor(t, d, r.hook(nil))
if err := d.Close(); err != nil { t.Fatalf("Close: %v", err) }
msg := knotfeed.Message{Type: knotfeed.TypeCommit, Commit: &knotfeed.Commit{ Repo: feedTestRepoDid, Seq: 9, Records: []knotfeed.RecordOp{ refOpFor(t, "refs/heads/main", tapc.RecordCreateAction, feedTestSha), }, }} if err := f.handle(feedTestHost)(context.Background(), msg); err == nil { t.Fatal("handle accepted a commit while the database was down; the cursor would advance past a dropped deploy") } if calls := r.got(); len(calls) != 0 { t.Errorf("deploys = %v, want none while the database is down", calls) }}
func TestFeedSubscribeDeduplicatesAndUnsubscribes(t *testing.T) { d := feedTestDB(t) f := feedFor(t, d, nil) ctx, cancel := context.WithCancel(context.Background()) defer cancel()
f.Subscribe(ctx, feedTestHost) f.Subscribe(ctx, feedTestHost)
f.mu.Lock() count := len(f.sources) f.mu.Unlock() if count != 1 { t.Fatalf("sources = %d, want 1 after a duplicate subscribe", count) }
f.Unsubscribe(feedTestHost) f.mu.Lock() count = len(f.sources) f.mu.Unlock() if count != 0 { t.Fatalf("sources = %d, want 0 after unsubscribe", count) }
f.Unsubscribe(feedTestOtherHost) f.mu.Lock() count = len(f.sources) f.mu.Unlock() if count != 0 { t.Fatalf("sources = %d, want 0 after unsubscribing a host with no source", count) }}
func TestFeedTrigger_LastPushWinsOverARacingDeploy(t *testing.T) { d := feedTestDB(t) seedFeedRepo(t, d) setSiteConfig(t, d) r := &deployRecorder{} f := feedFor(t, d, r.hook(nil)) repo := &models.Repo{RepoDid: feedTestRepoDid, Knot: feedTestHost}
f.maybeDeploy(context.Background(), repo, "main", object(t, feedTestSha)) f.maybeDeploy(context.Background(), repo, "main", object(t, feedTestOtherSha))
var winner models.SiteDeploy pollFor(t, "a successful deploy row for the later push", func() bool { rows, err := db.GetSiteDeploys(d, feedTestRepoDid, 10) if err != nil { t.Fatalf("GetSiteDeploys: %v", err) } for _, row := range rows { if row.Id > winner.Id { winner = row } } return winner.CommitSHA == feedTestOtherSha && winner.Status == models.SiteDeployStatusSuccess })}