Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353package spindle
import ( "context" "encoding/json" "os" "path/filepath" "testing"
comatproto "github.com/bluesky-social/indigo/api/atproto" "github.com/bluesky-social/indigo/atproto/syntax" jmodels "github.com/bluesky-social/jetstream/pkg/models" "tangled.org/core/api/tangled" "tangled.org/core/jetstream" "tangled.org/core/log" "tangled.org/core/notifier" "tangled.org/core/rbac" "tangled.org/core/spindle/artifactstore" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/secrets")
func newWipeTestSpindle(t *testing.T) *Spindle { t.Helper() tmp := t.TempDir() ctx := context.Background()
d, err := db.Make(ctx, filepath.Join(tmp, "spindle.db")) if err != nil { t.Fatal(err) } t.Cleanup(func() { d.Close() })
e, err := rbac.NewEnforcer(filepath.Join(tmp, "rbac.db")) if err != nil { t.Fatal(err) }
vault, err := secrets.NewSQLiteManager(filepath.Join(tmp, "secrets.db")) if err != nil { t.Fatal(err) }
stores, err := artifactstore.NewStores(config.ArtifactStores{ Disk: config.ArtifactStoreDisk{Dir: filepath.Join(tmp, "artifacts")}, }, "", "") if err != nil { t.Fatal(err) }
jc, err := jetstream.NewJetstreamClient("", "test", nil, nil, log.New("test"), d, false, false) if err != nil { t.Fatal(err) }
return &Spindle{ l: log.New("test"), db: d, e: e, vault: vault, stores: stores, jc: jc, n: ptr(notifier.New()), engs: map[string]models.Engine{}, cfg: &config.Config{ Server: config.Server{ Hostname: "spindle.example.com", RepoDir: filepath.Join(tmp, "repos"), LogDir: filepath.Join(tmp, "logs"), }, }, }}
func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) { t.Helper() ctx := context.Background()
sfx := repoDid.String()[len(repoDid.String())-1:]
if err := s.db.AddRepo(db.Repo{Knot: "knot.example.com", Owner: owner, Rkey: syntax.RecordKey("r" + sfx), RepoDid: repoDid}); err != nil { t.Fatal(err) }
if err := s.db.AddRepoCollaborator(db.RepoCollaborator{OwnerDid: owner, Rkey: syntax.RecordKey("c" + sfx), Subject: "did:plc:collab", RepoDid: repoDid}); err != nil { t.Fatal(err) } if err := s.db.EnqueueJob(ctx, repoDid.String(), models.PipelineId("p"+sfx), nil, tangled.Pipeline{}, "", ""); err != nil { t.Fatal(err) } if err := s.db.SaveMillLease(db.MillLease{LeaseID: "l" + sfx, NodeID: "n1", Epoch: "e1", Engine: "microvm", PipelineID: "p" + sfx, Workflow: "w" + sfx, State: "active", RepoDID: repoDid.String()}); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into mill_artifacts (lease_id, repo_did, pipeline_id, workflow, ref, hash) values (?, ?, ?, ?, ?, 'h')`, "l"+sfx, repoDid.String(), "p"+sfx, "w"+sfx, "out/l"+sfx+".bin"); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into quota_allocations (repo_did, resource, kind, key, amount) values (?, 'compute', 'generic', 'k', 1)`, repoDid.String()); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into webhooks (repo_did, url, events) values (?, 'https://example.com/hook', 'push')`, repoDid.String()); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into webhook_deliveries (webhook_id, event, delivery_id, url, success) values ((select max(id) from webhooks), 'push', ?, 'https://example.com/hook', 1)`, "d"+sfx); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into pull_rounds (repo_did, rkey, rounds) values (?, 'pull1', 2)`, repoDid.String()); err != nil { t.Fatal(err) } if err := s.vault.AddSecret(ctx, secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: owner}); err != nil { t.Fatal(err) } pipeline := &tangled.CiPipeline{ Id: "p" + sfx, Repo: repoDid.String(), Workflows: []*tangled.CiPipeline_Workflow{{Id: "w" + sfx, Name: "w" + sfx, Status: "pending"}}, } payload, err := json.Marshal(pipeline) if err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into pipelines (pipeline_id, repo_did, commit_sha, kind, payload) values (?, ?, '', '', ?)`, pipeline.Id, repoDid.String(), string(payload)); err != nil { t.Fatal(err) } if _, err := s.db.Exec(`insert into workflow_statuses (pipeline_id, workflow, status, created_at) values (?, ?, 'pending', 'now')`, pipeline.Id, "w"+sfx); err != nil { t.Fatal(err) }
if err := os.MkdirAll(s.newRepoPath(repoDid), 0755); err != nil { t.Fatal(err) } for _, ref := range []string{"out/l" + sfx + ".bin", "logs/l" + sfx + ".log"} { if err := s.stores.PutFile(ctx, ref, writeTempFile(t, "data")); err != nil && len(err) > 0 { t.Fatal(err) } }}
func writeTempFile(t *testing.T, content string) string { t.Helper() p := filepath.Join(t.TempDir(), "src") if err := os.WriteFile(p, []byte(content), 0644); err != nil { t.Fatal(err) } return p}
func TestWipeRepoRemovesAllState(t *testing.T) { s := newWipeTestSpindle(t) ctx := context.Background() repoDid := syntax.DID("did:plc:repo1") owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner)
if err := s.WipeRepo(ctx, repoDid, "test"); err != nil { t.Fatalf("WipeRepo: %v", err) }
var n int for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "pipelines", "workflow_statuses", "webhooks", "webhook_deliveries", "pull_rounds"} { if err := s.db.QueryRow(`select count(*) from ` + table).Scan(&n); err != nil { t.Fatal(err) } if n != 0 { t.Fatalf("%s: expected empty, got %d rows", table, n) } }
got, err := s.vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid.String())) if err != nil { t.Fatal(err) } if len(got) != 0 { t.Fatalf("expected secrets gone, got %d", len(got)) }
if _, err := os.Stat(s.newRepoPath(repoDid)); !os.IsNotExist(err) { t.Fatalf("expected clone dir gone, stat err: %v", err) } if _, err := s.stores.Open(ctx, "out/l1.bin"); err == nil { t.Fatal("expected artifact gone") } if _, err := s.stores.Open(ctx, "logs/l1.log"); err == nil { t.Fatal("expected log artifact gone") }
if err := s.WipeRepo(ctx, repoDid, "test again"); err != nil { t.Fatalf("second WipeRepo: %v", err) }}
func TestWipeShouldKeepAllowListedOwners(t *testing.T) { s := newWipeTestSpindle(t) ctx := context.Background() repoDid := syntax.DID("did:plc:repo1") owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner) if err := db.AddToAllowlist(s.db, owner); err != nil { t.Fatal(err) }
if err := s.WipeRepo(ctx, repoDid, "test"); err != nil { t.Fatalf("WipeRepo: %v", err) }
onList, err := db.AllowlistContains(s.db, owner) if err != nil { t.Fatal(err) } if !onList { t.Fatal("A wipe mustn't take the owner off the allow-list.") }}
func TestWipeRepoRemovesLocalEngineArtifacts(t *testing.T) { s := newWipeTestSpindle(t) ctx := context.Background() repoDid := syntax.DID("did:plc:repo1") owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, repoDid, owner)
wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"} localRef := "logs/" + wid.String() + ".log" if err := s.db.SaveArtifactRef(wid.String(), repoDid.String(), wid, localRef, "h"); err != nil { t.Fatal(err) } if err := s.stores.PutFile(ctx, localRef, writeTempFile(t, "log")); err != nil && len(err) > 0 { t.Fatal(err) }
if err := s.WipeRepo(ctx, repoDid, "test"); err != nil { t.Fatalf("WipeRepo: %v", err) }
if _, err := s.stores.Open(ctx, localRef); err == nil { t.Fatal("expected local engine artifact gone") } var n int if err := s.db.QueryRow(`select count(*) from mill_artifacts`).Scan(&n); err != nil { t.Fatal(err) } if n != 0 { t.Fatalf("expected mill_artifacts empty, got %d rows", n) }}
func TestWipeOwnerWipesAllOwnedRepos(t *testing.T) { s := newWipeTestSpindle(t) ctx := context.Background() owner := syntax.DID("did:plc:owner1") other := syntax.DID("did:plc:owner2")
seedWipeRepo(t, s, "did:plc:repo1", owner) seedWipeRepo(t, s, "did:plc:repo2", owner) seedWipeRepo(t, s, "did:plc:repo3", other)
if err := s.WipeOwner(ctx, owner, "test"); err != nil { t.Fatalf("WipeOwner: %v", err) }
var n int if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil { t.Fatal(err) } if n != 1 { t.Fatalf("expected only the other owner's repo left, got %d", n) } repos, err := s.db.ReposByDid("did:plc:repo3") if err != nil || len(repos) != 1 { t.Fatalf("expected repo3 intact, got %v err %v", repos, err) }}
type destroyRecorder struct { models.Engine destroyed []models.WorkflowId}
func (e *destroyRecorder) DestroyWorkflow(_ context.Context, wid models.WorkflowId) error { e.destroyed = append(e.destroyed, wid) return nil}
func TestWipeDestroysFinishedWorkflowStillHoldingItsVM(t *testing.T) { s := newWipeTestSpindle(t) repoDid := syntax.DID("did:plc:repo1") seedWipeRepo(t, s, repoDid, "did:plc:owner1") eng := &destroyRecorder{} s.engs["microvm"] = eng
wid := models.WorkflowId{PipelineId: models.PipelineId("p1"), Name: "w1"} if err := s.db.StatusFailed(wid, "User step error: exited with code 1", 1, s.n); err != nil { t.Fatal(err) }
if _, err := s.cancelRepoWorkflows(context.Background(), repoDid, "banned"); err != nil { t.Fatal(err) } if len(eng.destroyed) != 1 || eng.destroyed[0] != wid { t.Fatalf("destroyed = %v, want the held workflow", eng.destroyed) } if st, err := s.db.GetStatus(wid); err != nil || st != models.StatusKindFailed { t.Fatalf("status = %q (err %v), want the failure kept", st, err) }}
func ptr[T any](v T) *T { return &v }
func TestIngestAccountTakedownWipes(t *testing.T) { s := newWipeTestSpindle(t) owner := syntax.DID("did:plc:owner1")
seedWipeRepo(t, s, "did:plc:repo1", owner)
status := "takendown" err := s.ingest()(context.Background(), &jmodels.Event{ Kind: jmodels.EventKindAccount, Did: owner.String(), Account: &comatproto.SyncSubscribeRepos_Account{Did: owner.String(), Active: false, Status: &status}, }) if err != nil { t.Fatalf("ingest account event: %v", err) }
var n int if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil { t.Fatal(err) } if n != 0 { t.Fatalf("expected repos wiped after takedown, got %d", n) }
seedWipeRepo(t, s, "did:plc:repo1", owner) err = s.ingest()(context.Background(), &jmodels.Event{ Kind: jmodels.EventKindAccount, Did: owner.String(), Account: &comatproto.SyncSubscribeRepos_Account{Did: owner.String(), Active: true}, }) if err != nil { t.Fatal(err) } if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil { t.Fatal(err) } if n != 1 { t.Fatalf("expected active account untouched, got %d", n) }}