From b712c91c2d562538c837a05ea0202094d6a18c93 Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Wed, 28 Oct 2026 04:32:50 +0000 Subject: [PATCH] test(e2e): cover host mode, reusable workflows, scheduling, reconnect and CLI flags (#1347) Extends the e2e suite with scenarios for host labels, local `workflow_call`, label routing, `runner.capacity`, a dropped and restored connection to Gitea, and the daemon's `--once`, `--labels` and `shutdown_timeout` behaviour. Stacked on https://gitea.com/gitea/runner/pulls/1346, so its commit shows here until that merges. The new scenarios pass locally against gitea/gitea:latest and main-nightly. Assisted-by: Codet:codet Co-authored-by: Zettat123 Co-authored-by: bircni Reviewed-on: https://gitea.com/gitea/runner/pulls/1347 Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com> --- DEVELOPMENT.md | 6 +- e2e/cli_test.go | 224 +++++++++++++++++---- e2e/compatibility_test.go | 9 + e2e/failure_test.go | 18 +- e2e/gitea_api.go | 21 +- e2e/host_test.go | 33 +++ e2e/main_test.go | 6 + e2e/reconnect_test.go | 127 ++++++++++++ e2e/reusable_test.go | 22 ++ e2e/runner_harness.go | 20 +- e2e/scheduling_test.go | 113 +++++++++++ e2e/testdata/workflows/capacity.yml | 17 ++ e2e/testdata/workflows/cli_labels.yml | 7 + e2e/testdata/workflows/cli_once.yml | 11 + e2e/testdata/workflows/cli_shutdown.yml | 10 + e2e/testdata/workflows/host.yml | 10 + e2e/testdata/workflows/reconnect.yml | 15 ++ e2e/testdata/workflows/reconnect_idle.yml | 10 + e2e/testdata/workflows/reusable_callee.yml | 20 ++ e2e/testdata/workflows/reusable_caller.yml | 12 ++ e2e/testdata/workflows/route.yml | 15 ++ 21 files changed, 661 insertions(+), 65 deletions(-) create mode 100644 e2e/host_test.go create mode 100644 e2e/reconnect_test.go create mode 100644 e2e/reusable_test.go create mode 100644 e2e/scheduling_test.go create mode 100644 e2e/testdata/workflows/capacity.yml create mode 100644 e2e/testdata/workflows/cli_labels.yml create mode 100644 e2e/testdata/workflows/cli_once.yml create mode 100644 e2e/testdata/workflows/cli_shutdown.yml create mode 100644 e2e/testdata/workflows/host.yml create mode 100644 e2e/testdata/workflows/reconnect.yml create mode 100644 e2e/testdata/workflows/reconnect_idle.yml create mode 100644 e2e/testdata/workflows/reusable_callee.yml create mode 100644 e2e/testdata/workflows/reusable_caller.yml create mode 100644 e2e/testdata/workflows/route.yml diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 5149ab89..899e469f 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -59,8 +59,10 @@ Each lives in `internal/pkg/action/` and is registered in `builtinActions` It requires Docker and is excluded from `make test`. CI runs stable and nightly variants in parallel. -The `cli_register_daemon` scenario builds the binary and drives `register` and `daemon` as a user would. The -`failure_paths` scenario covers failed, skipped, tolerated and timed-out jobs. +The `cli_*` scenarios build the binary and drive `register` and `daemon` as a user would, including `--once`, +`--labels` and shutdown with a running job. `failure_paths` covers failed, skipped, tolerated and timed-out jobs. +`label_routing` and `capacity` cover scheduling, `reconnect` drops the connection to Gitea through a proxy, and +`host_mode` and `reusable_workflow` cover host labels and `workflow_call`. The suite shares one Gitea and regular runner. Cache and ephemeral scenarios use isolated repository runners. A run path is `@` and a log row is `Z `. diff --git a/e2e/cli_test.go b/e2e/cli_test.go index eb89eb3c..62541d64 100644 --- a/e2e/cli_test.go +++ b/e2e/cli_test.go @@ -7,7 +7,6 @@ package e2e import ( "bytes" - "context" "fmt" "os" "os/exec" @@ -37,13 +36,19 @@ func (b *syncBuffer) String() string { return b.buf.String() } -// testCLIRegisterDaemon drives the built binary the way a user does: register, then daemon. -func testCLIRegisterDaemon(t *testing.T) { - t.Parallel() - ctx := t.Context() +type cliRunner struct { + api *GiteaAPI + repo string + bin string + cfgFile string + dir string +} +// newCLIRunner registers the built binary against a fresh repo with a single docker label. +func newCLIRunner(t *testing.T, label, extraRunnerConfig string) *cliRunner { + t.Helper() api, repo := newScenario(t) - token, err := fixture.RegistrationToken(ctx, repo) + token, err := fixture.RegistrationToken(t.Context(), repo) if err != nil { t.Fatalf("get registration token: %v", err) } @@ -54,71 +59,210 @@ func testCLIRegisterDaemon(t *testing.T) { t.Fatal(err) } - cfgFile := filepath.Join(dir, "config.yaml") cfg := fmt.Sprintf(`runner: file: %s fetch_interval: 250ms fetch_interval_max: 250ms state_report_interval: 500ms log_report_interval: 500ms -cache: +%scache: dir: %s -`, filepath.Join(dir, ".runner"), filepath.Join(dir, "cache")) +`, filepath.Join(dir, ".runner"), extraRunnerConfig, filepath.Join(dir, "cache")) if fixture.network != "" { cfg += "container:\n network: " + fixture.network + "\n" } - if err := os.WriteFile(cfgFile, []byte(cfg), 0o600); err != nil { + c := &cliRunner{api: api, repo: repo, bin: bin, cfgFile: filepath.Join(dir, "config.yaml"), dir: dir} + if err := os.WriteFile(c.cfgFile, []byte(cfg), 0o600); err != nil { t.Fatalf("write config: %v", err) } - label := "e2e-cli:docker://" + os.Getenv("E2E_JOB_IMAGE") - out, err := exec.CommandContext(ctx, bin, "register", "--no-interactive", "-c", cfgFile, - "--instance", fixture.baseURL, "--token", token, "--name", repo, "--labels", label).CombinedOutput() + out, err := exec.CommandContext(t.Context(), c.bin, "register", "--no-interactive", "-c", c.cfgFile, + "--instance", fixture.baseURL, "--token", token, "--name", repo, + "--labels", label+":docker://"+os.Getenv("E2E_JOB_IMAGE")).CombinedOutput() if err != nil { t.Fatalf("register: %v\n%s", err, out) } if _, err := os.Stat(filepath.Join(dir, ".runner")); err != nil { t.Fatalf("register did not write the runner file: %v\n%s", err, out) } + return c +} - daemonLog := &syncBuffer{} - daemon := exec.Command(bin, "daemon", "-c", cfgFile) - daemon.Stdout, daemon.Stderr = daemonLog, daemonLog - if err := daemon.Start(); err != nil { +type cliDaemon struct { + cmd *exec.Cmd + log *syncBuffer + exited chan error +} + +func (c *cliRunner) startDaemon(t *testing.T, args ...string) *cliDaemon { + t.Helper() + d := &cliDaemon{log: &syncBuffer{}, exited: make(chan error, 1)} + d.cmd = exec.Command(c.bin, append([]string{"daemon", "-c", c.cfgFile}, args...)...) + d.cmd.Stdout, d.cmd.Stderr = d.log, d.log + if err := d.cmd.Start(); err != nil { t.Fatalf("start daemon: %v", err) } - exited := make(chan error, 1) - go func() { exited <- daemon.Wait() }() + go func() { d.exited <- d.cmd.Wait() }() t.Cleanup(func() { - _ = daemon.Process.Kill() // no-op once the daemon has exited + _ = d.cmd.Process.Kill() // no-op once the daemon has exited }) + return d +} - pushWorkflow(t, api, repo, "cli.yml") - wfRun := waitForRun(t, api, repo) - requireSuccessOrDumpDaemon(t, api, repo, wfRun.ID, daemonLog) - if logs := runLogs(t, api, repo, wfRun.ID); commandRow(logs, "e2e-cli-job-ran") == "" { - t.Errorf("job logs are missing the marker:\n%s", logs) - } - - if err := daemon.Process.Signal(syscall.SIGTERM); err != nil { - t.Fatalf("signal daemon: %v", err) - } +// wait returns the daemon's exit error, failing the test when it is still running after timeout. +func (d *cliDaemon) wait(t *testing.T, timeout time.Duration) error { + t.Helper() select { - case err := <-exited: - if err != nil { - t.Errorf("daemon did not exit cleanly on SIGTERM: %v\n%s", err, daemonLog) - } - case <-time.After(30 * time.Second): - t.Errorf("daemon did not exit within 30s of SIGTERM\n%s", daemonLog) + case err := <-d.exited: + return err + case <-time.After(timeout): + t.Fatalf("daemon still running after %s\n%s", timeout, d.log) + return nil } } -func requireSuccessOrDumpDaemon(t *testing.T, api *GiteaAPI, repo string, runID int64, daemonLog *syncBuffer) { +func (d *cliDaemon) terminate(t *testing.T) { t.Helper() - completed, err := api.WaitForRunConclusion(context.WithoutCancel(t.Context()), repo, runID, runTimeout) + if err := d.cmd.Process.Signal(syscall.SIGTERM); err != nil { + t.Fatalf("signal daemon: %v", err) + } +} + +// requireSuccess waits for the run and dumps the daemon output next to the job logs on failure. +func (d *cliDaemon) requireSuccess(t *testing.T, c *cliRunner, runID int64) { + t.Helper() + completed, err := c.api.WaitForRunConclusion(t.Context(), c.repo, runID, runTimeout) if err != nil || completed.Conclusion != "success" { - t.Logf("daemon output:\n%s", daemonLog) - requireSuccess(t, api, repo, runID) + t.Logf("daemon output:\n%s", d.log) + requireSuccess(t, c.api, c.repo, runID) + } +} + +// testCLIRegisterDaemon drives the built binary the way a user does: register, then daemon. +func testCLIRegisterDaemon(t *testing.T) { + t.Parallel() + c := newCLIRunner(t, "e2e-cli", "") + d := c.startDaemon(t) + + pushWorkflow(t, c.api, c.repo, "cli.yml") + wfRun := waitForRun(t, c.api, c.repo) + d.requireSuccess(t, c, wfRun.ID) + if logs := runLogs(t, c.api, c.repo, wfRun.ID); commandRow(logs, "e2e-cli-job-ran") == "" { + t.Errorf("job logs are missing the marker:\n%s", logs) + } + + d.terminate(t) + if err := d.wait(t, 30*time.Second); err != nil { + t.Errorf("daemon did not exit cleanly on SIGTERM: %v\n%s", err, d.log) + } +} + +// testCLIOnce checks that --once takes a single job and then exits on its own. +func testCLIOnce(t *testing.T) { + t.Parallel() + c := newCLIRunner(t, "e2e-once", "") + d := c.startDaemon(t, "--once") + + pushWorkflow(t, c.api, c.repo, "cli_once.yml") + wfRun := waitForRun(t, c.api, c.repo) + if err := d.wait(t, runTimeout); err != nil { + t.Errorf("daemon --once exited with %v\n%s", err, d.log) + } + + jobs, err := c.api.Jobs(t.Context(), c.repo, wfRun.ID) + if err != nil { + t.Fatalf("list jobs: %v", err) + } + var succeeded, pending int + for _, job := range jobs { + switch { + case job.Conclusion == "success": + succeeded++ + case job.Status != "in_progress" && job.Status != "completed": + pending++ + } + } + if succeeded != 1 || pending != 1 { + t.Errorf("got %d succeeded and %d waiting jobs, want 1 and 1: %+v\n%s", succeeded, pending, jobs, d.log) + } +} + +// testCLILabelsOverride checks that daemon --labels replaces the labels given at registration. +func testCLILabelsOverride(t *testing.T) { + t.Parallel() + c := newCLIRunner(t, "e2e-registered", "") + d := c.startDaemon(t, "--labels", "e2e-override:docker://"+os.Getenv("E2E_JOB_IMAGE")) + + pushWorkflow(t, c.api, c.repo, "cli_labels.yml") + wfRun := waitForRun(t, c.api, c.repo) + d.requireSuccess(t, c, wfRun.ID) + + runners, err := c.api.Runners(t.Context(), c.repo) + if err != nil { + t.Fatalf("list runners: %v", err) + } + for _, runner := range runners { + if runner.Name != c.repo { + continue + } + var labels []string + for _, label := range runner.Labels { + labels = append(labels, label.Name) + } + if !slices.Equal(labels, []string{"e2e-override"}) { + t.Errorf("runner has labels %v on the server, want only the --labels override", labels) + } + return + } + t.Errorf("runner %q not found among %+v", c.repo, runners) +} + +// testCLIShutdownDrains checks that SIGTERM lets a running job finish within shutdown_timeout. +func testCLIShutdownDrains(t *testing.T) { + t.Parallel() + c := newCLIRunner(t, "e2e-shutdown", " shutdown_timeout: 1m\n") + d := c.startDaemon(t) + + pushWorkflow(t, c.api, c.repo, "cli_shutdown.yml") + wfRun := waitForRun(t, c.api, c.repo) + waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started") + + d.terminate(t) + if err := d.wait(t, runTimeout); err != nil { + t.Errorf("daemon did not exit cleanly: %v\n%s", err, d.log) + } + if strings.Contains(d.log.String(), "cancelled in progress jobs") { + t.Errorf("the running job was cancelled instead of drained\n%s", d.log) + } + d.requireSuccess(t, c, wfRun.ID) + if logs := runLogs(t, c.api, c.repo, wfRun.ID); commandRow(logs, "e2e-shutdown-finished") == "" { + t.Errorf("job did not run to the end:\n%s", logs) + } +} + +// testCLIShutdownCancels checks that a job still running at shutdown_timeout is cancelled. +func testCLIShutdownCancels(t *testing.T) { + t.Parallel() + c := newCLIRunner(t, "e2e-shutdown", " shutdown_timeout: 1s\n") + d := c.startDaemon(t) + + pushWorkflow(t, c.api, c.repo, "cli_shutdown.yml") + wfRun := waitForRun(t, c.api, c.repo) + waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started") + + d.terminate(t) + if err := d.wait(t, runTimeout); err != nil { + t.Errorf("daemon did not exit cleanly after cancelling the job: %v\n%s", err, d.log) + } + if !strings.Contains(d.log.String(), "cancelled in progress jobs") { + t.Errorf("daemon did not report cancelling the running job\n%s", d.log) + } + completed, err := c.api.WaitForRunConclusion(t.Context(), c.repo, wfRun.ID, time.Minute) + if err != nil { + t.Fatalf("cancelled run was never finalized: %v", err) + } + if completed.Conclusion == "success" { + t.Errorf("run concluded success although the job was cancelled") } } diff --git a/e2e/compatibility_test.go b/e2e/compatibility_test.go index d0c47971..13a8029e 100644 --- a/e2e/compatibility_test.go +++ b/e2e/compatibility_test.go @@ -35,6 +35,15 @@ func TestCompatibility(t *testing.T) { t.Run("ephemeral", testEphemeralRunner) t.Run("failure_paths", testFailurePaths) t.Run("cli_register_daemon", testCLIRegisterDaemon) + t.Run("cli_once", testCLIOnce) + t.Run("cli_labels_override", testCLILabelsOverride) + t.Run("cli_shutdown_drains", testCLIShutdownDrains) + t.Run("cli_shutdown_cancels", testCLIShutdownCancels) + t.Run("host_mode", testHostMode) + t.Run("reusable_workflow", testReusableWorkflow) + t.Run("label_routing", testLabelRouting) + t.Run("capacity", testCapacity) + t.Run("reconnect", testReconnect) t.Run("kubernetes", testKubernetes) } diff --git a/e2e/failure_test.go b/e2e/failure_test.go index ba33b55a..53c188ef 100644 --- a/e2e/failure_test.go +++ b/e2e/failure_test.go @@ -5,10 +5,7 @@ package e2e -import ( - "strings" - "testing" -) +import "testing" func testFailurePaths(t *testing.T) { t.Parallel() @@ -47,18 +44,7 @@ func testFailurePaths(t *testing.T) { t.Errorf("job %q not found", name) } - var all strings.Builder - for _, job := range jobs { - if job.Conclusion == "skipped" { // a skipped job never started, so it has no log - continue - } - jobLogs, err := api.JobLogs(ctx, repo, job.ID) - if err != nil { - t.Fatalf("job logs for %s: %v", job.Name, err) - } - all.WriteString(jobLogs) - } - logs := all.String() + logs := runLogs(t, api, repo, wfRun.ID) for _, marker := range []string{"e2e-before-fail", "e2e-failure-cleanup", "e2e-tolerated-next", "e2e-timeout-started"} { if commandRow(logs, marker) != marker { t.Errorf("logs are missing %q", marker) diff --git a/e2e/gitea_api.go b/e2e/gitea_api.go index 1568887a..72e5dc63 100644 --- a/e2e/gitea_api.go +++ b/e2e/gitea_api.go @@ -37,6 +37,14 @@ type ActionJob struct { Name string `json:"name"` Status string `json:"status"` Conclusion string `json:"conclusion"` + RunnerName string `json:"runner_name"` +} + +type ActionRunner struct { + Name string `json:"name"` + Labels []struct { + Name string `json:"name"` + } `json:"labels"` } func (a *GiteaAPI) CreateRepo(ctx context.Context, name string) error { @@ -146,6 +154,17 @@ func (a *GiteaAPI) Jobs(ctx context.Context, repo string, runID int64) ([]Action return resp.Jobs, nil } +func (a *GiteaAPI) Runners(ctx context.Context, repo string) ([]ActionRunner, error) { + var resp struct { + Runners []ActionRunner `json:"runners"` + } + url := fmt.Sprintf("/api/v1/repos/%s/%s/actions/runners", giteaAdminUser, repo) + if err := a.doJSON(ctx, http.MethodGet, url, nil, &resp); err != nil { + return nil, err + } + return resp.Runners, nil +} + func (a *GiteaAPI) JobLogs(ctx context.Context, repo string, jobID int64) (string, error) { url := fmt.Sprintf("/api/v1/repos/%s/%s/actions/jobs/%d/logs", giteaAdminUser, repo, jobID) req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.baseURL+url, nil) @@ -165,7 +184,7 @@ func (a *GiteaAPI) JobLogs(ctx context.Context, repo string, jobID int64) (strin return "", err } if resp.StatusCode >= 300 { - return "", fmt.Errorf("GET %s: %d: %s", url, resp.StatusCode, body) + return "", statusError{method: http.MethodGet, path: url, code: resp.StatusCode, body: string(body)} } return string(body), nil } diff --git a/e2e/host_test.go b/e2e/host_test.go new file mode 100644 index 00000000..4630a8a4 --- /dev/null +++ b/e2e/host_test.go @@ -0,0 +1,33 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +//go:build e2e + +package e2e + +import ( + "os" + "testing" +) + +func testHostMode(t *testing.T) { + t.Parallel() + ctx := t.Context() + + api, repo := newScenario(t) + hostname, err := os.Hostname() + if err != nil { + t.Fatalf("hostname: %v", err) + } + if err := api.CreateVariable(ctx, repo, "E2E_HOSTNAME", hostname); err != nil { + t.Fatalf("create variable: %v", err) + } + startRunner(t, repo, "e2e-host", runnerOptions{host: true}) + pushWorkflow(t, api, repo, "host.yml") + + wfRun := waitForRun(t, api, repo) + requireSuccess(t, api, repo, wfRun.ID) + if logs := runLogs(t, api, repo, wfRun.ID); commandRow(logs, "e2e-host-job-ran") == "" { + t.Errorf("job logs are missing the marker:\n%s", logs) + } +} diff --git a/e2e/main_test.go b/e2e/main_test.go index 770bd3b5..8eff7ce0 100644 --- a/e2e/main_test.go +++ b/e2e/main_test.go @@ -8,7 +8,9 @@ package e2e import ( "cmp" "context" + "errors" "fmt" + "net/http" "os" "os/exec" "path/filepath" @@ -129,6 +131,10 @@ func runLogs(t *testing.T, api *GiteaAPI, repo string, runID int64) string { var all strings.Builder for _, job := range jobs { logs, err := api.JobLogs(ctx, repo, job.ID) + var response statusError + if errors.As(err, &response) && response.code == http.StatusNotFound { + continue // skipped jobs and reusable workflow callers never ran a task, so have no log + } if err != nil { t.Fatalf("job logs for %s: %v", job.Name, err) } diff --git a/e2e/reconnect_test.go b/e2e/reconnect_test.go new file mode 100644 index 00000000..346bfec6 --- /dev/null +++ b/e2e/reconnect_test.go @@ -0,0 +1,127 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +//go:build e2e + +package e2e + +import ( + "fmt" + "io" + "net" + "net/url" + "sync" + "sync/atomic" + "testing" + "time" +) + +// flakyProxy forwards TCP to Gitea and can drop every connection to simulate an outage. +type flakyProxy struct { + ln net.Listener + target string + down atomic.Bool + refused atomic.Int64 + mu sync.Mutex + conns map[net.Conn]struct{} +} + +func newFlakyProxy(t *testing.T, targetURL string) *flakyProxy { + t.Helper() + u, err := url.Parse(targetURL) + if err != nil { + t.Fatalf("parse %q: %v", targetURL, err) + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + p := &flakyProxy{ln: ln, target: u.Host, conns: map[net.Conn]struct{}{}} + go p.serve() + t.Cleanup(func() { + _ = ln.Close() + p.dropAll() + }) + return p +} + +func (p *flakyProxy) URL() string { return "http://" + p.ln.Addr().String() } + +func (p *flakyProxy) serve() { + for { + c, err := p.ln.Accept() + if err != nil { + return + } + if p.down.Load() { + p.refused.Add(1) + _ = c.Close() + continue + } + up, err := net.Dial("tcp", p.target) + if err != nil { + _ = c.Close() + continue + } + p.mu.Lock() + p.conns[c], p.conns[up] = struct{}{}, struct{}{} + p.mu.Unlock() + go func() { _, _ = io.Copy(up, c); _ = up.Close() }() + go func() { _, _ = io.Copy(c, up); _ = c.Close() }() + } +} + +func (p *flakyProxy) dropAll() { + p.mu.Lock() + defer p.mu.Unlock() + for c := range p.conns { + _ = c.Close() + } + clear(p.conns) +} + +// cut starts the outage and returns once the runner has been refused at least refusals more times. +func (p *flakyProxy) cut(t *testing.T, refusals int64) { + t.Helper() + before := p.refused.Load() + p.down.Store(true) + p.dropAll() + deadline := time.Now().Add(30 * time.Second) + for p.refused.Load() < before+refusals { + if time.Now().After(deadline) { + t.Fatal("runner never retried during the outage") + } + time.Sleep(pollInterval) + } +} + +func (p *flakyProxy) restore() { p.down.Store(false) } + +// testReconnect checks that a runner recovers from a lost connection to Gitea, idle or mid-job. +func testReconnect(t *testing.T) { + t.Parallel() + api, repo := newScenario(t) + proxy := newFlakyProxy(t, fixture.baseURL) + startRunner(t, repo, "e2e-reconnect", runnerOptions{host: true, address: proxy.URL()}) + + proxy.cut(t, 2) + pushWorkflow(t, api, repo, "reconnect_idle.yml") + idle := waitForRun(t, api, repo) + proxy.restore() + requireSuccess(t, api, repo, idle.ID) + + pushWorkflow(t, api, repo, "reconnect.yml") + slow := waitForRunAfter(t, api, repo, idle.ID) + waitForRunningJobLog(t, api, repo, slow.ID, "e2e-before-outage") + // the job logs a line per second, so a long outage spans several that can only arrive after it + proxy.cut(t, 8) + proxy.restore() + requireSuccess(t, api, repo, slow.ID) + logs := runLogs(t, api, repo, slow.ID) + for i := 1; i <= 15; i++ { + if tick := fmt.Sprintf("tick-%02d", i); commandRow(logs, tick) != tick { + t.Errorf("log line %s written during or after the outage was lost:\n%s", tick, logs) + break + } + } +} diff --git a/e2e/reusable_test.go b/e2e/reusable_test.go new file mode 100644 index 00000000..0f0820aa --- /dev/null +++ b/e2e/reusable_test.go @@ -0,0 +1,22 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +//go:build e2e + +package e2e + +import "testing" + +// testReusableWorkflow runs a local workflow_call workflow and consumes its output. +func testReusableWorkflow(t *testing.T) { + t.Parallel() + api, repo := newScenario(t) + pushWorkflow(t, api, repo, "reusable_callee.yml") + pushWorkflow(t, api, repo, "reusable_caller.yml") + + wfRun := waitForRun(t, api, repo) + requireSuccess(t, api, repo, wfRun.ID) + if logs := runLogs(t, api, repo, wfRun.ID); commandRow(logs, "e2e-reusable-ran") == "" { + t.Errorf("the called workflow did not run:\n%s", logs) + } +} diff --git a/e2e/runner_harness.go b/e2e/runner_harness.go index 81f4b700..fe832bc9 100644 --- a/e2e/runner_harness.go +++ b/e2e/runner_harness.go @@ -30,11 +30,15 @@ type runnerOptions struct { cacheHost string image string kubernetes *config.Kubernetes + host bool // register a host-mode label instead of a docker one + name string // suffix that keeps runner names unique when a test starts several + address string // Gitea address the runner talks to, default the fixture } func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *poll.Poller { t.Helper() ctx := t.Context() + address := cmp.Or(options.address, fixture.baseURL) token, err := fixture.RegistrationToken(ctx, repo) if err != nil { t.Fatalf("get registration token: %v", err) @@ -65,19 +69,23 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p scheme, cfg.Kubernetes = labels.SchemeKubernetes, *options.kubernetes } rawLabel := labelName + ":" + scheme + "://" + cmp.Or(options.image, os.Getenv("E2E_JOB_IMAGE")) + if options.host { + rawLabel = labelName + ":" + labels.SchemeHost + cfg.Host.WorkdirParent = t.TempDir() + } label, err := labels.Parse(rawLabel) if err != nil { t.Fatalf("parse label %q: %v", rawLabel, err) } labelNames := []string{label.Name} - pingCli := client.New(fixture.baseURL, cfg.Runner.TLSConfig(), "", "", config.RequestTimeout, cfg.Runner.ExtraHeaders) - if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name()}); err != nil { - t.Fatalf("ping %s: %v", fixture.baseURL, err) + pingCli := client.New(address, cfg.Runner.TLSConfig(), "", "", config.RequestTimeout, cfg.Runner.ExtraHeaders) + if _, err := pingCli.Ping(ctx, &pingv1.PingRequest{Data: t.Name() + options.name}); err != nil { + t.Fatalf("ping %s: %v", address, err) } regResp, err := pingCli.Register(ctx, &runnerv1.RegisterRequest{ - Name: t.Name(), + Name: t.Name() + options.name, Token: token, Version: "e2e", Labels: labelNames, @@ -96,12 +104,12 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p UUID: regResp.Runner.Uuid, Name: regResp.Runner.Name, Token: regResp.Runner.Token, - Address: fixture.baseURL, + Address: address, Labels: []string{rawLabel}, Ephemeral: regResp.Runner.Ephemeral, } - cli := client.New(fixture.baseURL, cfg.Runner.TLSConfig(), reg.UUID, reg.Token, config.RequestTimeout, cfg.Runner.ExtraHeaders) + cli := client.New(address, cfg.Runner.TLSConfig(), reg.UUID, reg.Token, config.RequestTimeout, cfg.Runner.ExtraHeaders) runner := run.NewRunner(cfg, reg, cli) if _, err := runner.Declare(ctx, labelNames); err != nil { diff --git a/e2e/scheduling_test.go b/e2e/scheduling_test.go new file mode 100644 index 00000000..b639a152 --- /dev/null +++ b/e2e/scheduling_test.go @@ -0,0 +1,113 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +//go:build e2e + +package e2e + +import ( + "context" + "strconv" + "strings" + "testing" + "time" +) + +// testLabelRouting checks that each job reaches the runner owning its label and that a +// job with no matching runner stays queued. +func testLabelRouting(t *testing.T) { + t.Parallel() + ctx := t.Context() + + api, repo := newScenario(t) + startRunner(t, repo, "e2e-route-a", runnerOptions{host: true, name: "-a"}) + startRunner(t, repo, "e2e-route-b", runnerOptions{host: true, name: "-b"}) + pushWorkflow(t, api, repo, "route.yml") + + wfRun := waitForRun(t, api, repo) + ctx, cancel := context.WithTimeout(ctx, runTimeout) + defer cancel() + var jobs map[string]ActionJob + for { + list, err := api.Jobs(ctx, repo, wfRun.ID) + if err != nil { + t.Fatalf("list jobs: %v", err) + } + jobs = map[string]ActionJob{} + for _, job := range list { + jobs[job.Name] = job + } + if jobs["a"].Status == "completed" && jobs["b"].Status == "completed" { + break + } + select { + case <-ctx.Done(): + t.Fatalf("jobs a and b did not complete: %+v", list) + case <-time.After(pollInterval): + } + } + + for name, suffix := range map[string]string{"a": "-a", "b": "-b"} { + if got := jobs[name].RunnerName; !strings.HasSuffix(got, suffix) { + t.Errorf("job %s ran on runner %q, want the one named *%s", name, got, suffix) + } + if jobs[name].Conclusion != "success" { + t.Errorf("job %s concluded %q", name, jobs[name].Conclusion) + } + } + // a and b were picked up after unmatched was queued, so a runner that could take it would have by now + if status := jobs["unmatched"].Status; status == "in_progress" || status == "completed" { + t.Errorf("job with no matching runner is %q, want it queued", status) + } +} + +// testCapacity checks that runner.capacity bounds how many jobs run at once. +func testCapacity(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + capacity int + concurrent bool + }{ + {"serial", 1, false}, + {"parallel", 2, true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + api, repo := newScenario(t) + startRunner(t, repo, "e2e-capacity", runnerOptions{capacity: tc.capacity}) + pushWorkflow(t, api, repo, "capacity.yml") + + wfRun := waitForRun(t, api, repo) + requireSuccess(t, api, repo, wfRun.ID) + + jobs, err := api.Jobs(t.Context(), repo, wfRun.ID) + if err != nil || len(jobs) != 2 { + t.Fatalf("list jobs: %v (%d jobs)", err, len(jobs)) + } + var start, end [2]int64 + for i, job := range jobs { + logs, err := api.JobLogs(t.Context(), repo, job.ID) + if err != nil { + t.Fatalf("job logs: %v", err) + } + start[i] = spanValue(t, logs, "span-start=") + end[i] = spanValue(t, logs, "span-end=") + } + overlap := max(start[0], start[1]) < min(end[0], end[1]) + if overlap != tc.concurrent { + t.Errorf("capacity %d: jobs overlapped=%v, want %v (starts %v ends %v)", tc.capacity, overlap, tc.concurrent, start, end) + } + }) + } +} + +func spanValue(t *testing.T, logs, prefix string) int64 { + t.Helper() + row := commandRow(logs, prefix) + n, err := strconv.ParseInt(strings.TrimPrefix(row, prefix), 10, 64) + if err != nil { + t.Fatalf("no %q value in logs:\n%s", prefix, logs) + } + return n +} diff --git a/e2e/testdata/workflows/capacity.yml b/e2e/testdata/workflows/capacity.yml new file mode 100644 index 00000000..108fbd2e --- /dev/null +++ b/e2e/testdata/workflows/capacity.yml @@ -0,0 +1,17 @@ +name: capacity +on: push +jobs: + one: + runs-on: e2e-capacity + steps: + - run: | + echo "span-start=$(date +%s%N)" + sleep 8 + echo "span-end=$(date +%s%N)" + two: + runs-on: e2e-capacity + steps: + - run: | + echo "span-start=$(date +%s%N)" + sleep 8 + echo "span-end=$(date +%s%N)" diff --git a/e2e/testdata/workflows/cli_labels.yml b/e2e/testdata/workflows/cli_labels.yml new file mode 100644 index 00000000..dfc4eb7c --- /dev/null +++ b/e2e/testdata/workflows/cli_labels.yml @@ -0,0 +1,7 @@ +name: cli-labels +on: push +jobs: + hello: + runs-on: e2e-override + steps: + - run: echo e2e-override-ran diff --git a/e2e/testdata/workflows/cli_once.yml b/e2e/testdata/workflows/cli_once.yml new file mode 100644 index 00000000..f1aec338 --- /dev/null +++ b/e2e/testdata/workflows/cli_once.yml @@ -0,0 +1,11 @@ +name: cli-once +on: push +jobs: + first: + runs-on: e2e-once + steps: + - run: echo e2e-once-first + second: + runs-on: e2e-once + steps: + - run: echo e2e-once-second diff --git a/e2e/testdata/workflows/cli_shutdown.yml b/e2e/testdata/workflows/cli_shutdown.yml new file mode 100644 index 00000000..4b4f718c --- /dev/null +++ b/e2e/testdata/workflows/cli_shutdown.yml @@ -0,0 +1,10 @@ +name: cli-shutdown +on: push +jobs: + slow: + runs-on: e2e-shutdown + steps: + - run: | + echo e2e-shutdown-started + sleep 8 + echo e2e-shutdown-finished diff --git a/e2e/testdata/workflows/host.yml b/e2e/testdata/workflows/host.yml new file mode 100644 index 00000000..e35ea036 --- /dev/null +++ b/e2e/testdata/workflows/host.yml @@ -0,0 +1,10 @@ +name: host +on: push +jobs: + host: + runs-on: e2e-host + steps: + - run: | + test "$(hostname)" = "${{ vars.E2E_HOSTNAME }}" + test -n "$GITHUB_WORKSPACE" + echo e2e-host-job-ran diff --git a/e2e/testdata/workflows/reconnect.yml b/e2e/testdata/workflows/reconnect.yml new file mode 100644 index 00000000..4b966476 --- /dev/null +++ b/e2e/testdata/workflows/reconnect.yml @@ -0,0 +1,15 @@ +name: reconnect +on: + push: + paths: + - .gitea/workflows/reconnect.yml +jobs: + slow: + runs-on: e2e-reconnect + steps: + - run: | + echo e2e-before-outage + for i in $(seq -w 1 15); do + echo "tick-$i" + sleep 1 + done diff --git a/e2e/testdata/workflows/reconnect_idle.yml b/e2e/testdata/workflows/reconnect_idle.yml new file mode 100644 index 00000000..90cbda06 --- /dev/null +++ b/e2e/testdata/workflows/reconnect_idle.yml @@ -0,0 +1,10 @@ +name: reconnect-idle +on: + push: + paths: + - .gitea/workflows/reconnect_idle.yml +jobs: + hello: + runs-on: e2e-reconnect + steps: + - run: echo e2e-idle-recovered diff --git a/e2e/testdata/workflows/reusable_callee.yml b/e2e/testdata/workflows/reusable_callee.yml new file mode 100644 index 00000000..523436c7 --- /dev/null +++ b/e2e/testdata/workflows/reusable_callee.yml @@ -0,0 +1,20 @@ +name: reusable-callee +on: + workflow_call: + inputs: + who: + type: string + required: true + outputs: + greeting: + value: ${{ jobs.say.outputs.greeting }} +jobs: + say: + runs-on: ubuntu-latest + outputs: + greeting: ${{ steps.greet.outputs.greeting }} + steps: + - id: greet + run: | + echo "greeting=hello-${{ inputs.who }}" >> "$GITHUB_OUTPUT" + echo e2e-reusable-ran diff --git a/e2e/testdata/workflows/reusable_caller.yml b/e2e/testdata/workflows/reusable_caller.yml new file mode 100644 index 00000000..5b8c6c2e --- /dev/null +++ b/e2e/testdata/workflows/reusable_caller.yml @@ -0,0 +1,12 @@ +name: reusable-caller +on: push +jobs: + call: + uses: ./.gitea/workflows/reusable_callee.yml + with: + who: e2e + verify: + needs: call + runs-on: ubuntu-latest + steps: + - run: test "${{ needs.call.outputs.greeting }}" = "hello-e2e" diff --git a/e2e/testdata/workflows/route.yml b/e2e/testdata/workflows/route.yml new file mode 100644 index 00000000..60eb2edc --- /dev/null +++ b/e2e/testdata/workflows/route.yml @@ -0,0 +1,15 @@ +name: route +on: push +jobs: + a: + runs-on: e2e-route-a + steps: + - run: echo route-a + b: + runs-on: e2e-route-b + steps: + - run: echo route-b + unmatched: + runs-on: e2e-route-none + steps: + - run: echo route-none