gate_test.go
1package engine
2
3import (
4 "context"
5 "os"
6 "path/filepath"
7 "slices"
8 "strings"
9 "syscall"
10 "testing"
11 "time"
12
13 "git.kilimanjaro.io/ci/internal/lex"
14)
15
16// create starts a run of f's manifest in runs/<id> (sharing f.root).
17func (f *fixture) create(runs, id string, meta Meta) *Engine {
18 f.t.Helper()
19 if meta.Run == "" {
20 meta = Meta{Run: id, Workflow: "w", Repo: "repo", Kind: "manual", Host: "local", Root: f.root}
21 }
22 e, err := Create(f.m, Dir(filepath.Join(runs, id)), meta, Bash{}, f.opts)
23 if err != nil {
24 f.t.Fatal(err)
25 }
26 return e
27}
28
29func reopen(t *testing.T, d Dir, opts Options) *Engine {
30 t.Helper()
31 e, err := Open(d, Bash{}, opts)
32 if err != nil {
33 t.Fatal(err)
34 }
35 return e
36}
37
38func runE(t *testing.T, e *Engine) *State {
39 t.Helper()
40 st, err := e.Run(context.Background())
41 if err != nil {
42 t.Fatal(err)
43 }
44 return st
45}
46
47func approve(i int) Gatekeeper {
48 return GateFunc(func(context.Context, *Engine, *PendingGate) (Choice, error) {
49 return Choice{Option: i, Actor: "test"}, nil
50 })
51}
52
53const gatedWF = `
54engine: devenv
55steps: [plan, deploy, health]
56actions:
57 plan: { command: "true" }
58 deploy: { command: "echo deployed > $CI_STATE/prev", gate: { prompt: "Deploy?" }, concurrency: { group: fleet, supersede: true } }
59 health:
60 command: |
61 n=$(cat $CI_STATE/n 2>/dev/null || echo 0); echo $((n+1)) > $CI_STATE/n
62 if [ "$n" = 0 ]; then
63 printf '%s' '{"version":1,"status":"failure","summary":"503","next":[{"action":"rollback"},{"action":"health","label":"Check again"}]}' > $CI_RESULT
64 fi
65 rollback: { command: "cat $CI_STATE/prev", manual: true, style: danger, gate: { prompt: "Roll back?" } }
66`
67
68func TestPauseAndResume(t *testing.T) {
69 f := newFixture(t, gatedWF)
70 runs := t.TempDir()
71 e := f.create(runs, "a", Meta{})
72 st, err := e.Run(context.Background())
73 e.Log.Close()
74 if err != nil {
75 t.Fatal(err)
76 }
77 if st.Status != StatusPaused || st.Gate == nil || st.Gate.Prompt != "Deploy?" || !slices.Equal(executed(st), []string{"plan"}) {
78 t.Fatalf("not paused at deploy: %s %v %+v", st.Status, executed(st), st.Gate)
79 }
80
81 // a new process reopens the run and approves
82 e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
83 st, err = e.Resume(context.Background(), Choice{Option: 0, Actor: "local:me"})
84 if err != nil {
85 t.Fatal(err)
86 }
87 // health fails the first time and offers a choice: paused again
88 if st.Status != StatusPaused || len(st.Gate.Options) != 2 || st.Gate.From != "health" {
89 t.Fatalf("want choice gate after health: %s %+v", st.Status, st.Gate)
90 }
91 e.Log.Close()
92
93 // "Check again" reruns health, which passes now
94 e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
95 st, err = e.Resume(context.Background(), Choice{Option: 1})
96 e.Log.Close()
97 if err != nil {
98 t.Fatal(err)
99 }
100 if st.Status != StatusSuccess || !slices.Equal(executed(st), []string{"plan", "deploy", "health", "health"}) {
101 t.Fatalf("%s %v %+v", st.Status, executed(st), st.End)
102 }
103
104 // gate events record waiting and approved, with the actor
105 b, _ := os.ReadFile(filepath.Join(runs, "a", "events.ndjson"))
106 if !strings.Contains(string(b), `"state":"approved","choice":"deploy","actor":"local:me"`) {
107 t.Errorf("gate decision not recorded:\n%s", b)
108 }
109
110 // resuming a finished run is an error
111 e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
112 if _, err := e.Resume(context.Background(), Choice{}); err == nil {
113 t.Error("resumed a finished run")
114 }
115 e.Log.Close()
116}
117
118func TestGatekeeperInline(t *testing.T) {
119 f := newFixture(t, gatedWF)
120 f.opts.Gates = approve(0) // deploy, then rollback at the health choice
121 e := f.create(t.TempDir(), "a", Meta{})
122 st, err := e.Run(context.Background())
123 e.Log.Close()
124 if err != nil {
125 t.Fatal(err)
126 }
127 // rollback is off-path: the run ends as a failure after it
128 if st.Status != StatusFailure || !slices.Equal(executed(st), []string{"plan", "deploy", "health", "rollback"}) {
129 t.Fatalf("%s %v %+v", st.Status, executed(st), st.End)
130 }
131}
132
133func TestDetachPausesEvenWithGatekeeper(t *testing.T) {
134 f := newFixture(t, gatedWF)
135 f.opts.Gates = approve(0)
136 f.opts.Detach = true
137 e := f.create(t.TempDir(), "a", Meta{})
138 st, _ := e.Run(context.Background())
139 e.Log.Close()
140 if st.Status != StatusPaused {
141 t.Fatalf("status %s", st.Status)
142 }
143}
144
145func TestDismissAndExpire(t *testing.T) {
146 f := newFixture(t, gatedWF)
147 runs := t.TempDir()
148 e := f.create(runs, "a", Meta{})
149 runE(t, e)
150 e.Log.Close()
151 e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
152 if _, err := e.Resume(context.Background(), Choice{Expire: true}); err == nil {
153 t.Error("expired a gate before its deadline")
154 }
155 st, err := e.Resume(context.Background(), Choice{Dismiss: true, Actor: "me"})
156 e.Log.Close()
157 if err != nil || st.Status != StatusCancelled || st.End.Reason != lex.ReasonDismiss {
158 t.Fatalf("dismiss: %v %+v", err, st)
159 }
160
161 // a gate past its deadline expires whatever the choice
162 f = newFixture(t, strings.Replace(gatedWF, `prompt: "Deploy?"`, `prompt: "Deploy?", timeout: 1ms`, 1))
163 e = f.create(runs, "b", Meta{})
164 runE(t, e)
165 e.Log.Close()
166 time.Sleep(5 * time.Millisecond)
167 e = reopen(t, Dir(filepath.Join(runs, "b")), Options{})
168 st, err = e.Resume(context.Background(), Choice{Option: 0})
169 e.Log.Close()
170 if err != nil || st.Status != StatusCancelled || st.End.Reason != lex.ReasonExpired || len(st.History) != 1 {
171 t.Fatalf("expire: %v %+v", err, st)
172 }
173}
174
175func TestSupersede(t *testing.T) {
176 f := newFixture(t, gatedWF)
177 runs := t.TempDir()
178 old := f.create(runs, "a", Meta{})
179 runE(t, old)
180 old.Log.Close()
181 time.Sleep(2 * time.Millisecond) // Created must differ
182 // another workflow waiting in the same group is not touched
183 other := f.create(runs, "c", Meta{Run: "c", Workflow: "other", Repo: "repo", Kind: "manual", Host: "local", Root: f.root})
184 runE(t, other)
185 other.Log.Close()
186 time.Sleep(2 * time.Millisecond)
187 newer := f.create(runs, "b", Meta{})
188 st, _ := newer.Run(context.Background())
189 newer.Log.Close()
190 if st.Status != StatusPaused {
191 t.Fatalf("newer: %s", st.Status)
192 }
193 if s, _ := Load(Dir(filepath.Join(runs, "a"))); s.Status != StatusCancelled || !strings.Contains(s.End.Detail, "superseded by run b") {
194 t.Errorf("older run: %+v", s.End)
195 }
196 if s, _ := Load(Dir(filepath.Join(runs, "c"))); s.Status != StatusPaused {
197 t.Errorf("other workflow: %s", s.Status)
198 }
199}
200
201func TestConcurrencyLockWaits(t *testing.T) {
202 f := newFixture(t, wf("deploy", "deploy: true"))
203 f.m.Actions["deploy"] = func() lex.Action {
204 a := f.m.Actions["deploy"]
205 a.Concurrency = &lex.Concurrency{Group: "fleet"}
206 return a
207 }()
208 runs := t.TempDir()
209 locks := filepath.Join(runs, ".locks")
210 if err := os.MkdirAll(locks, 0o755); err != nil {
211 t.Fatal(err)
212 }
213 held, err := os.OpenFile(filepath.Join(locks, "fleet.lock"), os.O_CREATE|os.O_RDWR, 0o644)
214 if err != nil {
215 t.Fatal(err)
216 }
217 if _, err := held.WriteString("other-run\n"); err != nil {
218 t.Fatal(err)
219 }
220 if err := syscall.Flock(int(held.Fd()), syscall.LOCK_EX); err != nil {
221 t.Fatal(err)
222 }
223 e := f.create(runs, "a", Meta{})
224 defer e.Log.Close()
225 done := make(chan *State)
226 go func() { st, _ := e.Run(context.Background()); done <- st }()
227 time.Sleep(300 * time.Millisecond)
228 select {
229 case st := <-done:
230 t.Fatalf("ran while the lock was held: %+v", st)
231 default:
232 }
233 held.Close() // releases the lock
234 st := <-done
235 if st.Status != StatusSuccess {
236 t.Fatalf("%+v", st.End)
237 }
238 b, _ := os.ReadFile(filepath.Join(runs, "a", "events.ndjson"))
239 if !strings.Contains(string(b), "waiting for concurrency group fleet (held by run other-run)") {
240 t.Errorf("no waiting message:\n%s", b)
241 }
242}
243
244func TestManualChildRun(t *testing.T) {
245 f := newFixture(t, gatedWF)
246 f.opts.Gates = approve(0)
247 runs := t.TempDir()
248 parent := f.create(runs, "a", Meta{})
249 parent.Opts.Gates = GateFunc(func(_ context.Context, _ *Engine, g *PendingGate) (Choice, error) {
250 if g.From == "health" {
251 return Choice{Option: 1}, nil // check again
252 }
253 return Choice{Option: 0}, nil
254 })
255 pst, _ := parent.Run(context.Background())
256 parent.Log.Close()
257 if pst.Status != StatusSuccess {
258 t.Fatalf("parent %+v", pst.End)
259 }
260
261 // later: roll back against the finished run
262 meta := pst.Meta
263 meta.Run, meta.Parent = "b", filepath.Join(runs, "a")
264 child, err := Create(f.m, Dir(filepath.Join(runs, "b")), meta, Bash{}, Options{Gates: approve(0)})
265 if err != nil {
266 t.Fatal(err)
267 }
268 st, err := child.RunFrom(context.Background(), "rollback")
269 child.Log.Close()
270 if err != nil {
271 t.Fatal(err)
272 }
273 if st.Status != StatusSuccess || !slices.Equal(executed(st), []string{"rollback"}) {
274 t.Fatalf("child %s %v %+v", st.Status, executed(st), st.End)
275 }
276 // it saw the parent's state; the parent is unchanged
277 b, _ := os.ReadFile(filepath.Join(runs, "b", "events.ndjson"))
278 if !strings.Contains(string(b), `"content":"deployed"`) {
279 t.Errorf("child did not read the parent's $CI_STATE:\n%s", b)
280 }
281 if again, _ := Load(Dir(filepath.Join(runs, "a"))); again.Status != StatusSuccess || len(again.History) != len(pst.History) {
282 t.Errorf("parent changed: %+v", again)
283 }
284}
