test(e2e): chain test of corpus jobs: fake RunnerService, host loop, container sandbox, executor, recordings
- commit
- 012b7089c34b661465a537fa1986286a49b5045c
- author
- brice · · 2026-10-12 18:55 UTC
- parent
- dddf7ab
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +576 -0
A internal/e2e/chain_test.go +320 -0
| @@ -0,0 +1,320 @@ | |||
| 1 | + | // SPDX-License-Identifier: MIT | |
| 2 | + | ||
| 3 | + | package e2e | |
| 4 | + | ||
| 5 | + | import ( | |
| 6 | + | "context" | |
| 7 | + | "encoding/json" | |
| 8 | + | "fmt" | |
| 9 | + | "net/http" | |
| 10 | + | "os" | |
| 11 | + | "path/filepath" | |
| 12 | + | "regexp" | |
| 13 | + | "sort" | |
| 14 | + | "strings" | |
| 15 | + | "testing" | |
| 16 | + | "time" | |
| 17 | + | ||
| 18 | + | ermite "bnema.dev/ermite-api/gen/ermite" | |
| 19 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |
| 20 | + | ||
| 21 | + | "bnema.dev/ermite-runner/internal/host" | |
| 22 | + | "bnema.dev/ermite-runner/internal/host/hosttest" | |
| 23 | + | ) | |
| 24 | + | ||
| 25 | + | // The chain test runs the whole path of a job: a fake RunnerService hands out a | |
| 26 | + | // corpus job, the real host loop starts a real sandbox (the runner image in a | |
| 27 | + | // container), the real executor runs the steps, and the host reports to the | |
| 28 | + | // fake with AppendLog, UpdateJob and CompleteJob. What the fake stored is | |
| 29 | + | // compared with the recording of the same job made on github.com. | |
| 30 | + | // | |
| 31 | + | // It needs a checkout of ermite-conformance (corpus/ and recordings/), found | |
| 32 | + | // through ERMITE_CONFORMANCE_DIR or as a sibling of this repository, and is | |
| 33 | + | // read-only on it. | |
| 34 | + | ||
| 35 | + | func conformanceDir(t *testing.T) string { | |
| 36 | + | t.Helper() | |
| 37 | + | cands := []string{os.Getenv("ERMITE_CONFORMANCE_DIR")} | |
| 38 | + | wd, _ := os.Getwd() | |
| 39 | + | for d := wd; d != "/" && d != "."; d = filepath.Dir(d) { | |
| 40 | + | cands = append(cands, filepath.Join(d, "ermite-conformance", ".worktrees", "corpus"), filepath.Join(d, "ermite-conformance")) | |
| 41 | + | } | |
| 42 | + | for _, c := range cands { | |
| 43 | + | if c == "" { | |
| 44 | + | continue | |
| 45 | + | } | |
| 46 | + | if st, err := os.Stat(filepath.Join(c, "recordings")); err == nil && st.IsDir() { | |
| 47 | + | return c | |
| 48 | + | } | |
| 49 | + | } | |
| 50 | + | t.Skip("ermite-conformance checkout not found (set ERMITE_CONFORMANCE_DIR)") | |
| 51 | + | return "" | |
| 52 | + | } | |
| 53 | + | ||
| 54 | + | type recStep struct { | |
| 55 | + | Name string `json:"name"` | |
| 56 | + | Conclusion string `json:"conclusion"` | |
| 57 | + | Log []string `json:"log"` | |
| 58 | + | } | |
| 59 | + | ||
| 60 | + | type recJob struct { | |
| 61 | + | Name string `json:"name"` | |
| 62 | + | Conclusion string `json:"conclusion"` | |
| 63 | + | Steps []recStep `json:"steps"` | |
| 64 | + | } | |
| 65 | + | ||
| 66 | + | type recording struct { | |
| 67 | + | Pushes []struct { | |
| 68 | + | Runs []struct { | |
| 69 | + | Jobs []recJob `json:"jobs"` | |
| 70 | + | } `json:"runs"` | |
| 71 | + | } `json:"pushes"` | |
| 72 | + | } | |
| 73 | + | ||
| 74 | + | var ( | |
| 75 | + | uuidRe = regexp.MustCompile(`[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}`) | |
| 76 | + | // Machine-dependent parts of the recordings, like the normalisation rules of | |
| 77 | + | // ermite-conformance: random script names and the repository name. | |
| 78 | + | repoRe = regexp.MustCompile(`/work/hello/hello`) | |
| 79 | + | ) | |
| 80 | + | ||
| 81 | + | func normalize(l string) string { | |
| 82 | + | l = repoRe.ReplaceAllString(l, "/work/<repo>/<repo>") | |
| 83 | + | return uuidRe.ReplaceAllString(l, "<uuid>") | |
| 84 | + | } | |
| 85 | + | ||
| 86 | + | func chainContexts(job string) string { | |
| 87 | + | return fmt.Sprintf(`{"github":{"repository":"octo/hello","repository_owner":"octo","run_id":"1","run_number":"1","run_attempt":"1", | |
| 88 | + | "ref":"refs/heads/scn/x/r1","ref_name":"scn/x/r1","ref_type":"branch","ref_protected":false,"sha":"0123456789abcdef0123456789abcdef01234567", | |
| 89 | + | "event_name":"push","event":{"ref":"refs/heads/scn/x/r1"},"job":%q,"workflow":"w","actor":"octo","triggering_actor":"octo", | |
| 90 | + | "server_url":"https://ermite.example","retention_days":"90"}}`, job) | |
| 91 | + | } | |
| 92 | + | ||
| 93 | + | func TestChainCorpus(t *testing.T) { | |
| 94 | + | image := needE2E(t) | |
| 95 | + | dir := conformanceDir(t) | |
| 96 | + | drv := newDriver(t, image) | |
| 97 | + | ||
| 98 | + | // scenario -> jobs to run (jobs with `needs` depend on the server and are left out) | |
| 99 | + | scenarios := []struct { | |
| 100 | + | name string | |
| 101 | + | jobs []string | |
| 102 | + | }{ | |
| 103 | + | {"env-precedence", []string{"main", "other"}}, | |
| 104 | + | {"env-expressions", []string{"main"}}, | |
| 105 | + | {"out-github-env", []string{"main"}}, | |
| 106 | + | {"out-github-env-reserved", []string{"node-options", "github-prefix"}}, | |
| 107 | + | {"out-github-path", []string{"main"}}, | |
| 108 | + | {"out-step-outputs", []string{"main"}}, | |
| 109 | + | {"ctx-steps", []string{"main"}}, | |
| 110 | + | {"if-step-conditions", []string{"main"}}, | |
| 111 | + | {"shell-builtin", []string{"main"}}, | |
| 112 | + | {"shell-custom", []string{"main"}}, | |
| 113 | + | {"shell-exit-codes", nil}, // every job of the workflow | |
| 114 | + | {"shell-working-directory", nil}, | |
| 115 | + | {"out-job-outputs", []string{"producer"}}, | |
| 116 | + | } | |
| 117 | + | ||
| 118 | + | srv := hosttest.New() | |
| 119 | + | t.Cleanup(srv.Close) | |
| 120 | + | ||
| 121 | + | type want struct { | |
| 122 | + | scenario string | |
| 123 | + | job recJob | |
| 124 | + | } | |
| 125 | + | wants := map[string]want{} | |
| 126 | + | var queue []*ermite.Job | |
| 127 | + | for _, sc := range scenarios { | |
| 128 | + | files, _ := filepath.Glob(filepath.Join(dir, "corpus", sc.name, "tree", ".github", "workflows", "*.yml")) | |
| 129 | + | if len(files) != 1 { | |
| 130 | + | t.Fatalf("%s: workflow files %v", sc.name, files) | |
| 131 | + | } | |
| 132 | + | yaml, err := os.ReadFile(files[0]) | |
| 133 | + | if err != nil { | |
| 134 | + | t.Fatal(err) | |
| 135 | + | } | |
| 136 | + | raw, err := os.ReadFile(filepath.Join(dir, "recordings", sc.name+".json")) | |
| 137 | + | if err != nil { | |
| 138 | + | t.Fatal(err) | |
| 139 | + | } | |
| 140 | + | var rec recording | |
| 141 | + | if err := json.Unmarshal(raw, &rec); err != nil { | |
| 142 | + | t.Fatal(err) | |
| 143 | + | } | |
| 144 | + | for _, rj := range rec.Pushes[0].Runs[0].Jobs { | |
| 145 | + | if sc.jobs != nil && !contains(sc.jobs, rj.Name) { | |
| 146 | + | continue | |
| 147 | + | } | |
| 148 | + | id := sc.name + "/" + rj.Name | |
| 149 | + | queue = append(queue, &ermite.Job{ | |
| 150 | + | JobId: id, RunId: 1, Attempt: 1, JobToken: "ermitejob_" + strings.ReplaceAll(id, "/", "_"), | |
| 151 | + | WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: string(yaml), JobKey: rj.Name, Matrix: "{}", | |
| 152 | + | Contexts: chainContexts(rj.Name), Repo: "hello", TimeoutMinutes: 10, | |
| 153 | + | }) | |
| 154 | + | wants[id] = want{sc.name, rj} | |
| 155 | + | } | |
| 156 | + | } | |
| 157 | + | srv.Queue(queue...) | |
| 158 | + | ||
| 159 | + | r := &host.Runner{ | |
| 160 | + | Cfg: host.Config{URL: srv.URL, Name: "chain", Capacity: 2, Version: "test"}, | |
| 161 | + | Driver: drv, | |
| 162 | + | Poll: host.NewClient(srv.URL, hosttest.RunnerToken, http.DefaultTransport), | |
| 163 | + | Job: func(tok string) ermiteconnect.RunnerServiceClient { | |
| 164 | + | return host.NewClient(srv.URL, tok, http.DefaultTransport) | |
| 165 | + | }, | |
| 166 | + | Timing: host.Timing{LogFlush: 20 * time.Millisecond, NoJobPause: 100 * time.Millisecond}, | |
| 167 | + | Logf: func(f string, a ...any) { t.Logf(f, a...) }, | |
| 168 | + | } | |
| 169 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 170 | + | done := make(chan error, 1) | |
| 171 | + | go func() { done <- r.Run(ctx) }() | |
| 172 | + | defer func() { | |
| 173 | + | cancel() | |
| 174 | + | select { | |
| 175 | + | case err := <-done: | |
| 176 | + | if err != nil { | |
| 177 | + | t.Errorf("Run: %v", err) | |
| 178 | + | } | |
| 179 | + | case <-time.After(60 * time.Second): | |
| 180 | + | t.Error("Run did not stop") | |
| 181 | + | } | |
| 182 | + | }() | |
| 183 | + | ||
| 184 | + | deadline := time.After(5 * time.Minute) | |
| 185 | + | for n := 0; n < len(queue); n++ { | |
| 186 | + | select { | |
| 187 | + | case <-srv.Done(): | |
| 188 | + | case <-deadline: | |
| 189 | + | t.Fatalf("only %d of %d jobs finished", n, len(queue)) | |
| 190 | + | } | |
| 191 | + | } | |
| 192 | + | ||
| 193 | + | var ids []string | |
| 194 | + | for id := range wants { | |
| 195 | + | ids = append(ids, id) | |
| 196 | + | } | |
| 197 | + | sort.Strings(ids) | |
| 198 | + | for _, id := range ids { | |
| 199 | + | w := wants[id] | |
| 200 | + | t.Run(id, func(t *testing.T) { compareJob(t, srv.Result(id), w.job, w.scenario) }) | |
| 201 | + | } | |
| 202 | + | for _, e := range srv.Violations() { | |
| 203 | + | t.Errorf("the fake server saw a contract violation: %s", e) | |
| 204 | + | } | |
| 205 | + | } | |
| 206 | + | ||
| 207 | + | func contains(l []string, s string) bool { | |
| 208 | + | for _, x := range l { | |
| 209 | + | if x == s { | |
| 210 | + | return true | |
| 211 | + | } | |
| 212 | + | } | |
| 213 | + | return false | |
| 214 | + | } | |
| 215 | + | ||
| 216 | + | // compareJob checks the steps, conclusions, outputs and log lines reported for | |
| 217 | + | // a job against the recording. "Set up job" (runner and image details) and the | |
| 218 | + | // ermite-dump step that the harness appends to every job are not compared. | |
| 219 | + | func compareJob(t *testing.T, got hosttest.Result, want recJob, scenario string) { | |
| 220 | + | if got.Complete == nil { | |
| 221 | + | t.Fatal("the job did not complete") | |
| 222 | + | } | |
| 223 | + | // The conclusion of the job as GitHub recorded it, except where the dump | |
| 224 | + | // step failed on GitHub (it only fails after a failed step in some | |
| 225 | + | // scenarios and is not part of the job here). | |
| 226 | + | wantConcl := map[string]ermite.CiConclusion{ | |
| 227 | + | "success": ermite.CiConclusion_CI_CONCLUSION_SUCCESS, "failure": ermite.CiConclusion_CI_CONCLUSION_FAILURE, | |
| 228 | + | "cancelled": ermite.CiConclusion_CI_CONCLUSION_CANCELLED, | |
| 229 | + | } | |
| 230 | + | jobWant := wantConcl[want.Conclusion] | |
| 231 | + | for _, s := range want.Steps { | |
| 232 | + | if s.Name == "ermite-dump" && s.Conclusion != "success" && want.Conclusion == "failure" { | |
| 233 | + | // the job failed only because of the dump step | |
| 234 | + | jobWant = ermite.CiConclusion_CI_CONCLUSION_SUCCESS | |
| 235 | + | for _, o := range want.Steps { | |
| 236 | + | if o.Name != "ermite-dump" && o.Conclusion == "failure" { | |
| 237 | + | jobWant = ermite.CiConclusion_CI_CONCLUSION_FAILURE | |
| 238 | + | } | |
| 239 | + | } | |
| 240 | + | } | |
| 241 | + | } | |
| 242 | + | if got.Complete.Conclusion != jobWant { | |
| 243 | + | t.Errorf("job conclusion %v, want %v", got.Complete.Conclusion, jobWant) | |
| 244 | + | } | |
| 245 | + | ||
| 246 | + | // Steps by number, with their lines. | |
| 247 | + | byStep := map[int32][]string{} | |
| 248 | + | for _, l := range got.Lines { | |
| 249 | + | byStep[l.Step] = append(byStep[l.Step], normalize(l.Text)) | |
| 250 | + | } | |
| 251 | + | var steps []int32 | |
| 252 | + | for n := range got.Steps { | |
| 253 | + | steps = append(steps, n) | |
| 254 | + | } | |
| 255 | + | sort.Slice(steps, func(i, j int) bool { return steps[i] < steps[j] }) | |
| 256 | + | ||
| 257 | + | type es = recStep | |
| 258 | + | var wantSteps, gotSteps []es | |
| 259 | + | for _, s := range want.Steps { | |
| 260 | + | if s.Name == "Set up job" || s.Name == "ermite-dump" { | |
| 261 | + | continue | |
| 262 | + | } | |
| 263 | + | wantSteps = append(wantSteps, s) | |
| 264 | + | } | |
| 265 | + | conc := map[ermite.CiConclusion]string{ | |
| 266 | + | ermite.CiConclusion_CI_CONCLUSION_SUCCESS: "success", ermite.CiConclusion_CI_CONCLUSION_FAILURE: "failure", | |
| 267 | + | ermite.CiConclusion_CI_CONCLUSION_CANCELLED: "cancelled", ermite.CiConclusion_CI_CONCLUSION_SKIPPED: "skipped", | |
| 268 | + | } | |
| 269 | + | for _, n := range steps { | |
| 270 | + | u := got.Steps[n] | |
| 271 | + | if u.Name == "Set up job" { | |
| 272 | + | continue | |
| 273 | + | } | |
| 274 | + | if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.StartedAt == nil || u.CompletedAt == nil { | |
| 275 | + | t.Errorf("step %d %q is not a complete snapshot: %v", n, u.Name, u) | |
| 276 | + | } | |
| 277 | + | gotSteps = append(gotSteps, es{Name: u.Name, Conclusion: conc[u.Conclusion], Log: byStep[n]}) | |
| 278 | + | } | |
| 279 | + | if len(gotSteps) != len(wantSteps) { | |
| 280 | + | var names []string | |
| 281 | + | for _, s := range gotSteps { | |
| 282 | + | names = append(names, s.Name) | |
| 283 | + | } | |
| 284 | + | t.Fatalf("steps %q, want %d steps", names, len(wantSteps)) | |
| 285 | + | } | |
| 286 | + | for i := range wantSteps { | |
| 287 | + | w, g := wantSteps[i], gotSteps[i] | |
| 288 | + | if g.Name != w.Name || g.Conclusion != w.Conclusion { | |
| 289 | + | t.Errorf("step %d: %q (%s), want %q (%s)", i+1, g.Name, g.Conclusion, w.Name, w.Conclusion) | |
| 290 | + | continue | |
| 291 | + | } | |
| 292 | + | if strings.Join(g.Log, "\n") != strings.Join(w.Log, "\n") { | |
| 293 | + | t.Errorf("step %q log differs:\n--- got\n%s\n--- want\n%s", w.Name, strings.Join(g.Log, "\n"), strings.Join(w.Log, "\n")) | |
| 294 | + | } | |
| 295 | + | } | |
| 296 | + | ||
| 297 | + | if scenario == "out-job-outputs" { | |
| 298 | + | checkProducerOutputs(t, got) | |
| 299 | + | } | |
| 300 | + | } | |
| 301 | + | ||
| 302 | + | // checkProducerOutputs compares the outputs of the producer job of | |
| 303 | + | // out-job-outputs with what the recorded consumer printed from `needs`. | |
| 304 | + | func checkProducerOutputs(t *testing.T, got hosttest.Result) { | |
| 305 | + | t.Helper() | |
| 306 | + | // The values the consumer printed on GitHub (recordings/out-job-outputs.json, | |
| 307 | + | // step "print needs outputs"): empty outputs are absent. | |
| 308 | + | want := map[string]string{ | |
| 309 | + | "composed": "hello-push", "from-env": "env-value", "from-failed": "written-before-failure", | |
| 310 | + | "overwritten": "second", "simple": "hello", "multi": "line one\nline two", "literal": "fixed-text", | |
| 311 | + | } | |
| 312 | + | if len(got.Complete.Outputs) != len(want) { | |
| 313 | + | t.Errorf("outputs %v, want %v", got.Complete.Outputs, want) | |
| 314 | + | } | |
| 315 | + | for k, v := range want { | |
| 316 | + | if got.Complete.Outputs[k] != v { | |
| 317 | + | t.Errorf("output %q = %q, want %q", k, got.Complete.Outputs[k], v) | |
| 318 | + | } | |
| 319 | + | } | |
| 320 | + | } | |
| @@ -0,0 +1,320 @@ | |||||
| 1 | + | // SPDX-License-Identifier: MIT | |||
| 2 | + | ||||
| 3 | + | package e2e | |||
| 4 | + | ||||
| 5 | + | import ( | |||
| 6 | + | "context" | |||
| 7 | + | "encoding/json" | |||
| 8 | + | "fmt" | |||
| 9 | + | "net/http" | |||
| 10 | + | "os" | |||
| 11 | + | "path/filepath" | |||
| 12 | + | "regexp" | |||
| 13 | + | "sort" | |||
| 14 | + | "strings" | |||
| 15 | + | "testing" | |||
| 16 | + | "time" | |||
| 17 | + | ||||
| 18 | + | ermite "bnema.dev/ermite-api/gen/ermite" | |||
| 19 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |||
| 20 | + | ||||
| 21 | + | "bnema.dev/ermite-runner/internal/host" | |||
| 22 | + | "bnema.dev/ermite-runner/internal/host/hosttest" | |||
| 23 | + | ) | |||
| 24 | + | ||||
| 25 | + | // The chain test runs the whole path of a job: a fake RunnerService hands out a | |||
| 26 | + | // corpus job, the real host loop starts a real sandbox (the runner image in a | |||
| 27 | + | // container), the real executor runs the steps, and the host reports to the | |||
| 28 | + | // fake with AppendLog, UpdateJob and CompleteJob. What the fake stored is | |||
| 29 | + | // compared with the recording of the same job made on github.com. | |||
| 30 | + | // | |||
| 31 | + | // It needs a checkout of ermite-conformance (corpus/ and recordings/), found | |||
| 32 | + | // through ERMITE_CONFORMANCE_DIR or as a sibling of this repository, and is | |||
| 33 | + | // read-only on it. | |||
| 34 | + | ||||
| 35 | + | func conformanceDir(t *testing.T) string { | |||
| 36 | + | t.Helper() | |||
| 37 | + | cands := []string{os.Getenv("ERMITE_CONFORMANCE_DIR")} | |||
| 38 | + | wd, _ := os.Getwd() | |||
| 39 | + | for d := wd; d != "/" && d != "."; d = filepath.Dir(d) { | |||
| 40 | + | cands = append(cands, filepath.Join(d, "ermite-conformance", ".worktrees", "corpus"), filepath.Join(d, "ermite-conformance")) | |||
| 41 | + | } | |||
| 42 | + | for _, c := range cands { | |||
| 43 | + | if c == "" { | |||
| 44 | + | continue | |||
| 45 | + | } | |||
| 46 | + | if st, err := os.Stat(filepath.Join(c, "recordings")); err == nil && st.IsDir() { | |||
| 47 | + | return c | |||
| 48 | + | } | |||
| 49 | + | } | |||
| 50 | + | t.Skip("ermite-conformance checkout not found (set ERMITE_CONFORMANCE_DIR)") | |||
| 51 | + | return "" | |||
| 52 | + | } | |||
| 53 | + | ||||
| 54 | + | type recStep struct { | |||
| 55 | + | Name string `json:"name"` | |||
| 56 | + | Conclusion string `json:"conclusion"` | |||
| 57 | + | Log []string `json:"log"` | |||
| 58 | + | } | |||
| 59 | + | ||||
| 60 | + | type recJob struct { | |||
| 61 | + | Name string `json:"name"` | |||
| 62 | + | Conclusion string `json:"conclusion"` | |||
| 63 | + | Steps []recStep `json:"steps"` | |||
| 64 | + | } | |||
| 65 | + | ||||
| 66 | + | type recording struct { | |||
| 67 | + | Pushes []struct { | |||
| 68 | + | Runs []struct { | |||
| 69 | + | Jobs []recJob `json:"jobs"` | |||
| 70 | + | } `json:"runs"` | |||
| 71 | + | } `json:"pushes"` | |||
| 72 | + | } | |||
| 73 | + | ||||
| 74 | + | var ( | |||
| 75 | + | uuidRe = regexp.MustCompile(`[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}`) | |||
| 76 | + | // Machine-dependent parts of the recordings, like the normalisation rules of | |||
| 77 | + | // ermite-conformance: random script names and the repository name. | |||
| 78 | + | repoRe = regexp.MustCompile(`/work/hello/hello`) | |||
| 79 | + | ) | |||
| 80 | + | ||||
| 81 | + | func normalize(l string) string { | |||
| 82 | + | l = repoRe.ReplaceAllString(l, "/work/<repo>/<repo>") | |||
| 83 | + | return uuidRe.ReplaceAllString(l, "<uuid>") | |||
| 84 | + | } | |||
| 85 | + | ||||
| 86 | + | func chainContexts(job string) string { | |||
| 87 | + | return fmt.Sprintf(`{"github":{"repository":"octo/hello","repository_owner":"octo","run_id":"1","run_number":"1","run_attempt":"1", | |||
| 88 | + | "ref":"refs/heads/scn/x/r1","ref_name":"scn/x/r1","ref_type":"branch","ref_protected":false,"sha":"0123456789abcdef0123456789abcdef01234567", | |||
| 89 | + | "event_name":"push","event":{"ref":"refs/heads/scn/x/r1"},"job":%q,"workflow":"w","actor":"octo","triggering_actor":"octo", | |||
| 90 | + | "server_url":"https://ermite.example","retention_days":"90"}}`, job) | |||
| 91 | + | } | |||
| 92 | + | ||||
| 93 | + | func TestChainCorpus(t *testing.T) { | |||
| 94 | + | image := needE2E(t) | |||
| 95 | + | dir := conformanceDir(t) | |||
| 96 | + | drv := newDriver(t, image) | |||
| 97 | + | ||||
| 98 | + | // scenario -> jobs to run (jobs with `needs` depend on the server and are left out) | |||
| 99 | + | scenarios := []struct { | |||
| 100 | + | name string | |||
| 101 | + | jobs []string | |||
| 102 | + | }{ | |||
| 103 | + | {"env-precedence", []string{"main", "other"}}, | |||
| 104 | + | {"env-expressions", []string{"main"}}, | |||
| 105 | + | {"out-github-env", []string{"main"}}, | |||
| 106 | + | {"out-github-env-reserved", []string{"node-options", "github-prefix"}}, | |||
| 107 | + | {"out-github-path", []string{"main"}}, | |||
| 108 | + | {"out-step-outputs", []string{"main"}}, | |||
| 109 | + | {"ctx-steps", []string{"main"}}, | |||
| 110 | + | {"if-step-conditions", []string{"main"}}, | |||
| 111 | + | {"shell-builtin", []string{"main"}}, | |||
| 112 | + | {"shell-custom", []string{"main"}}, | |||
| 113 | + | {"shell-exit-codes", nil}, // every job of the workflow | |||
| 114 | + | {"shell-working-directory", nil}, | |||
| 115 | + | {"out-job-outputs", []string{"producer"}}, | |||
| 116 | + | } | |||
| 117 | + | ||||
| 118 | + | srv := hosttest.New() | |||
| 119 | + | t.Cleanup(srv.Close) | |||
| 120 | + | ||||
| 121 | + | type want struct { | |||
| 122 | + | scenario string | |||
| 123 | + | job recJob | |||
| 124 | + | } | |||
| 125 | + | wants := map[string]want{} | |||
| 126 | + | var queue []*ermite.Job | |||
| 127 | + | for _, sc := range scenarios { | |||
| 128 | + | files, _ := filepath.Glob(filepath.Join(dir, "corpus", sc.name, "tree", ".github", "workflows", "*.yml")) | |||
| 129 | + | if len(files) != 1 { | |||
| 130 | + | t.Fatalf("%s: workflow files %v", sc.name, files) | |||
| 131 | + | } | |||
| 132 | + | yaml, err := os.ReadFile(files[0]) | |||
| 133 | + | if err != nil { | |||
| 134 | + | t.Fatal(err) | |||
| 135 | + | } | |||
| 136 | + | raw, err := os.ReadFile(filepath.Join(dir, "recordings", sc.name+".json")) | |||
| 137 | + | if err != nil { | |||
| 138 | + | t.Fatal(err) | |||
| 139 | + | } | |||
| 140 | + | var rec recording | |||
| 141 | + | if err := json.Unmarshal(raw, &rec); err != nil { | |||
| 142 | + | t.Fatal(err) | |||
| 143 | + | } | |||
| 144 | + | for _, rj := range rec.Pushes[0].Runs[0].Jobs { | |||
| 145 | + | if sc.jobs != nil && !contains(sc.jobs, rj.Name) { | |||
| 146 | + | continue | |||
| 147 | + | } | |||
| 148 | + | id := sc.name + "/" + rj.Name | |||
| 149 | + | queue = append(queue, &ermite.Job{ | |||
| 150 | + | JobId: id, RunId: 1, Attempt: 1, JobToken: "ermitejob_" + strings.ReplaceAll(id, "/", "_"), | |||
| 151 | + | WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: string(yaml), JobKey: rj.Name, Matrix: "{}", | |||
| 152 | + | Contexts: chainContexts(rj.Name), Repo: "hello", TimeoutMinutes: 10, | |||
| 153 | + | }) | |||
| 154 | + | wants[id] = want{sc.name, rj} | |||
| 155 | + | } | |||
| 156 | + | } | |||
| 157 | + | srv.Queue(queue...) | |||
| 158 | + | ||||
| 159 | + | r := &host.Runner{ | |||
| 160 | + | Cfg: host.Config{URL: srv.URL, Name: "chain", Capacity: 2, Version: "test"}, | |||
| 161 | + | Driver: drv, | |||
| 162 | + | Poll: host.NewClient(srv.URL, hosttest.RunnerToken, http.DefaultTransport), | |||
| 163 | + | Job: func(tok string) ermiteconnect.RunnerServiceClient { | |||
| 164 | + | return host.NewClient(srv.URL, tok, http.DefaultTransport) | |||
| 165 | + | }, | |||
| 166 | + | Timing: host.Timing{LogFlush: 20 * time.Millisecond, NoJobPause: 100 * time.Millisecond}, | |||
| 167 | + | Logf: func(f string, a ...any) { t.Logf(f, a...) }, | |||
| 168 | + | } | |||
| 169 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 170 | + | done := make(chan error, 1) | |||
| 171 | + | go func() { done <- r.Run(ctx) }() | |||
| 172 | + | defer func() { | |||
| 173 | + | cancel() | |||
| 174 | + | select { | |||
| 175 | + | case err := <-done: | |||
| 176 | + | if err != nil { | |||
| 177 | + | t.Errorf("Run: %v", err) | |||
| 178 | + | } | |||
| 179 | + | case <-time.After(60 * time.Second): | |||
| 180 | + | t.Error("Run did not stop") | |||
| 181 | + | } | |||
| 182 | + | }() | |||
| 183 | + | ||||
| 184 | + | deadline := time.After(5 * time.Minute) | |||
| 185 | + | for n := 0; n < len(queue); n++ { | |||
| 186 | + | select { | |||
| 187 | + | case <-srv.Done(): | |||
| 188 | + | case <-deadline: | |||
| 189 | + | t.Fatalf("only %d of %d jobs finished", n, len(queue)) | |||
| 190 | + | } | |||
| 191 | + | } | |||
| 192 | + | ||||
| 193 | + | var ids []string | |||
| 194 | + | for id := range wants { | |||
| 195 | + | ids = append(ids, id) | |||
| 196 | + | } | |||
| 197 | + | sort.Strings(ids) | |||
| 198 | + | for _, id := range ids { | |||
| 199 | + | w := wants[id] | |||
| 200 | + | t.Run(id, func(t *testing.T) { compareJob(t, srv.Result(id), w.job, w.scenario) }) | |||
| 201 | + | } | |||
| 202 | + | for _, e := range srv.Violations() { | |||
| 203 | + | t.Errorf("the fake server saw a contract violation: %s", e) | |||
| 204 | + | } | |||
| 205 | + | } | |||
| 206 | + | ||||
| 207 | + | func contains(l []string, s string) bool { | |||
| 208 | + | for _, x := range l { | |||
| 209 | + | if x == s { | |||
| 210 | + | return true | |||
| 211 | + | } | |||
| 212 | + | } | |||
| 213 | + | return false | |||
| 214 | + | } | |||
| 215 | + | ||||
| 216 | + | // compareJob checks the steps, conclusions, outputs and log lines reported for | |||
| 217 | + | // a job against the recording. "Set up job" (runner and image details) and the | |||
| 218 | + | // ermite-dump step that the harness appends to every job are not compared. | |||
| 219 | + | func compareJob(t *testing.T, got hosttest.Result, want recJob, scenario string) { | |||
| 220 | + | if got.Complete == nil { | |||
| 221 | + | t.Fatal("the job did not complete") | |||
| 222 | + | } | |||
| 223 | + | // The conclusion of the job as GitHub recorded it, except where the dump | |||
| 224 | + | // step failed on GitHub (it only fails after a failed step in some | |||
| 225 | + | // scenarios and is not part of the job here). | |||
| 226 | + | wantConcl := map[string]ermite.CiConclusion{ | |||
| 227 | + | "success": ermite.CiConclusion_CI_CONCLUSION_SUCCESS, "failure": ermite.CiConclusion_CI_CONCLUSION_FAILURE, | |||
| 228 | + | "cancelled": ermite.CiConclusion_CI_CONCLUSION_CANCELLED, | |||
| 229 | + | } | |||
| 230 | + | jobWant := wantConcl[want.Conclusion] | |||
| 231 | + | for _, s := range want.Steps { | |||
| 232 | + | if s.Name == "ermite-dump" && s.Conclusion != "success" && want.Conclusion == "failure" { | |||
| 233 | + | // the job failed only because of the dump step | |||
| 234 | + | jobWant = ermite.CiConclusion_CI_CONCLUSION_SUCCESS | |||
| 235 | + | for _, o := range want.Steps { | |||
| 236 | + | if o.Name != "ermite-dump" && o.Conclusion == "failure" { | |||
| 237 | + | jobWant = ermite.CiConclusion_CI_CONCLUSION_FAILURE | |||
| 238 | + | } | |||
| 239 | + | } | |||
| 240 | + | } | |||
| 241 | + | } | |||
| 242 | + | if got.Complete.Conclusion != jobWant { | |||
| 243 | + | t.Errorf("job conclusion %v, want %v", got.Complete.Conclusion, jobWant) | |||
| 244 | + | } | |||
| 245 | + | ||||
| 246 | + | // Steps by number, with their lines. | |||
| 247 | + | byStep := map[int32][]string{} | |||
| 248 | + | for _, l := range got.Lines { | |||
| 249 | + | byStep[l.Step] = append(byStep[l.Step], normalize(l.Text)) | |||
| 250 | + | } | |||
| 251 | + | var steps []int32 | |||
| 252 | + | for n := range got.Steps { | |||
| 253 | + | steps = append(steps, n) | |||
| 254 | + | } | |||
| 255 | + | sort.Slice(steps, func(i, j int) bool { return steps[i] < steps[j] }) | |||
| 256 | + | ||||
| 257 | + | type es = recStep | |||
| 258 | + | var wantSteps, gotSteps []es | |||
| 259 | + | for _, s := range want.Steps { | |||
| 260 | + | if s.Name == "Set up job" || s.Name == "ermite-dump" { | |||
| 261 | + | continue | |||
| 262 | + | } | |||
| 263 | + | wantSteps = append(wantSteps, s) | |||
| 264 | + | } | |||
| 265 | + | conc := map[ermite.CiConclusion]string{ | |||
| 266 | + | ermite.CiConclusion_CI_CONCLUSION_SUCCESS: "success", ermite.CiConclusion_CI_CONCLUSION_FAILURE: "failure", | |||
| 267 | + | ermite.CiConclusion_CI_CONCLUSION_CANCELLED: "cancelled", ermite.CiConclusion_CI_CONCLUSION_SKIPPED: "skipped", | |||
| 268 | + | } | |||
| 269 | + | for _, n := range steps { | |||
| 270 | + | u := got.Steps[n] | |||
| 271 | + | if u.Name == "Set up job" { | |||
| 272 | + | continue | |||
| 273 | + | } | |||
| 274 | + | if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.StartedAt == nil || u.CompletedAt == nil { | |||
| 275 | + | t.Errorf("step %d %q is not a complete snapshot: %v", n, u.Name, u) | |||
| 276 | + | } | |||
| 277 | + | gotSteps = append(gotSteps, es{Name: u.Name, Conclusion: conc[u.Conclusion], Log: byStep[n]}) | |||
| 278 | + | } | |||
| 279 | + | if len(gotSteps) != len(wantSteps) { | |||
| 280 | + | var names []string | |||
| 281 | + | for _, s := range gotSteps { | |||
| 282 | + | names = append(names, s.Name) | |||
| 283 | + | } | |||
| 284 | + | t.Fatalf("steps %q, want %d steps", names, len(wantSteps)) | |||
| 285 | + | } | |||
| 286 | + | for i := range wantSteps { | |||
| 287 | + | w, g := wantSteps[i], gotSteps[i] | |||
| 288 | + | if g.Name != w.Name || g.Conclusion != w.Conclusion { | |||
| 289 | + | t.Errorf("step %d: %q (%s), want %q (%s)", i+1, g.Name, g.Conclusion, w.Name, w.Conclusion) | |||
| 290 | + | continue | |||
| 291 | + | } | |||
| 292 | + | if strings.Join(g.Log, "\n") != strings.Join(w.Log, "\n") { | |||
| 293 | + | t.Errorf("step %q log differs:\n--- got\n%s\n--- want\n%s", w.Name, strings.Join(g.Log, "\n"), strings.Join(w.Log, "\n")) | |||
| 294 | + | } | |||
| 295 | + | } | |||
| 296 | + | ||||
| 297 | + | if scenario == "out-job-outputs" { | |||
| 298 | + | checkProducerOutputs(t, got) | |||
| 299 | + | } | |||
| 300 | + | } | |||
| 301 | + | ||||
| 302 | + | // checkProducerOutputs compares the outputs of the producer job of | |||
| 303 | + | // out-job-outputs with what the recorded consumer printed from `needs`. | |||
| 304 | + | func checkProducerOutputs(t *testing.T, got hosttest.Result) { | |||
| 305 | + | t.Helper() | |||
| 306 | + | // The values the consumer printed on GitHub (recordings/out-job-outputs.json, | |||
| 307 | + | // step "print needs outputs"): empty outputs are absent. | |||
| 308 | + | want := map[string]string{ | |||
| 309 | + | "composed": "hello-push", "from-env": "env-value", "from-failed": "written-before-failure", | |||
| 310 | + | "overwritten": "second", "simple": "hello", "multi": "line one\nline two", "literal": "fixed-text", | |||
| 311 | + | } | |||
| 312 | + | if len(got.Complete.Outputs) != len(want) { | |||
| 313 | + | t.Errorf("outputs %v, want %v", got.Complete.Outputs, want) | |||
| 314 | + | } | |||
| 315 | + | for k, v := range want { | |||
| 316 | + | if got.Complete.Outputs[k] != v { | |||
| 317 | + | t.Errorf("output %q = %q, want %q", k, got.Complete.Outputs[k], v) | |||
| 318 | + | } | |||
| 319 | + | } | |||
| 320 | + | } | |||
A internal/host/hosttest/hosttest.go +256 -0
| @@ -0,0 +1,256 @@ | |||
| 1 | + | // SPDX-License-Identifier: MIT | |
| 2 | + | ||
| 3 | + | // Package hosttest is an in-memory RunnerService for tests that run the real | |
| 4 | + | // host loop against a real sandbox. It enforces the rules of runner.proto that | |
| 5 | + | // matter to a client: tokens per kind and per job, the offsets of AppendLog, | |
| 6 | + | // the limits of a request, and the end of a job after CompleteJob. | |
| 7 | + | package hosttest | |
| 8 | + | ||
| 9 | + | import ( | |
| 10 | + | "context" | |
| 11 | + | "io" | |
| 12 | + | "net/http" | |
| 13 | + | "net/http/httptest" | |
| 14 | + | "strings" | |
| 15 | + | "sync" | |
| 16 | + | ||
| 17 | + | "connectrpc.com/connect" | |
| 18 | + | ||
| 19 | + | ermite "bnema.dev/ermite-api/gen/ermite" | |
| 20 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |
| 21 | + | ) | |
| 22 | + | ||
| 23 | + | // RunnerToken is the runner token the server accepts. | |
| 24 | + | const RunnerToken = "ermiterun_test" | |
| 25 | + | ||
| 26 | + | // Line is one stored log line. | |
| 27 | + | type Line struct { | |
| 28 | + | Step int32 | |
| 29 | + | Text string | |
| 30 | + | Continued bool | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | // Result is what a job reported. | |
| 34 | + | type Result struct { | |
| 35 | + | Complete *ermite.CompleteJobRequest | |
| 36 | + | Lines []Line | |
| 37 | + | Steps map[int32]*ermite.StepUpdate | |
| 38 | + | } | |
| 39 | + | ||
| 40 | + | // Server is the fake. Create it with New; URL is where to point the runner. | |
| 41 | + | type Server struct { | |
| 42 | + | ermiteconnect.UnimplementedRunnerServiceHandler | |
| 43 | + | ||
| 44 | + | URL string | |
| 45 | + | srv *httptest.Server | |
| 46 | + | ||
| 47 | + | mu sync.Mutex | |
| 48 | + | queue []*ermite.Job | |
| 49 | + | jobs map[string]*jobState | |
| 50 | + | assigned map[string]*ermite.Job // by job token | |
| 51 | + | // Errors lists contract violations seen (a gap in AppendLog, a request over | |
| 52 | + | // the limits, a call after the end): a test fails when it is not empty. | |
| 53 | + | Errors []string | |
| 54 | + | done chan string | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | type jobState struct { | |
| 58 | + | lines []Line | |
| 59 | + | steps map[int32]*ermite.StepUpdate | |
| 60 | + | complete *ermite.CompleteJobRequest | |
| 61 | + | } | |
| 62 | + | ||
| 63 | + | // New starts the server; Close stops it. | |
| 64 | + | func New() *Server { | |
| 65 | + | s := &Server{jobs: map[string]*jobState{}, assigned: map[string]*ermite.Job{}, done: make(chan string, 256)} | |
| 66 | + | _, h := ermiteconnect.NewRunnerServiceHandler(s) | |
| 67 | + | s.srv = httptest.NewServer(h) | |
| 68 | + | s.URL = s.srv.URL | |
| 69 | + | return s | |
| 70 | + | } | |
| 71 | + | ||
| 72 | + | // Close stops the server. | |
| 73 | + | func (s *Server) Close() { s.srv.Close() } | |
| 74 | + | ||
| 75 | + | // Queue adds jobs to hand out, in order. | |
| 76 | + | func (s *Server) Queue(jobs ...*ermite.Job) { | |
| 77 | + | s.mu.Lock() | |
| 78 | + | defer s.mu.Unlock() | |
| 79 | + | s.queue = append(s.queue, jobs...) | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | // Violations returns the contract violations seen so far. | |
| 83 | + | func (s *Server) Violations() []string { | |
| 84 | + | s.mu.Lock() | |
| 85 | + | defer s.mu.Unlock() | |
| 86 | + | return append([]string(nil), s.Errors...) | |
| 87 | + | } | |
| 88 | + | ||
| 89 | + | // Done receives the id of each job when its CompleteJob has been applied. | |
| 90 | + | func (s *Server) Done() <-chan string { return s.done } | |
| 91 | + | ||
| 92 | + | // Result returns what a job reported so far. | |
| 93 | + | func (s *Server) Result(jobID string) Result { | |
| 94 | + | s.mu.Lock() | |
| 95 | + | defer s.mu.Unlock() | |
| 96 | + | st := s.jobs[jobID] | |
| 97 | + | if st == nil { | |
| 98 | + | return Result{} | |
| 99 | + | } | |
| 100 | + | steps := map[int32]*ermite.StepUpdate{} | |
| 101 | + | for k, v := range st.steps { | |
| 102 | + | steps[k] = v | |
| 103 | + | } | |
| 104 | + | return Result{Complete: st.complete, Lines: append([]Line(nil), st.lines...), Steps: steps} | |
| 105 | + | } | |
| 106 | + | ||
| 107 | + | func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } | |
| 108 | + | ||
| 109 | + | func unauth() error { return connect.NewError(connect.CodeUnauthenticated, io.EOF) } | |
| 110 | + | ||
| 111 | + | func (s *Server) job(h http.Header, id string) (*jobState, error) { | |
| 112 | + | j := s.assigned[strings.TrimPrefix(h.Get("Authorization"), "Bearer ")] | |
| 113 | + | if j == nil { | |
| 114 | + | return nil, unauth() | |
| 115 | + | } | |
| 116 | + | if j.JobId != id { | |
| 117 | + | return nil, connect.NewError(connect.CodePermissionDenied, io.EOF) | |
| 118 | + | } | |
| 119 | + | return s.jobs[id], nil | |
| 120 | + | } | |
| 121 | + | ||
| 122 | + | func ended(c ermite.CiConclusion) error { | |
| 123 | + | e := connect.NewError(connect.CodeFailedPrecondition, io.EOF) | |
| 124 | + | if d, err := connect.NewErrorDetail(&ermite.JobEnded{Conclusion: c}); err == nil { | |
| 125 | + | e.AddDetail(d) | |
| 126 | + | } | |
| 127 | + | return e | |
| 128 | + | } | |
| 129 | + | ||
| 130 | + | // FetchJob hands out the queued jobs. | |
| 131 | + | func (s *Server) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) { | |
| 132 | + | s.mu.Lock() | |
| 133 | + | defer s.mu.Unlock() | |
| 134 | + | if r.Header().Get("Authorization") != "Bearer "+RunnerToken { | |
| 135 | + | return nil, unauth() | |
| 136 | + | } | |
| 137 | + | if len(s.queue) == 0 { | |
| 138 | + | return connect.NewResponse(&ermite.FetchJobResponse{}), nil | |
| 139 | + | } | |
| 140 | + | j := s.queue[0] | |
| 141 | + | s.queue = s.queue[1:] | |
| 142 | + | s.assigned[j.JobToken] = j | |
| 143 | + | s.jobs[j.JobId] = &jobState{steps: map[int32]*ermite.StepUpdate{}} | |
| 144 | + | return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil | |
| 145 | + | } | |
| 146 | + | ||
| 147 | + | // Heartbeat accepts heartbeats of running jobs. | |
| 148 | + | func (s *Server) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) { | |
| 149 | + | s.mu.Lock() | |
| 150 | + | defer s.mu.Unlock() | |
| 151 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |
| 152 | + | if err != nil { | |
| 153 | + | return nil, err | |
| 154 | + | } | |
| 155 | + | if st.complete != nil { | |
| 156 | + | return nil, ended(st.complete.Conclusion) | |
| 157 | + | } | |
| 158 | + | return connect.NewResponse(&ermite.HeartbeatResponse{}), nil | |
| 159 | + | } | |
| 160 | + | ||
| 161 | + | // UpdateJob stores step snapshots. | |
| 162 | + | func (s *Server) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) { | |
| 163 | + | s.mu.Lock() | |
| 164 | + | defer s.mu.Unlock() | |
| 165 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |
| 166 | + | if err != nil { | |
| 167 | + | return nil, err | |
| 168 | + | } | |
| 169 | + | if st.complete != nil { | |
| 170 | + | return nil, ended(st.complete.Conclusion) | |
| 171 | + | } | |
| 172 | + | for _, u := range r.Msg.Steps { | |
| 173 | + | if old := st.steps[u.Number]; old != nil && old.State == ermite.StepState_STEP_STATE_COMPLETED { | |
| 174 | + | s.violation("UpdateJob: a completed step changed") | |
| 175 | + | } | |
| 176 | + | st.steps[u.Number] = u | |
| 177 | + | } | |
| 178 | + | return connect.NewResponse(&ermite.UpdateJobResponse{}), nil | |
| 179 | + | } | |
| 180 | + | ||
| 181 | + | // AppendLog applies the offset rules of runner.proto. | |
| 182 | + | func (s *Server) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) { | |
| 183 | + | s.mu.Lock() | |
| 184 | + | defer s.mu.Unlock() | |
| 185 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |
| 186 | + | if err != nil { | |
| 187 | + | return nil, err | |
| 188 | + | } | |
| 189 | + | if st.complete != nil { | |
| 190 | + | s.violation("AppendLog after CompleteJob") | |
| 191 | + | return nil, ended(st.complete.Conclusion) | |
| 192 | + | } | |
| 193 | + | n, bytes := len(r.Msg.Lines), 0 | |
| 194 | + | if n == 0 || n > 1000 { | |
| 195 | + | s.violation("AppendLog: bad number of lines") | |
| 196 | + | return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF) | |
| 197 | + | } | |
| 198 | + | for i, l := range r.Msg.Lines { | |
| 199 | + | bytes += len(l.Text) | |
| 200 | + | if strings.Contains(l.Text, "\n") || l.Time == nil || (l.Continued && i == 0 && len(st.lines) == 0 && r.Msg.Offset == 0) { | |
| 201 | + | s.violation("AppendLog: invalid line") | |
| 202 | + | return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF) | |
| 203 | + | } | |
| 204 | + | } | |
| 205 | + | if bytes > 64*1024 { | |
| 206 | + | s.violation("AppendLog: more than 64 KiB") | |
| 207 | + | return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF) | |
| 208 | + | } | |
| 209 | + | have := int64(len(st.lines)) | |
| 210 | + | if r.Msg.Offset > have { | |
| 211 | + | s.violation("AppendLog: a gap in the offsets") | |
| 212 | + | return nil, connect.NewError(connect.CodeOutOfRange, io.EOF) | |
| 213 | + | } | |
| 214 | + | if have > 0 && r.Msg.Offset == have && r.Msg.Step < st.lines[have-1].Step { | |
| 215 | + | s.violation("AppendLog: a step went back") | |
| 216 | + | return nil, connect.NewError(connect.CodeFailedPrecondition, io.EOF) | |
| 217 | + | } | |
| 218 | + | for i, l := range r.Msg.Lines { | |
| 219 | + | if int64(i) < have-r.Msg.Offset { | |
| 220 | + | continue | |
| 221 | + | } | |
| 222 | + | st.lines = append(st.lines, Line{Step: r.Msg.Step, Text: l.Text, Continued: l.Continued}) | |
| 223 | + | } | |
| 224 | + | return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(st.lines))}), nil | |
| 225 | + | } | |
| 226 | + | ||
| 227 | + | // CompleteJob ends a job, idempotently. | |
| 228 | + | func (s *Server) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) { | |
| 229 | + | s.mu.Lock() | |
| 230 | + | defer s.mu.Unlock() | |
| 231 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |
| 232 | + | if err != nil { | |
| 233 | + | return nil, err | |
| 234 | + | } | |
| 235 | + | if st.complete != nil { | |
| 236 | + | return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: st.complete.Conclusion}), nil | |
| 237 | + | } | |
| 238 | + | st.complete = r.Msg | |
| 239 | + | for _, u := range r.Msg.Steps { | |
| 240 | + | st.steps[u.Number] = u | |
| 241 | + | } | |
| 242 | + | s.done <- r.Msg.JobId | |
| 243 | + | return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil | |
| 244 | + | } | |
| 245 | + | ||
| 246 | + | // WatchJob keeps the stream open until the job ends or the client leaves. | |
| 247 | + | func (s *Server) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], _ *connect.ServerStream[ermite.WatchJobResponse]) error { | |
| 248 | + | s.mu.Lock() | |
| 249 | + | _, err := s.job(r.Header(), r.Msg.JobId) | |
| 250 | + | s.mu.Unlock() | |
| 251 | + | if err != nil { | |
| 252 | + | return err | |
| 253 | + | } | |
| 254 | + | <-ctx.Done() | |
| 255 | + | return ctx.Err() | |
| 256 | + | } | |
| @@ -0,0 +1,256 @@ | |||||
| 1 | + | // SPDX-License-Identifier: MIT | |||
| 2 | + | ||||
| 3 | + | // Package hosttest is an in-memory RunnerService for tests that run the real | |||
| 4 | + | // host loop against a real sandbox. It enforces the rules of runner.proto that | |||
| 5 | + | // matter to a client: tokens per kind and per job, the offsets of AppendLog, | |||
| 6 | + | // the limits of a request, and the end of a job after CompleteJob. | |||
| 7 | + | package hosttest | |||
| 8 | + | ||||
| 9 | + | import ( | |||
| 10 | + | "context" | |||
| 11 | + | "io" | |||
| 12 | + | "net/http" | |||
| 13 | + | "net/http/httptest" | |||
| 14 | + | "strings" | |||
| 15 | + | "sync" | |||
| 16 | + | ||||
| 17 | + | "connectrpc.com/connect" | |||
| 18 | + | ||||
| 19 | + | ermite "bnema.dev/ermite-api/gen/ermite" | |||
| 20 | + | "bnema.dev/ermite-api/gen/ermite/ermiteconnect" | |||
| 21 | + | ) | |||
| 22 | + | ||||
| 23 | + | // RunnerToken is the runner token the server accepts. | |||
| 24 | + | const RunnerToken = "ermiterun_test" | |||
| 25 | + | ||||
| 26 | + | // Line is one stored log line. | |||
| 27 | + | type Line struct { | |||
| 28 | + | Step int32 | |||
| 29 | + | Text string | |||
| 30 | + | Continued bool | |||
| 31 | + | } | |||
| 32 | + | ||||
| 33 | + | // Result is what a job reported. | |||
| 34 | + | type Result struct { | |||
| 35 | + | Complete *ermite.CompleteJobRequest | |||
| 36 | + | Lines []Line | |||
| 37 | + | Steps map[int32]*ermite.StepUpdate | |||
| 38 | + | } | |||
| 39 | + | ||||
| 40 | + | // Server is the fake. Create it with New; URL is where to point the runner. | |||
| 41 | + | type Server struct { | |||
| 42 | + | ermiteconnect.UnimplementedRunnerServiceHandler | |||
| 43 | + | ||||
| 44 | + | URL string | |||
| 45 | + | srv *httptest.Server | |||
| 46 | + | ||||
| 47 | + | mu sync.Mutex | |||
| 48 | + | queue []*ermite.Job | |||
| 49 | + | jobs map[string]*jobState | |||
| 50 | + | assigned map[string]*ermite.Job // by job token | |||
| 51 | + | // Errors lists contract violations seen (a gap in AppendLog, a request over | |||
| 52 | + | // the limits, a call after the end): a test fails when it is not empty. | |||
| 53 | + | Errors []string | |||
| 54 | + | done chan string | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | type jobState struct { | |||
| 58 | + | lines []Line | |||
| 59 | + | steps map[int32]*ermite.StepUpdate | |||
| 60 | + | complete *ermite.CompleteJobRequest | |||
| 61 | + | } | |||
| 62 | + | ||||
| 63 | + | // New starts the server; Close stops it. | |||
| 64 | + | func New() *Server { | |||
| 65 | + | s := &Server{jobs: map[string]*jobState{}, assigned: map[string]*ermite.Job{}, done: make(chan string, 256)} | |||
| 66 | + | _, h := ermiteconnect.NewRunnerServiceHandler(s) | |||
| 67 | + | s.srv = httptest.NewServer(h) | |||
| 68 | + | s.URL = s.srv.URL | |||
| 69 | + | return s | |||
| 70 | + | } | |||
| 71 | + | ||||
| 72 | + | // Close stops the server. | |||
| 73 | + | func (s *Server) Close() { s.srv.Close() } | |||
| 74 | + | ||||
| 75 | + | // Queue adds jobs to hand out, in order. | |||
| 76 | + | func (s *Server) Queue(jobs ...*ermite.Job) { | |||
| 77 | + | s.mu.Lock() | |||
| 78 | + | defer s.mu.Unlock() | |||
| 79 | + | s.queue = append(s.queue, jobs...) | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | // Violations returns the contract violations seen so far. | |||
| 83 | + | func (s *Server) Violations() []string { | |||
| 84 | + | s.mu.Lock() | |||
| 85 | + | defer s.mu.Unlock() | |||
| 86 | + | return append([]string(nil), s.Errors...) | |||
| 87 | + | } | |||
| 88 | + | ||||
| 89 | + | // Done receives the id of each job when its CompleteJob has been applied. | |||
| 90 | + | func (s *Server) Done() <-chan string { return s.done } | |||
| 91 | + | ||||
| 92 | + | // Result returns what a job reported so far. | |||
| 93 | + | func (s *Server) Result(jobID string) Result { | |||
| 94 | + | s.mu.Lock() | |||
| 95 | + | defer s.mu.Unlock() | |||
| 96 | + | st := s.jobs[jobID] | |||
| 97 | + | if st == nil { | |||
| 98 | + | return Result{} | |||
| 99 | + | } | |||
| 100 | + | steps := map[int32]*ermite.StepUpdate{} | |||
| 101 | + | for k, v := range st.steps { | |||
| 102 | + | steps[k] = v | |||
| 103 | + | } | |||
| 104 | + | return Result{Complete: st.complete, Lines: append([]Line(nil), st.lines...), Steps: steps} | |||
| 105 | + | } | |||
| 106 | + | ||||
| 107 | + | func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } | |||
| 108 | + | ||||
| 109 | + | func unauth() error { return connect.NewError(connect.CodeUnauthenticated, io.EOF) } | |||
| 110 | + | ||||
| 111 | + | func (s *Server) job(h http.Header, id string) (*jobState, error) { | |||
| 112 | + | j := s.assigned[strings.TrimPrefix(h.Get("Authorization"), "Bearer ")] | |||
| 113 | + | if j == nil { | |||
| 114 | + | return nil, unauth() | |||
| 115 | + | } | |||
| 116 | + | if j.JobId != id { | |||
| 117 | + | return nil, connect.NewError(connect.CodePermissionDenied, io.EOF) | |||
| 118 | + | } | |||
| 119 | + | return s.jobs[id], nil | |||
| 120 | + | } | |||
| 121 | + | ||||
| 122 | + | func ended(c ermite.CiConclusion) error { | |||
| 123 | + | e := connect.NewError(connect.CodeFailedPrecondition, io.EOF) | |||
| 124 | + | if d, err := connect.NewErrorDetail(&ermite.JobEnded{Conclusion: c}); err == nil { | |||
| 125 | + | e.AddDetail(d) | |||
| 126 | + | } | |||
| 127 | + | return e | |||
| 128 | + | } | |||
| 129 | + | ||||
| 130 | + | // FetchJob hands out the queued jobs. | |||
| 131 | + | func (s *Server) FetchJob(_ context.Context, r *connect.Request[ermite.FetchJobRequest]) (*connect.Response[ermite.FetchJobResponse], error) { | |||
| 132 | + | s.mu.Lock() | |||
| 133 | + | defer s.mu.Unlock() | |||
| 134 | + | if r.Header().Get("Authorization") != "Bearer "+RunnerToken { | |||
| 135 | + | return nil, unauth() | |||
| 136 | + | } | |||
| 137 | + | if len(s.queue) == 0 { | |||
| 138 | + | return connect.NewResponse(&ermite.FetchJobResponse{}), nil | |||
| 139 | + | } | |||
| 140 | + | j := s.queue[0] | |||
| 141 | + | s.queue = s.queue[1:] | |||
| 142 | + | s.assigned[j.JobToken] = j | |||
| 143 | + | s.jobs[j.JobId] = &jobState{steps: map[int32]*ermite.StepUpdate{}} | |||
| 144 | + | return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil | |||
| 145 | + | } | |||
| 146 | + | ||||
| 147 | + | // Heartbeat accepts heartbeats of running jobs. | |||
| 148 | + | func (s *Server) Heartbeat(_ context.Context, r *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error) { | |||
| 149 | + | s.mu.Lock() | |||
| 150 | + | defer s.mu.Unlock() | |||
| 151 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |||
| 152 | + | if err != nil { | |||
| 153 | + | return nil, err | |||
| 154 | + | } | |||
| 155 | + | if st.complete != nil { | |||
| 156 | + | return nil, ended(st.complete.Conclusion) | |||
| 157 | + | } | |||
| 158 | + | return connect.NewResponse(&ermite.HeartbeatResponse{}), nil | |||
| 159 | + | } | |||
| 160 | + | ||||
| 161 | + | // UpdateJob stores step snapshots. | |||
| 162 | + | func (s *Server) UpdateJob(_ context.Context, r *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error) { | |||
| 163 | + | s.mu.Lock() | |||
| 164 | + | defer s.mu.Unlock() | |||
| 165 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |||
| 166 | + | if err != nil { | |||
| 167 | + | return nil, err | |||
| 168 | + | } | |||
| 169 | + | if st.complete != nil { | |||
| 170 | + | return nil, ended(st.complete.Conclusion) | |||
| 171 | + | } | |||
| 172 | + | for _, u := range r.Msg.Steps { | |||
| 173 | + | if old := st.steps[u.Number]; old != nil && old.State == ermite.StepState_STEP_STATE_COMPLETED { | |||
| 174 | + | s.violation("UpdateJob: a completed step changed") | |||
| 175 | + | } | |||
| 176 | + | st.steps[u.Number] = u | |||
| 177 | + | } | |||
| 178 | + | return connect.NewResponse(&ermite.UpdateJobResponse{}), nil | |||
| 179 | + | } | |||
| 180 | + | ||||
| 181 | + | // AppendLog applies the offset rules of runner.proto. | |||
| 182 | + | func (s *Server) AppendLog(_ context.Context, r *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) { | |||
| 183 | + | s.mu.Lock() | |||
| 184 | + | defer s.mu.Unlock() | |||
| 185 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |||
| 186 | + | if err != nil { | |||
| 187 | + | return nil, err | |||
| 188 | + | } | |||
| 189 | + | if st.complete != nil { | |||
| 190 | + | s.violation("AppendLog after CompleteJob") | |||
| 191 | + | return nil, ended(st.complete.Conclusion) | |||
| 192 | + | } | |||
| 193 | + | n, bytes := len(r.Msg.Lines), 0 | |||
| 194 | + | if n == 0 || n > 1000 { | |||
| 195 | + | s.violation("AppendLog: bad number of lines") | |||
| 196 | + | return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF) | |||
| 197 | + | } | |||
| 198 | + | for i, l := range r.Msg.Lines { | |||
| 199 | + | bytes += len(l.Text) | |||
| 200 | + | if strings.Contains(l.Text, "\n") || l.Time == nil || (l.Continued && i == 0 && len(st.lines) == 0 && r.Msg.Offset == 0) { | |||
| 201 | + | s.violation("AppendLog: invalid line") | |||
| 202 | + | return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF) | |||
| 203 | + | } | |||
| 204 | + | } | |||
| 205 | + | if bytes > 64*1024 { | |||
| 206 | + | s.violation("AppendLog: more than 64 KiB") | |||
| 207 | + | return nil, connect.NewError(connect.CodeInvalidArgument, io.EOF) | |||
| 208 | + | } | |||
| 209 | + | have := int64(len(st.lines)) | |||
| 210 | + | if r.Msg.Offset > have { | |||
| 211 | + | s.violation("AppendLog: a gap in the offsets") | |||
| 212 | + | return nil, connect.NewError(connect.CodeOutOfRange, io.EOF) | |||
| 213 | + | } | |||
| 214 | + | if have > 0 && r.Msg.Offset == have && r.Msg.Step < st.lines[have-1].Step { | |||
| 215 | + | s.violation("AppendLog: a step went back") | |||
| 216 | + | return nil, connect.NewError(connect.CodeFailedPrecondition, io.EOF) | |||
| 217 | + | } | |||
| 218 | + | for i, l := range r.Msg.Lines { | |||
| 219 | + | if int64(i) < have-r.Msg.Offset { | |||
| 220 | + | continue | |||
| 221 | + | } | |||
| 222 | + | st.lines = append(st.lines, Line{Step: r.Msg.Step, Text: l.Text, Continued: l.Continued}) | |||
| 223 | + | } | |||
| 224 | + | return connect.NewResponse(&ermite.AppendLogResponse{NextOffset: int64(len(st.lines))}), nil | |||
| 225 | + | } | |||
| 226 | + | ||||
| 227 | + | // CompleteJob ends a job, idempotently. | |||
| 228 | + | func (s *Server) CompleteJob(_ context.Context, r *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error) { | |||
| 229 | + | s.mu.Lock() | |||
| 230 | + | defer s.mu.Unlock() | |||
| 231 | + | st, err := s.job(r.Header(), r.Msg.JobId) | |||
| 232 | + | if err != nil { | |||
| 233 | + | return nil, err | |||
| 234 | + | } | |||
| 235 | + | if st.complete != nil { | |||
| 236 | + | return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: st.complete.Conclusion}), nil | |||
| 237 | + | } | |||
| 238 | + | st.complete = r.Msg | |||
| 239 | + | for _, u := range r.Msg.Steps { | |||
| 240 | + | st.steps[u.Number] = u | |||
| 241 | + | } | |||
| 242 | + | s.done <- r.Msg.JobId | |||
| 243 | + | return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil | |||
| 244 | + | } | |||
| 245 | + | ||||
| 246 | + | // WatchJob keeps the stream open until the job ends or the client leaves. | |||
| 247 | + | func (s *Server) WatchJob(ctx context.Context, r *connect.Request[ermite.WatchJobRequest], _ *connect.ServerStream[ermite.WatchJobResponse]) error { | |||
| 248 | + | s.mu.Lock() | |||
| 249 | + | _, err := s.job(r.Header(), r.Msg.JobId) | |||
| 250 | + | s.mu.Unlock() | |||
| 251 | + | if err != nil { | |||
| 252 | + | return err | |||
| 253 | + | } | |||
| 254 | + | <-ctx.Done() | |||
| 255 | + | return ctx.Err() | |||
| 256 | + | } | |||