Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356package engine
import ( "context" "log/slog" "os" "path/filepath" "reflect" "sync" "testing" "time"
"tangled.org/core/api/tangled" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/secrets")
type mockStep struct { name string command string}
func (m mockStep) Name() string { return m.name }func (m mockStep) Command() string { return m.command }func (m mockStep) Kind() models.StepKind { return models.StepKindUser }
type emptyVault struct{}
func (emptyVault) AddSecret(context.Context, secrets.UnlockedSecret) error { return nil}
func (emptyVault) RemoveSecret(context.Context, secrets.Secret[any]) error { return nil}
func (emptyVault) GetSecretsLocked(context.Context, secrets.RepoIdentifier) ([]secrets.LockedSecret, error) { return nil, nil}
func (emptyVault) GetSecretsUnlocked(context.Context, secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) { return nil, nil}
type staticCacheController struct { bindings []models.CacheBinding}
func (c staticCacheController) Plan(context.Context, *models.Pipeline, *models.Workflow) ([]models.CacheBinding, error) { return c.bindings, nil}
func (staticCacheController) Apply(context.Context, CacheUpdate) error { return nil}
type mockEngine struct { mu sync.Mutex setupCalls []models.WorkflowId runStepCalls []models.WorkflowId setupFunc func(ctx context.Context, wid models.WorkflowId) error runStepFunc func(ctx context.Context, wid models.WorkflowId, idx int) error runWorkflowFunc func(*models.Workflow) timeout time.Duration}
func (m *mockEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { return &models.Workflow{}, nil}
func (m *mockEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { m.mu.Lock() m.setupCalls = append(m.setupCalls, wid) fn := m.setupFunc m.mu.Unlock() if fn != nil { return fn(ctx, wid) } return nil}
func (m *mockEngine) WorkflowTimeout() time.Duration { if m.timeout != 0 { return m.timeout } return 5 * time.Second}
func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { return nil}
func (m *mockEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { m.mu.Lock() m.runStepCalls = append(m.runStepCalls, wid) runStepFn := m.runStepFunc runWorkflowFn := m.runWorkflowFunc m.mu.Unlock()
if runWorkflowFn != nil { runWorkflowFn(w) } if runStepFn != nil { return runStepFn(ctx, wid, idx) } return nil}
func newTestDB(t *testing.T) *db.DB { t.Helper() d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatalf("failed to create test db: %v", err) } return d}
func TestStartWorkflows_CollisionRejection(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
eng := &mockEngine{} pipelineId := models.PipelineId{ Knot: "test-knot", Rkey: "test-rkey", }
// two names that normalize to the same wid must not both run wfColliding1 := models.Workflow{ Name: "test-job", Steps: []models.Step{mockStep{name: "step1"}}, } wfColliding2 := models.Workflow{ Name: "test job", Steps: []models.Step{mockStep{name: "step1"}}, } wfUnique := models.Workflow{ Name: "unique_job", Steps: []models.Step{mockStep{name: "step1"}}, }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {wfColliding1, wfColliding2, wfUnique}, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
eng.mu.Lock() setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...) eng.mu.Unlock()
for _, call := range setupCalls { if call.Name == "test-job" || call.Name == "test job" { t.Fatalf("expected colliding workflow %s to not be started", call.Name) } }
hasUnique := false for _, call := range setupCalls { if call.Name == "unique_job" { hasUnique = true } } if !hasUnique { t.Fatalf("expected unique workflow unique_job to be started") }
widColliding1 := models.WorkflowId{PipelineId: pipelineId, Name: "test-job"} widColliding2 := models.WorkflowId{PipelineId: pipelineId, Name: "test job"} widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"}
status1, err := testDB.GetStatus(widColliding1) if err != nil || status1.Status != string(models.StatusKindFailed) { t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err) }
status2, err := testDB.GetStatus(widColliding2) if err != nil || status2.Status != string(models.StatusKindFailed) { t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err) }
statusUnique, err := testDB.GetStatus(widUnique) if err != nil || statusUnique.Status != string(models.StatusKindSuccess) { t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err) }}
func TestCancelWorkflow_NotOverwritten(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
stepStarted := make(chan struct{}) eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int) error { close(stepStarted) <-ctx.Done() return ctx.Err() }, }
pipelineId := models.PipelineId{ Knot: "test-knot", Rkey: "test-rkey", }
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "cancel_test_job", }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "cancel_test_job", Steps: []models.Step{mockStep{name: "step1"}}, }, }, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} doneChan := make(chan struct{}) go func() { StartWorkflows(logger, nil, cfg, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(doneChan) }()
select { case <-stepStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for step to start") }
_ = testDB.StatusCancelled(wid, "User canceled the workflow", -1, nil) CancelWorkflow(wid)
select { case <-doneChan: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }
// the runner writes StatusCancelled itself when it sees the canceled ctx // the handler writes nothing for a live wf, so nothing lands after to overwrite it st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st.Status != string(models.StatusKindCancelled) { t.Fatalf("expected status to be cancelled, got %s", st.Status) }}
func TestSetupTimeout_ReportsTimeout(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
// setup blocks past the workflow timeout, so it should land as timeout not failed eng := &mockEngine{ timeout: 100 * time.Millisecond, setupFunc: func(ctx context.Context, wid models.WorkflowId) error { <-ctx.Done() return ctx.Err() }, }
pipelineId := models.PipelineId{Knot: "test-knot", Rkey: "test-rkey"} wid := models.WorkflowId{PipelineId: pipelineId, Name: "timeout_job"}
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "timeout_job", Steps: []models.Step{mockStep{name: "step1"}}}}, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st.Status != string(models.StatusKindTimeout) { t.Fatalf("expected status to be timeout, got %s", st.Status) }
if len(eng.runStepCalls) != 0 { t.Fatalf("expected no steps to run after setup timeout, got %d", len(eng.runStepCalls)) }}
func TestStartWorkflowsPassesPlannedCacheBindingsToEngine(t *testing.T) { t.Parallel()
database := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) expected := []models.CacheBinding{{ EntryIndex: 0, Paths: []string{"deps"}, Key: "deps", SaveID: "cache-id", SaveKey: "objects/cache-id", }} var got []models.CacheBinding eng := &mockEngine{ runWorkflowFunc: func(workflow *models.Workflow) { got = append([]models.CacheBinding(nil), workflow.CacheBindings...) }, } pipelineID := models.PipelineId{Knot: "test-knot", Rkey: "test-rkey"} pipeline := &models.Pipeline{ RepoDid: "did:web:example.com", TrustedSource: true, Workflows: map[models.Engine][]models.Workflow{ eng: {{ Name: "cache-job", Engine: "microvm", Steps: []models.Step{mockStep{name: "step1"}}, Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}}, }}, }, } cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} store := &fakeStorage{objects: make(map[string][]byte)}
StartWorkflows( logger, emptyVault{}, cfg, nil, database, nil, store, staticCacheController{bindings: expected}, context.Background(), pipeline, pipelineID, )
if !reflect.DeepEqual(got, expected) { t.Fatalf("cache bindings passed to engine = %#v, want %#v", got, expected) }}