runner_test.go
1package runner
2
3import (
4 "context"
5 "encoding/json"
6 "io"
7 "net/http"
8 "net/http/httptest"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/gorilla/websocket"
17
18 "git.kilimanjaro.io/ci/internal/engine"
19 "git.kilimanjaro.io/ci/internal/eventlog"
20 "git.kilimanjaro.io/ci/internal/ident"
21 "git.kilimanjaro.io/ci/internal/lex"
22)
23
24var ciBinary string
25
26func TestMain(m *testing.M) {
27 dir, err := os.MkdirTemp("", "ci-runner-test")
28 if err != nil {
29 panic(err)
30 }
31 ciBinary = filepath.Join(dir, "ci")
32 if out, err := exec.Command("go", "build", "-o", ciBinary, "../../cmd/ci").CombinedOutput(); err != nil {
33 panic(string(out))
34 }
35 os.Symlink("ci", filepath.Join(dir, "ci-result"))
36 code := m.Run()
37 os.RemoveAll(dir)
38 os.Exit(code)
39}
40
41func sh(t *testing.T, dir string, args ...string) string {
42 t.Helper()
43 cmd := exec.Command(args[0], args[1:]...)
44 cmd.Dir = dir
45 cmd.Env = append(os.Environ(), "GIT_AUTHOR_NAME=t", "GIT_AUTHOR_EMAIL=t@t", "GIT_COMMITTER_NAME=t",
46 "GIT_COMMITTER_EMAIL=t@t", "GIT_CONFIG_GLOBAL=/dev/null", "GIT_CONFIG_SYSTEM=/dev/null")
47 out, err := cmd.CombinedOutput()
48 if err != nil {
49 t.Fatalf("%v: %v\n%s", args, err, out)
50 }
51 return strings.TrimSpace(string(out))
52}
53
54const deployWF = `
55when: [{ event: push, branch: main }]
56engine: devenv
57steps: [build, deploy, check]
58actions:
59 build: { command: "echo building; echo out > $CI_STATE/built" }
60 deploy: { command: "echo deploying", gate: { prompt: "Deploy demo?" } }
61 check: { command: "echo checking" }
62 undo: { command: "cat $CI_STATE/built", manual: true }
63`
64
65type fixture struct {
66 t *testing.T
67 r *Runner
68 forge string
69 sha string
70}
71
72func newFixture(t *testing.T, files map[string]string) *fixture {
73 t.Helper()
74 t.Setenv("GIT_CONFIG_GLOBAL", "/dev/null")
75 forge := t.TempDir()
76 work := filepath.Join(t.TempDir(), "demo")
77 os.MkdirAll(work, 0o755)
78 sh(t, work, "git", "init", "-q", "-b", "main")
79 for p, c := range files {
80 os.MkdirAll(filepath.Dir(filepath.Join(work, p)), 0o755)
81 os.WriteFile(filepath.Join(work, p), []byte(c), 0o644)
82 }
83 sh(t, work, "git", "add", "-A")
84 sh(t, work, "git", "commit", "-qm", "init")
85 sha := sh(t, work, "git", "rev-parse", "HEAD")
86 sh(t, "", "git", "clone", "-q", "--bare", work, filepath.Join(forge, "demo.git"))
87
88 state := t.TempDir()
89 log, err := eventlog.Open(filepath.Join(state, "runner.db"))
90 if err != nil {
91 t.Fatal(err)
92 }
93 t.Cleanup(func() { log.Close() })
94 cfg := Config{RunsDir: filepath.Join(state, "runs"), CacheDir: filepath.Join(state, "cache"),
95 Knot: "git.test", OwnerDid: "did:plc:me", CloneURL: filepath.Join(forge, "{repo}.git")}
96 l := &Local{Engine: ciBinary, Shell: "bash", LocksDir: filepath.Join(state, "locks")}
97 r, err := New(cfg, log, l, nil)
98 if err != nil {
99 t.Fatal(err)
100 }
101 return &fixture{t: t, r: r, forge: forge, sha: sha}
102}
103
104func (f *fixture) push(ref string, opts ...string) error {
105 return f.r.OnRefUpdate(context.Background(), lex.GitRefUpdate{
106 Repo: ident.RepoDID("git.test", "demo"), Ref: ref, OldSha: strings.Repeat("0", 40), NewSha: f.sha, PushOptions: opts,
107 })
108}
109
110// waitStatus polls until run reaches status.
111func (f *fixture) waitStatus(id, status string) Run {
112 f.t.Helper()
113 deadline := time.Now().Add(30 * time.Second)
114 for time.Now().Before(deadline) {
115 x, err := f.r.GetRun(id)
116 if err == nil && x.Status == status {
117 f.r.mu.Lock()
118 busy := f.r.tailing[id]
119 f.r.mu.Unlock()
120 if !busy {
121 return x
122 }
123 }
124 time.Sleep(50 * time.Millisecond)
125 }
126 x, _ := f.r.GetRun(id)
127 b, _ := os.ReadFile(filepath.Join(x.Dir, "segment-1.log"))
128 f.t.Fatalf("run %s: status %q, want %q\n%s", id, x.Status, status, b)
129 return Run{}
130}
131
132func (f *fixture) records() map[string][]json.RawMessage {
133 evs, _ := f.r.Log.List(0, 1000)
134 out := map[string][]json.RawMessage{}
135 for _, e := range evs {
136 out[e.Nsid] = append(out[e.Nsid], e.Event)
137 }
138 return out
139}
140
141func statuses(t *testing.T, recs []json.RawMessage, workflow string) []string {
142 t.Helper()
143 var out []string
144 for _, b := range recs {
145 var s lex.PipelineStatus
146 json.Unmarshal(b, &s)
147 if s.Workflow == workflow {
148 out = append(out, s.Status)
149 }
150 }
151 return out
152}
153
154func TestPushGateApproveAndLogs(t *testing.T) {
155 f := newFixture(t, map[string]string{
156 ".tangled/workflows/deploy.yml": deployWF,
157 ".tangled/workflows/bad.yml": "engine: devenv\nsteps: [nope]\nactions: {}\n",
158 ".tangled/workflows/other.yml": "engine: nixery\nsteps: []\n",
159 })
160 if err := f.push("refs/heads/main"); err != nil {
161 t.Fatal(err)
162 }
163 runs, _ := f.r.ListRuns("demo", 10)
164 if len(runs) != 1 {
165 t.Fatalf("runs %+v", runs)
166 }
167 id := runs[0].ID
168 f.waitStatus(id, engine.StatusPaused)
169
170 recs := f.records()
171 var p lex.Pipeline
172 json.Unmarshal(recs[lex.PipelineNSID][0], &p)
173 if len(p.Workflows) != 2 || p.TriggerMetadata.Push.NewSha != f.sha || *p.TriggerMetadata.Repo.RepoDid != "did:web:git.test:demo" {
174 t.Errorf("pipeline %+v", p)
175 }
176 if got := statuses(t, recs[lex.PipelineStatusNSID], "bad"); len(got) != 1 || got[0] != lex.StatusFailed {
177 t.Errorf("bad workflow statuses %v", got)
178 }
179 if len(recs[lex.GateNSID]) != 1 || !strings.Contains(string(recs[lex.GateNSID][0]), "Deploy demo?") {
180 t.Errorf("gate records %s", recs[lex.GateNSID])
181 }
182
183 x, _ := f.r.GetRun(id)
184 if err := f.r.AnswerGate(context.Background(), x, gateAnswer{Action: "deploy", Actor: "did:plc:me"}); err != nil {
185 t.Fatal(err)
186 }
187 f.waitStatus(id, engine.StatusSuccess)
188 if got := statuses(t, f.records()[lex.PipelineStatusNSID], "deploy"); strings.Join(got, ",") != "pending,running,success" {
189 t.Errorf("deploy statuses %v", got)
190 }
191
192 // API: log.txt and /logs in upstream's LogLine format
193 srv := httptest.NewServer(f.r.Handler("tok"))
194 defer srv.Close()
195 req, _ := http.NewRequest("GET", srv.URL+"/api/runs/"+id+"/log.txt", nil)
196 req.Header.Set("Authorization", "Bearer tok")
197 resp, err := http.DefaultClient.Do(req)
198 if err != nil {
199 t.Fatal(err)
200 }
201 txt, _ := io.ReadAll(resp.Body)
202 resp.Body.Close()
203 for _, want := range []string{"== build (#1)", "building", "== gate approved", "checking", "== run success"} {
204 if !strings.Contains(string(txt), want) {
205 t.Errorf("log.txt lacks %q:\n%s", want, txt)
206 }
207 }
208 h := http.Header{"Authorization": {"Bearer tok"}}
209 conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http")+"/logs/git.test/"+x.Rkey+"/deploy", h)
210 if err != nil {
211 t.Fatal(err)
212 }
213 var lines []lex.LogLine
214 for {
215 _, msg, err := conn.ReadMessage()
216 if err != nil {
217 break
218 }
219 var l lex.LogLine
220 json.Unmarshal(msg, &l)
221 lines = append(lines, l)
222 }
223 conn.Close()
224 if len(lines) < 9 || lines[0].Kind != "control" || lines[0].StepStatus != "start" || lines[0].Content != "build" ||
225 lines[1].Kind != "data" || lines[1].Content != "building" || lines[1].StepId != 1 {
226 t.Errorf("log lines %+v", lines)
227 }
228
229 // unauthenticated API calls are refused
230 if resp, _ := http.Get(srv.URL + "/api/runs"); resp.StatusCode != 401 {
231 t.Errorf("unauthenticated: %d", resp.StatusCode)
232 }
233
234 // a manual action run against the finished run sees its $CI_STATE
235 child, err := f.r.ManualAction(context.Background(), x, "undo", "did:plc:me")
236 if err != nil {
237 t.Fatal(err)
238 }
239 cx := f.waitStatus(child, engine.StatusSuccess)
240 b, _ := os.ReadFile(engine.Dir(cx.Dir).Events())
241 if !strings.Contains(string(b), `"content":"out"`) {
242 t.Errorf("child did not read the parent's state:\n%s", b)
243 }
244 if _, err := f.r.ManualAction(context.Background(), x, "check", ""); err == nil {
245 t.Error("non-manual action accepted")
246 }
247 // the child's run doesn't touch the workflow's Tangled status
248 if got := statuses(t, f.records()[lex.PipelineStatusNSID], "deploy"); len(got) != 3 {
249 t.Errorf("child run published statuses: %v", got)
250 }
251}
252
253func TestPushFilters(t *testing.T) {
254 f := newFixture(t, map[string]string{".tangled/workflows/deploy.yml": deployWF})
255 f.push("refs/heads/feature")
256 f.push("refs/heads/main", "skip-ci")
257 f.r.OnRefUpdate(context.Background(), lex.GitRefUpdate{Repo: ident.RepoDID("git.test", "demo"), Ref: "refs/heads/main", NewSha: strings.Repeat("0", 40)})
258 if runs, _ := f.r.ListRuns("", 10); len(runs) != 0 {
259 t.Errorf("runs started: %+v", runs)
260 }
261 if err := f.r.OnRefUpdate(context.Background(), lex.GitRefUpdate{Repo: ident.RepoDID("other.host", "demo"), Ref: "refs/heads/main", NewSha: f.sha}); err == nil {
262 t.Error("accepted a repo on another forge")
263 }
264}
265
266func TestReconcile(t *testing.T) {
267 f := newFixture(t, map[string]string{".tangled/workflows/deploy.yml": "engine: devenv\nsteps: [a]\nactions: { a: { command: 'true' } }\n"})
268 if err := f.push("refs/heads/main"); err != nil {
269 t.Fatal(err)
270 }
271 runs, _ := f.r.ListRuns("", 10)
272 done := f.waitStatus(runs[0].ID, engine.StatusSuccess)
273
274 // a run the runner lost track of mid-way: its state says running, its
275 // process is gone. After a restart it is settled as failed.
276 lost := done
277 lost.ID, lost.Dir = "lost", filepath.Join(f.r.Cfg.RunsDir, "lost")
278 if err := os.CopyFS(lost.Dir, os.DirFS(done.Dir)); err != nil {
279 t.Fatal(err)
280 }
281 st, _ := engine.Load(engine.Dir(lost.Dir))
282 st.Status, st.End = engine.StatusRunning, nil
283 st.Meta.Run = "lost"
284 st.Save(engine.Dir(lost.Dir))
285 f.r.insertRun(Run{ID: "lost", Pipeline: done.Pipeline, Rkey: done.Rkey, Workflow: "lostwf", Repo: "demo", Dir: lost.Dir, Status: "running", Created: now(), Updated: now()})
286 f.r.updateRun("lost", `handle = ?, segment = 1`, "pid:999999999")
287
288 if err := f.r.Reconcile(context.Background()); err != nil {
289 t.Fatal(err)
290 }
291 x := f.waitStatus("lost", engine.StatusFailure)
292 st, _ = engine.Load(engine.Dir(x.Dir))
293 if !strings.Contains(st.End.Detail, "ended without finishing") {
294 t.Errorf("end %+v", st.End)
295 }
296 if got := statuses(t, f.records()[lex.PipelineStatusNSID], "lostwf"); len(got) != 1 || got[0] != lex.StatusFailed {
297 t.Errorf("lost statuses %v", got)
298 }
299}
300
301func TestSystemdCommand(t *testing.T) {
302 s := &Systemd{Engine: "/bin/ci", LocksDir: "/var/lib/ci/locks", CacheDir: "/var/lib/ci/cache",
303 Credentials: map[string]string{"age-identity": "/run/keys/ci-age", "ssh-key": "/run/keys/ci-ssh"}, Pasta: "/bin/pasta", Properties: []string{"MemoryMax=8G"}, Env: []string{"CI_CLONE_URL=git@h:{repo}"}}
304 argv := strings.Join(s.Command(Job{Run: "r1-deploy", Dir: "/var/lib/ci/runs/r1-deploy", Repo: "sovrn", Segment: 2, Args: []string{"--choose", "1"}}), " ")
305 for _, want := range []string{
306 "--unit=ci-r1-deploy-2", "--slice=ci.slice", "--uid=ci", "ProtectSystem=strict", "PrivateTmp=yes",
307 "ReadWritePaths=/var/lib/ci/runs/r1-deploy /var/lib/ci/cache/home/sovrn /var/lib/ci/locks /nix/var/nix/daemon-socket",
308 "LoadCredential=age-identity:/run/keys/ci-age", "LoadCredential=ssh-key:/run/keys/ci-ssh", "--setenv=HOME=/var/lib/ci/cache/home/sovrn", "MemoryMax=8G",
309 "-- /bin/pasta --config-net --quiet -t none -u none -T none -U none -- /bin/ci runner-exec --shell devenv --locks-dir /var/lib/ci/locks --choose 1 /var/lib/ci/runs/r1-deploy",
310 } {
311 if !strings.Contains(argv, want) {
312 t.Errorf("command lacks %q:\n%s", want, argv)
313 }
314 }
315 s.PastaHostPorts = []int{23231, 22}
316 if p := strings.Join(s.PastaCommand(), " "); p != "/bin/pasta --config-net --quiet -t none -u none -T 23231,22 -U none" {
317 t.Errorf("pasta: %s", p)
318 }
319}
