alice.pds.demo.boltless.dev
This repository has no description
1package engine
2
3import (
4 "context"
5 "log/slog"
6 "os"
7 "path/filepath"
8 "sync"
9 "testing"
10 "time"
11
12 "tangled.org/core/api/tangled"
13 "tangled.org/core/spindle/config"
14 "tangled.org/core/spindle/db"
15 "tangled.org/core/spindle/models"
16 "tangled.org/core/spindle/secrets"
17)
18
19type mockStep struct {
20 name string
21 command string
22}
23
24func (m mockStep) Name() string { return m.name }
25func (m mockStep) Command() string { return m.command }
26func (m mockStep) Kind() models.StepKind { return models.StepKindUser }
27
28type mockEngine struct {
29 mu sync.Mutex
30 setupCalls []models.WorkflowId
31 runStepCalls []models.WorkflowId
32 setupFunc func(ctx context.Context, wid models.WorkflowId) error
33 runStepFunc func(ctx context.Context, wid models.WorkflowId, idx int) error
34 timeout time.Duration
35}
36
37func (m *mockEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
38 return &models.Workflow{}, nil
39}
40
41func (m *mockEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
42 m.mu.Lock()
43 m.setupCalls = append(m.setupCalls, wid)
44 fn := m.setupFunc
45 m.mu.Unlock()
46 if fn != nil {
47 return fn(ctx, wid)
48 }
49 return nil
50}
51
52func (m *mockEngine) WorkflowTimeout() time.Duration {
53 if m.timeout != 0 {
54 return m.timeout
55 }
56 return 5 * time.Second
57}
58
59func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
60 return nil
61}
62
63func (m *mockEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
64 m.mu.Lock()
65 m.runStepCalls = append(m.runStepCalls, wid)
66 fn := m.runStepFunc
67 m.mu.Unlock()
68
69 if fn != nil {
70 return fn(ctx, wid, idx)
71 }
72 return nil
73}
74
75func newTestDB(t *testing.T) *db.DB {
76 t.Helper()
77 d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db"))
78 if err != nil {
79 t.Fatalf("failed to create test db: %v", err)
80 }
81 return d
82}
83
84func TestStartWorkflows_CollisionRejection(t *testing.T) {
85 t.Parallel()
86
87 testDB := newTestDB(t)
88 logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
89
90 eng := &mockEngine{}
91 pipelineId := models.PipelineId{
92 Knot: "test-knot",
93 Rkey: "test-rkey",
94 }
95
96 // two names that normalize to the same wid must not both run
97 wfColliding1 := models.Workflow{
98 Name: "test-job",
99 Steps: []models.Step{mockStep{name: "step1"}},
100 }
101 wfColliding2 := models.Workflow{
102 Name: "test job",
103 Steps: []models.Step{mockStep{name: "step1"}},
104 }
105 wfUnique := models.Workflow{
106 Name: "unique_job",
107 Steps: []models.Step{mockStep{name: "step1"}},
108 }
109
110 pipeline := &models.Pipeline{
111 Workflows: map[models.Engine][]models.Workflow{
112 eng: {wfColliding1, wfColliding2, wfUnique},
113 },
114 }
115
116 cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
117 StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
118
119 eng.mu.Lock()
120 setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...)
121 eng.mu.Unlock()
122
123 for _, call := range setupCalls {
124 if call.Name == "test-job" || call.Name == "test job" {
125 t.Fatalf("expected colliding workflow %s to not be started", call.Name)
126 }
127 }
128
129 hasUnique := false
130 for _, call := range setupCalls {
131 if call.Name == "unique_job" {
132 hasUnique = true
133 }
134 }
135 if !hasUnique {
136 t.Fatalf("expected unique workflow unique_job to be started")
137 }
138
139 widColliding1 := models.WorkflowId{PipelineId: pipelineId, Name: "test-job"}
140 widColliding2 := models.WorkflowId{PipelineId: pipelineId, Name: "test job"}
141 widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"}
142
143 status1, err := testDB.GetStatus(widColliding1)
144 if err != nil || status1.Status != string(models.StatusKindFailed) {
145 t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err)
146 }
147
148 status2, err := testDB.GetStatus(widColliding2)
149 if err != nil || status2.Status != string(models.StatusKindFailed) {
150 t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err)
151 }
152
153 statusUnique, err := testDB.GetStatus(widUnique)
154 if err != nil || statusUnique.Status != string(models.StatusKindSuccess) {
155 t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err)
156 }
157}