diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 899e469f..a126c951 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -57,12 +57,15 @@ Each lives in `internal/pkg/action/` and is registered in `builtinActions` `make test-e2e` runs the runner against `E2E_GITEA_IMAGE`. It defaults to the nightly image. It requires Docker and is excluded from `make test`. CI runs stable and nightly variants in -parallel. +parallel. Outside a container it runs `go test` in a Linux container on its own network, as CI does, +so both take the same paths. `E2E_ARGS` passes extra flags, e.g. `make test-e2e E2E_ARGS='-run TestCompatibility/dispatch'`. 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`. +`--labels` and shutdown with a running job. `failure_paths` covers failed, skipped and tolerated jobs, `job_timeout` +a job killed at the runner's deadline. `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`. +`dispatch`, `failure_paths` and `reusable_workflow` run in every mode: through the daemon and `exec`, each on +Docker and Kubernetes. -The suite shares one Gitea and regular runner. Cache and ephemeral scenarios use isolated +The suite shares one Gitea, regular runner and k3s cluster. Cache and ephemeral scenarios use isolated repository runners. A run path is `@` and a log row is `Z `. diff --git a/Makefile b/Makefile index 297a907f..dbf9230e 100644 --- a/Makefile +++ b/Makefile @@ -154,7 +154,7 @@ E2E_CONCURRENCY ?= 8 .PHONY: test-e2e test-e2e: ## run Gitea compatibility tests against E2E_GITEA_IMAGE - @E2E_CONCURRENCY=$(E2E_CONCURRENCY) E2E_GITEA_IMAGE=$(E2E_GITEA_IMAGE) E2E_K3S_IMAGE=$(E2E_K3S_IMAGE) E2E_JOB_IMAGE=$(E2E_JOB_IMAGE) GO=$(GO) SERVICE_IMAGE=$(SERVICE_IMAGE) ./tools/test-e2e.sh + @E2E_CONCURRENCY=$(E2E_CONCURRENCY) E2E_GITEA_IMAGE=$(E2E_GITEA_IMAGE) E2E_K3S_IMAGE=$(E2E_K3S_IMAGE) E2E_JOB_IMAGE=$(E2E_JOB_IMAGE) GO=$(GO) SERVICE_IMAGE=$(SERVICE_IMAGE) ./tools/test-e2e.sh $(E2E_ARGS) .PHONY: install install: $(GOFILES) ## install the runner binary via `go install` diff --git a/act/runner/job_executor_test.go b/act/runner/job_executor_test.go index 447a30e0..d109c7a9 100644 --- a/act/runner/job_executor_test.go +++ b/act/runner/job_executor_test.go @@ -1047,23 +1047,25 @@ func makeTestRC(t *testing.T, jobYAML string) *RunContext { func TestApplyJobTimeout(t *testing.T) { cases := []struct { - name string - yaml string - wantTimeout bool + name string + yaml string + want time.Duration }{ - {"empty", "runs-on: ubuntu-latest", false}, - {"integer", "timeout-minutes: 5\nruns-on: ubuntu-latest", true}, - {"zero ignored", "timeout-minutes: 0\nruns-on: ubuntu-latest", false}, - {"non-numeric ignored", "timeout-minutes: abc\nruns-on: ubuntu-latest", false}, + {"empty", "runs-on: ubuntu-latest", 0}, + {"integer", "timeout-minutes: 5\nruns-on: ubuntu-latest", 5 * time.Minute}, + {"zero ignored", "timeout-minutes: 0\nruns-on: ubuntu-latest", 0}, + {"non-numeric ignored", "timeout-minutes: abc\nruns-on: ubuntu-latest", 0}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { rc := makeTestRC(t, tc.yaml) - ctx := t.Context() - newCtx, cancel := applyJobTimeout(ctx, rc, rc.Run.Job()) + newCtx, cancel := applyJobTimeout(t.Context(), rc, rc.Run.Job()) defer cancel() - _, hasDeadline := newCtx.Deadline() - assert.Equal(t, tc.wantTimeout, hasDeadline) + deadline, hasDeadline := newCtx.Deadline() + assert.Equal(t, tc.want != 0, hasDeadline) + if hasDeadline { + assert.WithinDuration(t, time.Now().Add(tc.want), deadline, time.Minute) + } }) } } diff --git a/e2e/cancel_test.go b/e2e/cancel_test.go index 2c06bd29..80d675cf 100644 --- a/e2e/cancel_test.go +++ b/e2e/cancel_test.go @@ -28,7 +28,7 @@ func testRunCancellation(t *testing.T) { pushWorkflow(t, api, repo, "cancel.yml") wfRun := waitForRun(t, api, repo) - waitForRunningJobLog(t, api, repo, wfRun.ID, "e2e-live-log-marker") + waitForRunningJobLog(t, api, repo, wfRun.ID, "e2e-live-log-marker", 1) if !cancelSupported { requireSuccess(t, api, repo, wfRun.ID) return @@ -49,31 +49,32 @@ func testRunCancellation(t *testing.T) { } } -func waitForRunningJobLog(t *testing.T, api *GiteaAPI, repo string, runID int64, substr string) { +func waitForRunningJobLog(t *testing.T, api *GiteaAPI, repo string, runID int64, prefix string, count int) { t.Helper() // The budget covers scheduling and the container start, which a loaded runner makes slow. ctx, cancel := context.WithTimeout(t.Context(), runTimeout) defer cancel() - status := "none" for { jobs, err := api.Jobs(ctx, repo, runID) if err != nil { t.Fatalf("list jobs: %v", err) } - if len(jobs) > 0 { - status = jobs[0].Status - if status == "in_progress" { - logs, err := api.JobLogs(ctx, repo, jobs[0].ID) - if err == nil && commandRow(logs, substr) == substr { - return + logged := 0 + for _, job := range jobs { + if job.Status == "in_progress" { + if logs, err := api.JobLogs(ctx, repo, job.ID); err == nil && commandRow(logs, prefix) != "" { + logged++ } } } + if logged >= count { + return + } select { case <-ctx.Done(): dumpRunLogs(t, api, repo, runID) - t.Fatalf("job of run %d is %q and never logged %q", runID, status, substr) + t.Fatalf("%d of %d running jobs of run %d logged %q", logged, count, runID, prefix) case <-time.After(pollInterval): } } diff --git a/e2e/cli_test.go b/e2e/cli_test.go index 689d52b4..d295701b 100644 --- a/e2e/cli_test.go +++ b/e2e/cli_test.go @@ -67,10 +67,9 @@ func newCLIRunner(t *testing.T, label, extraRunnerConfig string) *cliRunner { log_report_interval: 500ms %scache: dir: %s -`, filepath.Join(dir, ".runner"), extraRunnerConfig, filepath.Join(dir, "cache")) - if fixture.network != "" { - cfg += "container:\n network: " + fixture.network + "\n" - } +container: + network: %s +`, filepath.Join(dir, ".runner"), extraRunnerConfig, filepath.Join(dir, "cache"), fixture.network) 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) @@ -221,13 +220,26 @@ func testCLILabelsOverride(t *testing.T) { func testCLIShutdownDrains(t *testing.T) { t.Parallel() c := newCLIRunner(t, "e2e-shutdown", " shutdown_timeout: 1m\n") + gate, release := newGate(t) + if err := c.api.CreateVariable(t.Context(), c.repo, "E2E_GATE", gate); err != nil { + t.Fatalf("create gate variable: %v", err) + } 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") + waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started", 1) d.terminate(t) + deadline := time.After(time.Minute) + for !strings.Contains(d.log.String(), "shutdown initiated") { + select { + case <-deadline: + t.Fatalf("daemon never started shutting down\n%s", d.log) + case <-time.After(pollInterval): + } + } + release() if err := d.wait(t, runTimeout); err != nil { t.Errorf("daemon did not exit cleanly: %v\n%s", err, d.log) } @@ -248,7 +260,7 @@ func testCLIShutdownCancels(t *testing.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") + waitForRunningJobLog(t, c.api, c.repo, wfRun.ID, "e2e-shutdown-started", 1) d.terminate(t) if err := d.wait(t, runTimeout); err != nil { @@ -266,7 +278,7 @@ func testCLIShutdownCancels(t *testing.T) { } } -func testExec(t *testing.T) { +func testExecArtifacts(t *testing.T) { t.Parallel() ctx := t.Context() @@ -274,25 +286,18 @@ func testExec(t *testing.T) { if err != nil { t.Fatal(err) } - workflows, workdir := t.TempDir(), t.TempDir() - for dst, src := range map[string]string{ - filepath.Join(workflows, "a.yml"): "exec_artifact.yml", - filepath.Join(workflows, "b.yml"): "exec_artifact.yml", - filepath.Join(workflows, "reusable_caller.yml"): "reusable_caller.yml", - filepath.Join(workdir, ".gitea", "workflows", "reusable_callee.yml"): "reusable_callee.yml", - } { - content, err := os.ReadFile(filepath.Join("testdata", "workflows", src)) - if err != nil { - t.Fatal(err) - } - if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { - t.Fatal(err) - } - if err := os.WriteFile(dst, content, 0o644); err != nil { + workflow, err := os.ReadFile("testdata/workflows/exec_artifact.yml") + if err != nil { + t.Fatal(err) + } + workflows := t.TempDir() + for _, name := range []string{"a.yml", "b.yml"} { + if err := os.WriteFile(filepath.Join(workflows, name), workflow, 0o644); err != nil { t.Fatal(err) } } + workdir := t.TempDir() for _, args := range [][]string{ {"init", "-q"}, {"remote", "add", "origin", "https://example.com/e2e/exec.git"}, @@ -304,19 +309,12 @@ func testExec(t *testing.T) { } artifacts := t.TempDir() - args := []string{"exec", "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--no-skip-checkout"} - if fixture.network != "" { - args = append(args, "--network", fixture.network) - } + args := []string{"exec", "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--network", fixture.network} cmd := exec.CommandContext(ctx, bin, args...) cmd.Env = slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") }) - out, err := cmd.CombinedOutput() - if err != nil { + if out, err := cmd.CombinedOutput(); err != nil { t.Fatalf("exec: %v\n%s", err, out) } - if !strings.Contains(string(out), "e2e-reusable-ran") { - t.Errorf("the called workflow did not run:\n%s", out) - } if blobs, _ := filepath.Glob(filepath.Join(artifacts, "v4", "*", "*", "blob")); len(blobs) != 2 { t.Fatalf("want one v4 artifact blob per workflow run, got %v", blobs) } diff --git a/e2e/compatibility_test.go b/e2e/compatibility_test.go index 2ef0381b..7fa9331b 100644 --- a/e2e/compatibility_test.go +++ b/e2e/compatibility_test.go @@ -28,19 +28,20 @@ func TestCompatibility(t *testing.T) { t.Run("payloads", testPayloads) t.Run("cache_and_artifacts", testCacheAndArtifactRoundTrips) t.Run("cancellation_and_log_streaming", testRunCancellation) - t.Run("dispatch", testWorkflowDispatch) + t.Run("dispatch", inAllModes(testWorkflowDispatch)) t.Run("builtin_checkout", testBuiltinCheckout) - t.Run("exec", testExec) + t.Run("exec_artifacts", testExecArtifacts) t.Run("moving_action_ref", testMovingActionRef) t.Run("ephemeral", testEphemeralRunner) - t.Run("failure_paths", testFailurePaths) + t.Run("failure_paths", inAllModes(testFailurePaths)) + t.Run("job_timeout", testJobTimeout) 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("reusable_workflow", inAllModes(testReusableWorkflow)) t.Run("label_routing", testLabelRouting) t.Run("capacity", testCapacity) t.Run("reconnect", testReconnect) diff --git a/e2e/dispatch_test.go b/e2e/dispatch_test.go index 97b6aa4e..c3dc2c95 100644 --- a/e2e/dispatch_test.go +++ b/e2e/dispatch_test.go @@ -11,27 +11,11 @@ import ( "time" ) -func testWorkflowDispatch(t *testing.T) { - t.Parallel() - ctx := t.Context() - - api, repo := newScenario(t) - pushWorkflow(t, api, repo, "dispatch.yml") - - branch, err := api.DefaultBranch(ctx, repo) - if err != nil { - t.Fatalf("get default branch: %v", err) +func testWorkflowDispatch(t *testing.T, m mode) { + result := runWorkflow(t, m, workflowSpec{files: []string{"dispatch.yml"}, inputs: map[string]string{"subject": "dispatch-input-value"}}) + if result.conclusion != "success" { + t.Errorf("run concluded %q, want success", result.conclusion) } - inputs := map[string]string{"subject": "dispatch-input-value"} - if err := waitForDispatch(ctx, api, repo, "dispatch.yml", branch, inputs); err != nil { - t.Fatalf("dispatch workflow: %v", err) - } - - dispatched := waitForRun(t, api, repo) - if dispatched.Event != "workflow_dispatch" { - t.Fatalf("run event is %q, want workflow_dispatch", dispatched.Event) - } - requireSuccess(t, api, repo, dispatched.ID) } // Gitea indexes workflows asynchronously after the Contents API commit. diff --git a/e2e/failure_test.go b/e2e/failure_test.go index 53c188ef..8da917b1 100644 --- a/e2e/failure_test.go +++ b/e2e/failure_test.go @@ -5,55 +5,41 @@ package e2e -import "testing" +import ( + "testing" + "time" +) -func testFailurePaths(t *testing.T) { +func testFailurePaths(t *testing.T, m mode) { + result := runWorkflow(t, m, workflowSpec{files: []string{"failure.yml"}}) + if result.conclusion != "failure" { + t.Errorf("run concluded %q, want failure", result.conclusion) + } + for job, want := range map[string]string{"fail": "failure", "tolerated": "success", "dependent": "skipped"} { + if got := result.job(job); got != want { + t.Errorf("job %q concluded %q, want %q", job, got, want) + } + } + for _, marker := range []string{"e2e-before-fail", "e2e-failure-cleanup", "e2e-tolerated-next"} { + if !result.hasLine(marker) { + t.Errorf("logs are missing %q", marker) + } + } + if result.hasLine("e2e-skipped-step") { + t.Error("a success() step ran after a failed step") + } +} + +func testJobTimeout(t *testing.T) { t.Parallel() - ctx := t.Context() - - api, repo := newScenario(t) - pushWorkflow(t, api, repo, "failure.yml") - + api, repo, _ := startIsolatedScenario(t, "timeout.yml", "e2e-timeout", runnerOptions{timeout: 15 * time.Second}) wfRun := waitForRun(t, api, repo) - completed, err := api.WaitForRunConclusion(ctx, repo, wfRun.ID, runTimeout) + completed, err := api.WaitForRunConclusion(t.Context(), repo, wfRun.ID, runTimeout) if err != nil { dumpRunLogs(t, api, repo, wfRun.ID) t.Fatalf("wait for run: %v", err) } - if completed.Conclusion != "failure" { - t.Errorf("run concluded %q, want failure", completed.Conclusion) - } - - jobs, err := api.Jobs(ctx, repo, wfRun.ID) - if err != nil { - t.Fatalf("list jobs: %v", err) - } - want := map[string]string{ - "fail": "failure", - "tolerated": "success", - "dependent": "skipped", - "timeout": "failure", - } - for _, job := range jobs { - if w, ok := want[job.Name]; ok && job.Conclusion != w { - t.Errorf("job %q concluded %q, want %q", job.Name, job.Conclusion, w) - } - delete(want, job.Name) - } - for name := range want { - t.Errorf("job %q not found", name) - } - - 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) - } - } - if commandRow(logs, "e2e-skipped-step") != "" { - t.Error("a success() step ran after a failed step") - } - if t.Failed() { - t.Logf("full stored logs:\n%s", logs) + if logs := runLogs(t, api, repo, wfRun.ID); completed.Conclusion != "failure" || commandRow(logs, "e2e-timeout-started") == "" { + t.Errorf("run concluded %q, want failure after the job started, logs:\n%s", completed.Conclusion, logs) } } diff --git a/e2e/gitea_api.go b/e2e/gitea_api.go index 72e5dc63..6fb349b1 100644 --- a/e2e/gitea_api.go +++ b/e2e/gitea_api.go @@ -53,12 +53,16 @@ func (a *GiteaAPI) CreateRepo(ctx context.Context, name string) error { } func (a *GiteaAPI) CreateFile(ctx context.Context, repo, path, content, message string) error { - body := map[string]any{ - "content": base64.StdEncoding.EncodeToString([]byte(content)), - "message": message, + return a.CreateFiles(ctx, repo, map[string]string{path: content}, message) +} + +func (a *GiteaAPI) CreateFiles(ctx context.Context, repo string, files map[string]string, message string) error { + var operations []map[string]string + for path, content := range files { + operations = append(operations, map[string]string{"operation": "create", "path": path, "content": base64.StdEncoding.EncodeToString([]byte(content))}) } - url := fmt.Sprintf("/api/v1/repos/%s/%s/contents/%s", giteaAdminUser, repo, path) - return a.doJSON(ctx, http.MethodPost, url, body, nil) + body := map[string]any{"files": operations, "message": message} + return a.doJSON(ctx, http.MethodPost, fmt.Sprintf("/api/v1/repos/%s/%s/contents", giteaAdminUser, repo), body, nil) } func (a *GiteaAPI) CreateTag(ctx context.Context, repo, name string) error { diff --git a/e2e/gitea_fixture.go b/e2e/gitea_fixture.go index bc0e3ba0..3d19da8d 100644 --- a/e2e/gitea_fixture.go +++ b/e2e/gitea_fixture.go @@ -15,9 +15,7 @@ import ( "io" "net" "net/http" - "net/netip" "os" - "strconv" "strings" "time" @@ -40,7 +38,8 @@ type GiteaFixture struct { image string version string baseURL string - network string // set in container mode; jobs must join it + network string // the user-defined network every container of the suite joins + selfIP string // the test process's address on network, which job containers can reach adminToken string } @@ -55,44 +54,10 @@ func dockerClient(ctx context.Context) (mobyclient.APIClient, error) { return cli, nil } -// Prefer docker bridge gateway, then LAN IP; never 127.0.0.1 (ACTIONS_RUNTIME_URL must reach the host from job containers). -func hostAddress(ctx context.Context, cli mobyclient.APIClient) netip.Addr { - if gateway, ok := bridgeGateway(ctx, cli); ok && bindable(gateway) { - return gateway - } - - loopback := netip.AddrFrom4([4]byte{127, 0, 0, 1}) - - conn, err := net.Dial("udp", "192.0.2.1:80") // TEST-NET-1: routing table only - if err != nil { - return loopback - } - defer conn.Close() - - addr, ok := conn.LocalAddr().(*net.UDPAddr) - if !ok { - return loopback - } - return addr.AddrPort().Addr().Unmap() -} - -func bridgeGateway(ctx context.Context, cli mobyclient.APIClient) (netip.Addr, bool) { - inspected, err := cli.NetworkInspect(ctx, "bridge", mobyclient.NetworkInspectOptions{}) - if err != nil { - return netip.Addr{}, false - } - for _, cfg := range inspected.Network.IPAM.Config { - if cfg.Gateway.Is4() { - return cfg.Gateway, true - } - } - return netip.Addr{}, false -} - // When tests run inside a container (sibling docker daemon), join that network and address by name. -func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, bool) { +func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, string, bool) { if _, err := os.Stat("/.dockerenv"); err != nil { - return "", false + return "", "", false } id, err := os.ReadFile("/proc/sys/kernel/hostname") @@ -100,42 +65,19 @@ func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, bool) { id, err = os.ReadFile("/etc/hostname") } if err != nil { - return "", false + return "", "", false } inspected, err := cli.ContainerInspect(ctx, strings.TrimSpace(string(id)), mobyclient.ContainerInspectOptions{}) if err != nil || inspected.Container.NetworkSettings == nil { - return "", false + return "", "", false } - for name := range inspected.Container.NetworkSettings.Networks { + for name, endpoint := range inspected.Container.NetworkSettings.Networks { if name != "host" && name != "none" && name != "bridge" { // only user-defined nets have DNS - return name, true + return name, endpoint.IPAddress.String(), true } } - return "", false -} - -func bindable(addr netip.Addr) bool { - l, err := net.Listen("tcp", net.JoinHostPort(addr.String(), "0")) - if err != nil { - return false - } - _ = l.Close() - return true -} - -func freeHostPort(host string) (int, error) { - l, err := net.Listen("tcp", net.JoinHostPort(host, "0")) - if err != nil { - return 0, err - } - defer l.Close() - - addr, ok := l.Addr().(*net.TCPAddr) - if !ok { - return 0, fmt.Errorf("unexpected listener address type %T", l.Addr()) - } - return addr.Port, nil + return "", "", false } func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, error) { @@ -151,31 +93,12 @@ func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, e return nil, errors.New("E2E_GITEA_IMAGE is not set") } - name := fmt.Sprintf("gitea-runner-e2e-%d", time.Now().UnixNano()) - containerPort := network.MustParsePort("3000/tcp") - - var ( - baseURL string - hostConfig = &apicontainer.HostConfig{} - netConfig *network.NetworkingConfig - sharedNet, _ = selfNetwork(ctx, cli) - ) - if sharedNet != "" { - baseURL = "http://" + net.JoinHostPort(name, "3000") - netConfig = &network.NetworkingConfig{ - EndpointsConfig: map[string]*network.EndpointSettings{sharedNet: {}}, - } - } else { - host := hostAddress(ctx, cli) - port, err := freeHostPort(host.String()) - if err != nil { - return nil, fmt.Errorf("find a free host port: %w", err) - } - baseURL = "http://" + net.JoinHostPort(host.String(), strconv.Itoa(port)) - hostConfig.PortBindings = network.PortMap{ - containerPort: []network.PortBinding{{HostIP: host, HostPort: strconv.Itoa(port)}}, - } + sharedNet, selfIP, ok := selfNetwork(ctx, cli) + if !ok { + return nil, errors.New("e2e tests must run in a container on a user-defined network, `make test-e2e` provides one") } + name := fmt.Sprintf("gitea-runner-e2e-%d", time.Now().UnixNano()) + baseURL := "http://" + net.JoinHostPort(name, "3000") if _, err := cli.ImageInspect(ctx, image); err != nil { return nil, fmt.Errorf("inspect gitea image %s: %w", image, err) @@ -191,17 +114,16 @@ func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, e "GITEA__server__LFS_START_SERVER=true", "GITEA__server__ROOT_URL=" + baseURL + "/", }, - ExposedPorts: network.PortSet{containerPort: struct{}{}}, }, - HostConfig: hostConfig, - NetworkingConfig: netConfig, + HostConfig: &apicontainer.HostConfig{Tmpfs: map[string]string{"/data": "exec"}}, + NetworkingConfig: &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{sharedNet: {}}}, Name: name, }) if err != nil { return nil, fmt.Errorf("create gitea container: %w", err) } - f := &GiteaFixture{cli: cli, id: resp.ID, image: image, baseURL: baseURL, network: sharedNet} + f := &GiteaFixture{cli: cli, id: resp.ID, image: image, baseURL: baseURL, network: sharedNet, selfIP: selfIP} closeClient = false if _, err := cli.ContainerStart(ctx, f.id, mobyclient.ContainerStartOptions{}); err != nil { diff --git a/e2e/kubernetes_test.go b/e2e/kubernetes_test.go index 1df48ac5..33a35b05 100644 --- a/e2e/kubernetes_test.go +++ b/e2e/kubernetes_test.go @@ -9,15 +9,16 @@ import ( "bufio" "bytes" "context" + "errors" "fmt" "io" - "net" - "net/netip" "net/url" "os" "path/filepath" - "strconv" + "slices" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -32,125 +33,168 @@ import ( func testKubernetes(t *testing.T) { t.Parallel() ctx := t.Context() - k3sID, kubernetes, images := startK3s(t) + cluster := requireCluster(t) api, repo := newScenario(t) - if err := api.CreateVariable(ctx, repo, "E2E_SERVICE_IMAGE", images[1]); err != nil { + if err := api.CreateVariable(ctx, repo, "E2E_SERVICE_IMAGE", cluster.serviceImage); err != nil { t.Fatalf("create service image variable: %v", err) } - startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: images[0], kubernetes: kubernetes}) + startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: cluster.jobImage, kubernetes: cluster.kubernetes}) pushWorkflow(t, api, repo, "kubernetes.yml") wfRun := waitForRun(t, api, repo) requireSuccess(t, api, repo, wfRun.ID) +} - if out, err := containerExec(ctx, fixture.cli, k3sID, "", nil, +// checkClusterLeaks runs after every test, so a pod or secret any of them left behind fails the suite. +func checkClusterLeaks() error { + if !clusterStarted.Load() { + return nil + } + cluster, err := k3s() + if err != nil { + return nil // the tests that needed the cluster already failed + } + if out, err := containerExec(context.Background(), dockerCLI, cluster.id, "", nil, "kubectl", "wait", "--for=delete", "pods,secrets", "--selector=app.kubernetes.io/managed-by=gitea-runner", "--timeout=1m", ); err != nil { - t.Errorf("job pods or secrets left behind: %v: %s", err, out) + return fmt.Errorf("job pods or secrets left behind: %w: %s", err, out) + } + return nil +} + +type kubeCluster struct { + id string + kubernetes *config.Kubernetes + jobImage, serviceImage string +} + +var ( + k3s = sync.OnceValues(startK3s) // independent of Gitea, so it can boot alongside it + clusterStarted atomic.Bool + clusterCleanup []func() + sharedCluster = sync.OnceValues(func() (*kubeCluster, error) { + cluster, err := k3s() + if err != nil { + return nil, err + } + gitea, err := fixture.cli.ContainerInspect(context.Background(), fixture.id, mobyclient.ContainerInspectOptions{}) + if err != nil { + return nil, fmt.Errorf("inspect gitea container: %w", err) + } + giteaURL, err := url.Parse(fixture.baseURL) + if err != nil { + return nil, fmt.Errorf("parse gitea url: %w", err) + } + cluster.kubernetes.PodTemplate = map[string]any{"spec": map[string]any{"hostAliases": []any{map[string]any{ + "ip": gitea.Container.NetworkSettings.Networks[fixture.network].IPAddress.String(), + "hostnames": []any{giteaURL.Hostname()}, + }}}} + return cluster, nil + }) +) + +func requireCluster(t *testing.T) *kubeCluster { + t.Helper() + cluster, err := sharedCluster() + if err != nil { + t.Fatal(err) + } + return cluster +} + +func stopCluster() { + if !clusterStarted.Load() { + return + } + _, _ = k3s() + for _, cleanup := range slices.Backward(clusterCleanup) { + cleanup() } } -func startK3s(t *testing.T) (string, *config.Kubernetes, []string) { - t.Helper() - ctx := t.Context() - cli := fixture.cli +func startK3s() (*kubeCluster, error) { + clusterStarted.Store(true) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + cli := dockerCLI + sharedNet, _, ok := selfNetwork(ctx, cli) + if !ok { + return nil, errors.New("e2e tests must run in a container on a user-defined network") + } name := fmt.Sprintf("gitea-runner-e2e-k3s-%d", time.Now().UnixNano()) - apiPort := network.MustParsePort("6443/tcp") server := "https://" + name + ":6443" - hostConfig := &apicontainer.HostConfig{Privileged: true} - var netConfig *network.NetworkingConfig - if fixture.network != "" { - netConfig = &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{fixture.network: {}}} - } else { - loopback := netip.MustParseAddr("127.0.0.1") - port, err := freeHostPort(loopback.String()) - if err != nil { - t.Fatalf("find a free host port: %v", err) - } - server = "https://" + net.JoinHostPort(loopback.String(), strconv.Itoa(port)) - hostConfig.PortBindings = network.PortMap{apiPort: []network.PortBinding{{HostIP: loopback, HostPort: strconv.Itoa(port)}}} - } created, err := cli.ContainerCreate(ctx, mobyclient.ContainerCreateOptions{ Config: &apicontainer.Config{ - Image: os.Getenv("E2E_K3S_IMAGE"), - Cmd: []string{"server", "--disable=traefik,metrics-server,servicelb,local-storage,coredns", "--tls-san=" + name, "--kubelet-arg=fail-cgroupv1=false"}, - ExposedPorts: network.PortSet{apiPort: struct{}{}}, + Image: os.Getenv("E2E_K3S_IMAGE"), + Cmd: []string{"server", "--disable=traefik,metrics-server,servicelb,local-storage,coredns", "--tls-san=" + name, "--kubelet-arg=fail-cgroupv1=false"}, }, - HostConfig: hostConfig, - NetworkingConfig: netConfig, + HostConfig: &apicontainer.HostConfig{Privileged: true}, + NetworkingConfig: &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{sharedNet: {}}}, Name: name, }) if err != nil { - t.Fatalf("create k3s container: %v", err) + return nil, fmt.Errorf("create k3s container: %w", err) } - t.Cleanup(func() { + clusterCleanup = append(clusterCleanup, func() { _, _ = cli.ContainerRemove(context.Background(), created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true}) }) if _, err := cli.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{}); err != nil { - t.Fatalf("start k3s container: %v", err) + return nil, fmt.Errorf("start k3s container: %w", err) } - containerdCtx, stopWaiting := context.WithTimeout(ctx, 2*time.Minute) - defer stopWaiting() - if err := waitForContainerLog(containerdCtx, created.ID, "containerd is now running"); err != nil { - t.Fatalf("wait for k3s containerd: %v", err) + if err := waitForContainerLog(ctx, created.ID, "containerd is now running"); err != nil { + return nil, fmt.Errorf("wait for k3s containerd: %w", err) + } + sandboxPulled := make(chan struct{}) + go func() { // first pods would otherwise pull the sandbox image themselves + defer close(sandboxPulled) + pullCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + _, _ = containerExec(pullCtx, cli, created.ID, "", nil, "sh", "-c", + `ctr --namespace=k8s.io images pull "docker.io/$(sed -n 's/^ *sandbox = "\(.*\)"$/\1/p' /var/lib/rancher/k3s/agent/etc/containerd/config.toml)"`) + }() + images, err := preloadImages(ctx, created.ID, os.Getenv("E2E_JOB_IMAGE"), os.Getenv("SERVICE_IMAGE")) + <-sandboxPulled + if err != nil { + return nil, err } - images := preloadImages(t, created.ID, os.Getenv("E2E_JOB_IMAGE"), os.Getenv("SERVICE_IMAGE")) - waitCtx, cancel := context.WithTimeout(ctx, 2*time.Minute) - defer cancel() for { - _, err := containerExec(waitCtx, cli, created.ID, "", nil, "kubectl", "get", "serviceaccount", "default") + _, err := containerExec(ctx, cli, created.ID, "", nil, "kubectl", "get", "serviceaccount", "default") if err == nil { break } inspected, inspectErr := cli.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{}) - if inspectErr != nil || !inspected.Container.State.Running || waitCtx.Err() != nil { + if inspectErr != nil || !inspected.Container.State.Running || ctx.Err() != nil { var logs bytes.Buffer - if reader, logsErr := cli.ContainerLogs(ctx, created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil { + if reader, logsErr := cli.ContainerLogs(context.Background(), created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil { _, _ = stdcopy.StdCopy(&logs, &logs, reader) _ = reader.Close() } - t.Fatalf("k3s did not start: %v, state %+v %v, last log lines:\n%s", err, inspected.Container.State, inspectErr, logs.String()) + return nil, fmt.Errorf("k3s did not start: %w, state %+v %v, last log lines:\n%s", err, inspected.Container.State, inspectErr, logs.String()) } select { - case <-waitCtx.Done(): + case <-ctx.Done(): case <-time.After(pollInterval): } } kubeconfig, err := containerExec(ctx, cli, created.ID, "", nil, "cat", "/etc/rancher/k3s/k3s.yaml") if err != nil { - t.Fatalf("read kubeconfig: %v", err) + return nil, fmt.Errorf("read kubeconfig: %w", err) } - path := filepath.Join(t.TempDir(), "kubeconfig") + path := filepath.Join(binDir, "kubeconfig") if err := os.WriteFile(path, []byte(strings.ReplaceAll(kubeconfig, "https://127.0.0.1:6443", server)), 0o600); err != nil { - t.Fatalf("write kubeconfig: %v", err) + return nil, fmt.Errorf("write kubeconfig: %w", err) } - kubernetes := &config.Kubernetes{Kubeconfig: path, Namespace: "default"} - - if fixture.network != "" { - gitea, err := cli.ContainerInspect(ctx, fixture.id, mobyclient.ContainerInspectOptions{}) - if err != nil { - t.Fatalf("inspect gitea container: %v", err) - } - giteaURL, err := url.Parse(fixture.baseURL) - if err != nil { - t.Fatalf("parse gitea url: %v", err) - } - kubernetes.PodTemplate = map[string]any{"spec": map[string]any{"hostAliases": []any{map[string]any{ - "ip": gitea.Container.NetworkSettings.Networks[fixture.network].IPAddress.String(), - "hostnames": []any{giteaURL.Hostname()}, - }}}} - } - return created.ID, kubernetes, images + return &kubeCluster{id: created.ID, kubernetes: &config.Kubernetes{Kubeconfig: path, Namespace: "default"}, jobImage: images[0], serviceImage: images[1]}, nil } func waitForContainerLog(ctx context.Context, id, message string) error { - logs, err := fixture.cli.ContainerLogs(ctx, id, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true}) + logs, err := dockerCLI.ContainerLogs(ctx, id, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true}) if err != nil { return err } @@ -175,25 +219,23 @@ func waitForContainerLog(ctx context.Context, id, message string) error { return fmt.Errorf("container exited before logging %q", message) } -func preloadImages(t *testing.T, k3sID string, images ...string) []string { - t.Helper() - ctx := t.Context() +func preloadImages(ctx context.Context, k3sID string, images ...string) ([]string, error) { var aliases []string for index, image := range images { alias := fmt.Sprintf("gitea-runner-e2e/preload-%d:%d", index, time.Now().UnixNano()) - if _, err := fixture.cli.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil { - t.Fatalf("tag %s: %v", image, err) + if _, err := dockerCLI.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil { + return nil, fmt.Errorf("tag %s: %w", image, err) } - t.Cleanup(func() { _, _ = fixture.cli.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) }) + clusterCleanup = append(clusterCleanup, func() { _, _ = dockerCLI.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) }) aliases = append(aliases, alias) } - archive, err := fixture.cli.ImageSave(ctx, aliases) + archive, err := dockerCLI.ImageSave(ctx, aliases) if err != nil { - t.Fatalf("save %v: %v", images, err) + return nil, fmt.Errorf("save %v: %w", images, err) } defer archive.Close() - if out, err := containerExec(ctx, fixture.cli, k3sID, "", archive, "ctr", "--namespace=k8s.io", "images", "import", "-"); err != nil { - t.Fatalf("import %v into k3s: %v: %s", images, err, out) + if out, err := containerExec(ctx, dockerCLI, k3sID, "", archive, "ctr", "--namespace=k8s.io", "images", "import", "-"); err != nil { + return nil, fmt.Errorf("import %v into k3s: %w: %s", images, err, out) } - return aliases + return aliases, nil } diff --git a/e2e/main_test.go b/e2e/main_test.go index 8eff7ce0..5500987c 100644 --- a/e2e/main_test.go +++ b/e2e/main_test.go @@ -9,18 +9,24 @@ import ( "cmp" "context" "errors" + "flag" "fmt" + "net" "net/http" "os" "os/exec" "path/filepath" "regexp" + "strconv" "strings" "sync" + "sync/atomic" "testing" "time" "gitea.com/gitea/runner/internal/app/poll" + + mobyclient "github.com/moby/moby/client" ) const runTimeout = 3 * time.Minute @@ -33,6 +39,8 @@ var skipReason string var binDir string +var dockerCLI mobyclient.APIClient + var runnerBinary = sync.OnceValues(func() (string, error) { bin := filepath.Join(binDir, "gitea-runner") if out, err := exec.Command(cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil { @@ -54,6 +62,17 @@ func runSuite(m *testing.M) int { skipReason = fmt.Sprintf("docker unavailable: %v", err) return m.Run() } + dockerCLI = cli + + if binDir, err = os.MkdirTemp("", "gitea-runner-e2e-"); err != nil { + fmt.Fprintf(os.Stderr, "create binary dir: %v\n", err) + return 1 + } + defer os.RemoveAll(binDir) + defer stopCluster() + if flag.Parse(); flag.Lookup("test.run").Value.String() == "" { // a filtered run starts the cluster only if it needs it + go func() { _, _ = k3s() }() + } f, err := StartGitea(ctx, cli) if err != nil { @@ -63,15 +82,14 @@ func runSuite(m *testing.M) int { fixture = f defer func() { _ = fixture.Close(context.Background()) }() - if binDir, err = os.MkdirTemp("", "gitea-runner-e2e-"); err != nil { - fmt.Fprintf(os.Stderr, "create binary dir: %v\n", err) - return 1 - } - defer os.RemoveAll(binDir) - fmt.Fprintf(os.Stderr, "gitea fixture: image=%s version=%s\n", fixture.image, fixture.version) - return m.Run() + code := m.Run() + if err := checkClusterLeaks(); err != nil && code == 0 { + fmt.Fprintln(os.Stderr, err) + code = 1 + } + return code } func newScenario(t *testing.T) (*GiteaAPI, string) { @@ -162,6 +180,24 @@ func dumpRunLogs(t *testing.T, api *GiteaAPI, repo string, runID int64) { } } +// newGate serves a URL that answers 503 until release, for jobs to poll instead of sleeping. +func newGate(t *testing.T) (string, func()) { + t.Helper() + var released atomic.Bool + listener, err := net.Listen("tcp", ":0") + if err != nil { + t.Fatalf("listen for the gate: %v", err) + } + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if !released.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + } + }), ReadHeaderTimeout: time.Second} + go func() { _ = server.Serve(listener) }() + t.Cleanup(func() { _ = server.Close() }) + return fmt.Sprintf("http://%s/", net.JoinHostPort(fixture.selfIP, strconv.Itoa(listener.Addr().(*net.TCPAddr).Port))), func() { released.Store(true) } +} + func waitForRun(t *testing.T, api *GiteaAPI, repo string) *ActionRun { t.Helper() return waitForRunAfter(t, api, repo, 0) diff --git a/e2e/modes_test.go b/e2e/modes_test.go new file mode 100644 index 00000000..72841b1f --- /dev/null +++ b/e2e/modes_test.go @@ -0,0 +1,227 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +//go:build e2e + +package e2e + +import ( + "bufio" + "bytes" + "cmp" + "encoding/json" + "errors" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "testing" +) + +type mode struct { + exec, kubernetes bool +} + +var allModes = []mode{{}, {kubernetes: true}, {exec: true}, {exec: true, kubernetes: true}} + +func (m mode) String() string { + driver, backend := "daemon", "docker" + if m.exec { + driver = "exec" + } + if m.kubernetes { + backend = "kubernetes" + } + return driver + "_" + backend +} + +func inAllModes(scenario func(*testing.T, mode)) func(*testing.T) { + return func(t *testing.T) { + t.Parallel() + for _, m := range allModes { + t.Run(m.String(), func(t *testing.T) { + t.Parallel() + scenario(t, m) + }) + } + } +} + +type workflowSpec struct { + files []string + inputs map[string]string // dispatches files[0] instead of relying on push + execArgs []string +} + +type workflowResult struct { + conclusion string + jobs map[string]string + lines []string +} + +func (r *workflowResult) job(id string) string { + return cmp.Or(r.jobs[id], "skipped") // exec reports nothing for a job it skipped +} + +func (r *workflowResult) hasLine(line string) bool { + return slices.Contains(r.lines, line) +} + +func runWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { + t.Helper() + run := daemonWorkflow + if m.exec { + run = execWorkflow + } + result := run(t, m, spec) + t.Cleanup(func() { + if t.Failed() { + t.Logf("%s concluded %q, jobs %v, logs:\n%s", m, result.conclusion, result.jobs, strings.Join(result.lines, "\n")) + } + }) + return result +} + +func daemonWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { + t.Helper() + ctx := t.Context() + + api, repo := newScenario(t) + if m.kubernetes { + cluster := requireCluster(t) + startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: cluster.jobImage, kubernetes: cluster.kubernetes, capacity: 16}) + } + files := map[string]string{} + for _, file := range spec.files { + content, err := os.ReadFile(filepath.Join("testdata", "workflows", file)) + if err != nil { + t.Fatalf("read workflow fixture %s: %v", file, err) + } + if m.kubernetes { + content = bytes.ReplaceAll(content, []byte("runs-on: ubuntu-latest"), []byte("runs-on: e2e-kubernetes")) + } + files[".gitea/workflows/"+file] = string(content) + } + if err := api.CreateFiles(ctx, repo, files, "add workflows"); err != nil { + t.Fatalf("push workflows: %v", err) + } + if spec.inputs != nil { + branch, err := api.DefaultBranch(ctx, repo) + if err != nil { + t.Fatalf("get default branch: %v", err) + } + if err := waitForDispatch(ctx, api, repo, spec.files[0], branch, spec.inputs); err != nil { + t.Fatalf("dispatch workflow: %v", err) + } + } + + wfRun := waitForRun(t, api, repo) + completed, err := api.WaitForRunConclusion(ctx, repo, wfRun.ID, runTimeout) + if err != nil { + dumpRunLogs(t, api, repo, wfRun.ID) + t.Fatalf("wait for run: %v", err) + } + jobs, err := api.Jobs(ctx, repo, wfRun.ID) + if err != nil { + t.Fatalf("list jobs: %v", err) + } + result := &workflowResult{conclusion: completed.Conclusion, jobs: map[string]string{}} + for _, job := range jobs { + result.jobs[job.Name] = job.Conclusion + } + for line := range strings.SplitSeq(runLogs(t, api, repo, wfRun.ID), "\n") { + if _, payload, found := strings.Cut(line, "Z "); found { + result.lines = append(result.lines, strings.TrimRight(payload, "\r")) + } + } + return result +} + +func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { + t.Helper() + ctx := t.Context() + + bin, err := runnerBinary() + if err != nil { + t.Fatal(err) + } + workdir := t.TempDir() + workflows := filepath.Join(workdir, ".gitea", "workflows") + if err := os.MkdirAll(workflows, 0o755); err != nil { + t.Fatal(err) + } + for _, file := range spec.files { + content, err := os.ReadFile(filepath.Join("testdata", "workflows", file)) + if err != nil { + t.Fatalf("read workflow fixture %s: %v", file, err) + } + if err := os.WriteFile(filepath.Join(workflows, file), content, 0o644); err != nil { + t.Fatal(err) + } + } + for _, args := range [][]string{ + {"init", "-q"}, + {"remote", "add", "origin", "https://example.com/e2e/exec.git"}, + {"-c", "user.name=e2e", "-c", "user.email=e2e@example.com", "commit", "-q", "--allow-empty", "-m", "init"}, + } { + if out, err := exec.CommandContext(ctx, "git", append([]string{"-C", workdir}, args...)...).CombinedOutput(); err != nil { + t.Fatalf("git %v: %v\n%s", args, err, out) + } + } + + args := []string{"exec", "--json", "-C", workdir, "-W", workflows, "--artifact-server-path", t.TempDir(), "--artifact-server-port", "0"} + env := slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") }) + if m.kubernetes { + cluster := requireCluster(t) + args = append(args, "-i", "kubernetes://"+cluster.jobImage) + env = append(env, "KUBECONFIG="+cluster.kubernetes.Kubeconfig) + } else { + args = append(args, "-i", os.Getenv("E2E_JOB_IMAGE"), "--network", fixture.network) + } + if spec.inputs != nil { + args = append(args, "-E", "workflow_dispatch") + for name, value := range spec.inputs { + args = append(args, "--input", name+"="+value) + } + } + cmd := exec.CommandContext(ctx, bin, append(args, spec.execArgs...)...) + cmd.Env = env + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + runErr := cmd.Run() + var exitErr *exec.ExitError + if runErr != nil && !errors.As(runErr, &exitErr) { + t.Fatalf("exec: %v\n%s", runErr, stderr.String()) + } + + result := &workflowResult{conclusion: "success", jobs: map[string]string{}} + if runErr != nil { + result.conclusion = "failure" + } + scanner := bufio.NewScanner(&stdout) + scanner.Buffer(nil, 1<<20) + for scanner.Scan() { + var entry struct { + JobID string `json:"jobID"` + JobResult string `json:"jobResult"` + Msg string `json:"msg"` + RawOutput bool `json:"raw_output"` + } + if json.Unmarshal(scanner.Bytes(), &entry) != nil { + continue + } + if entry.JobResult != "" { + result.jobs[entry.JobID] = entry.JobResult + } + if entry.RawOutput { + for line := range strings.SplitSeq(strings.TrimRight(entry.Msg, "\n"), "\n") { + result.lines = append(result.lines, strings.TrimRight(line, "\r")) + } + } + } + if result.conclusion != "success" && len(result.jobs) == 0 { + t.Fatalf("exec ran no job: %v\n%s", runErr, stderr.String()) + } + return result +} diff --git a/e2e/reconnect_test.go b/e2e/reconnect_test.go index 346bfec6..1512c267 100644 --- a/e2e/reconnect_test.go +++ b/e2e/reconnect_test.go @@ -10,6 +10,8 @@ import ( "io" "net" "net/url" + "strconv" + "strings" "sync" "sync/atomic" "testing" @@ -101,6 +103,10 @@ func (p *flakyProxy) restore() { p.down.Store(false) } func testReconnect(t *testing.T) { t.Parallel() api, repo := newScenario(t) + gate, release := newGate(t) + if err := api.CreateVariable(t.Context(), repo, "E2E_GATE", gate); err != nil { + t.Fatalf("create gate variable: %v", err) + } proxy := newFlakyProxy(t, fixture.baseURL) startRunner(t, repo, "e2e-reconnect", runnerOptions{host: true, address: proxy.URL()}) @@ -112,14 +118,19 @@ func testReconnect(t *testing.T) { 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 + waitForRunningJobLog(t, api, repo, slow.ID, "e2e-before-outage", 1) + // the job logs a line every 200ms until released, so the outage spans several that can only arrive after it proxy.cut(t, 8) proxy.restore() + release() 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 { + ticks, err := strconv.Atoi(strings.TrimPrefix(commandRow(logs, "ticks="), "ticks=")) + if err != nil { + t.Fatalf("the job did not report its tick count: %v\n%s", err, logs) + } + for i := 1; i <= ticks; i++ { + if tick := fmt.Sprintf("tick-%d", 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 index 0f0820aa..4ecf735f 100644 --- a/e2e/reusable_test.go +++ b/e2e/reusable_test.go @@ -8,15 +8,12 @@ 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) +func testReusableWorkflow(t *testing.T, m mode) { + result := runWorkflow(t, m, workflowSpec{ + files: []string{"reusable_callee.yml", "reusable_caller.yml"}, + execArgs: []string{"--no-skip-checkout"}, + }) + if result.conclusion != "success" || !result.hasLine("e2e-reusable-ran") { + t.Errorf("run concluded %q, want success with the called workflow's marker", result.conclusion) } } diff --git a/e2e/runner_harness.go b/e2e/runner_harness.go index fe832bc9..e6950d4f 100644 --- a/e2e/runner_harness.go +++ b/e2e/runner_harness.go @@ -33,6 +33,7 @@ type runnerOptions struct { 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 + timeout time.Duration } func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *poll.Poller { @@ -60,9 +61,8 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p cfg.Runner.StateReportInterval = 500 * time.Millisecond // so cancel reaches the job quickly cfg.Runner.LogReportInterval = 500 * time.Millisecond cfg.Runner.Capacity = max(options.capacity, 1) - if fixture.network != "" { - cfg.Container.Network = fixture.network - } + cfg.Runner.Timeout = cmp.Or(options.timeout, cfg.Runner.Timeout) + cfg.Container.Network = fixture.network scheme := labels.SchemeDocker if options.kubernetes != nil { diff --git a/e2e/scheduling_test.go b/e2e/scheduling_test.go index b639a152..43e3c217 100644 --- a/e2e/scheduling_test.go +++ b/e2e/scheduling_test.go @@ -75,10 +75,16 @@ func testCapacity(t *testing.T) { t.Run(tc.name, func(t *testing.T) { t.Parallel() api, repo := newScenario(t) + gate, release := newGate(t) + if err := api.CreateVariable(t.Context(), repo, "E2E_GATE", gate); err != nil { + t.Fatalf("create gate variable: %v", err) + } startRunner(t, repo, "e2e-capacity", runnerOptions{capacity: tc.capacity}) pushWorkflow(t, api, repo, "capacity.yml") wfRun := waitForRun(t, api, repo) + waitForRunningJobLog(t, api, repo, wfRun.ID, "span-start=", tc.capacity) + release() requireSuccess(t, api, repo, wfRun.ID) jobs, err := api.Jobs(t.Context(), repo, wfRun.ID) diff --git a/e2e/testdata/workflows/capacity.yml b/e2e/testdata/workflows/capacity.yml index 108fbd2e..468de93e 100644 --- a/e2e/testdata/workflows/capacity.yml +++ b/e2e/testdata/workflows/capacity.yml @@ -6,12 +6,12 @@ jobs: steps: - run: | echo "span-start=$(date +%s%N)" - sleep 8 + until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do sleep 0.1; done echo "span-end=$(date +%s%N)" two: runs-on: e2e-capacity steps: - run: | echo "span-start=$(date +%s%N)" - sleep 8 + until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do sleep 0.1; done echo "span-end=$(date +%s%N)" diff --git a/e2e/testdata/workflows/cli_shutdown.yml b/e2e/testdata/workflows/cli_shutdown.yml index 4b4f718c..f471602d 100644 --- a/e2e/testdata/workflows/cli_shutdown.yml +++ b/e2e/testdata/workflows/cli_shutdown.yml @@ -6,5 +6,5 @@ jobs: steps: - run: | echo e2e-shutdown-started - sleep 8 + until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do sleep 0.1; done echo e2e-shutdown-finished diff --git a/e2e/testdata/workflows/failure.yml b/e2e/testdata/workflows/failure.yml index 7d9516f4..cd5ecc8b 100644 --- a/e2e/testdata/workflows/failure.yml +++ b/e2e/testdata/workflows/failure.yml @@ -21,10 +21,3 @@ jobs: runs-on: ubuntu-latest steps: - run: echo dependent - timeout: - runs-on: ubuntu-latest - timeout-minutes: 1 - steps: - - run: | - echo e2e-timeout-started - sleep 300 diff --git a/e2e/testdata/workflows/reconnect.yml b/e2e/testdata/workflows/reconnect.yml index 4b966476..9b3697fc 100644 --- a/e2e/testdata/workflows/reconnect.yml +++ b/e2e/testdata/workflows/reconnect.yml @@ -9,7 +9,10 @@ jobs: steps: - run: | echo e2e-before-outage - for i in $(seq -w 1 15); do + i=0 + until curl -fs "${{ vars.E2E_GATE }}" > /dev/null; do + i=$((i + 1)) echo "tick-$i" - sleep 1 + sleep 0.2 done + echo "ticks=$i" diff --git a/e2e/testdata/workflows/timeout.yml b/e2e/testdata/workflows/timeout.yml new file mode 100644 index 00000000..6225a583 --- /dev/null +++ b/e2e/testdata/workflows/timeout.yml @@ -0,0 +1,9 @@ +name: timeout +on: push +jobs: + timeout: + runs-on: e2e-timeout + steps: + - run: | + echo e2e-timeout-started + sleep 300 diff --git a/tools/test-e2e.sh b/tools/test-e2e.sh index 80173397..9e9bd89f 100755 --- a/tools/test-e2e.sh +++ b/tools/test-e2e.sh @@ -23,4 +23,16 @@ for pull in "${pulls[@]}"; do done ((result == 0)) || exit "$result" -exec "${GO:-go}" test -tags e2e -count=1 -parallel "$E2E_CONCURRENCY" -timeout 20m -v ./e2e/... +args=(test -tags e2e -count=1 -parallel "$E2E_CONCURRENCY" -timeout 20m -v "$@" ./e2e/...) +if [[ -f /.dockerenv ]]; then + exec "${GO:-go}" "${args[@]}" +fi + +# Run like CI: in a Linux container on a user-defined network, which every container of the suite joins. +network="gitea-runner-e2e-$$" +docker network create "$network" > /dev/null +trap 'docker network rm "$network" > /dev/null || true' EXIT +docker run --rm --network "$network" -w "$PWD" -v "$PWD:$PWD" -v /var/run/docker.sock:/var/run/docker.sock \ + -v gitea-runner-e2e-gomod:/go/pkg/mod -v gitea-runner-e2e-gocache:/root/.cache/go-build \ + -e E2E_CONCURRENCY -e E2E_JOB_IMAGE -e E2E_GITEA_IMAGE -e E2E_K3S_IMAGE -e SERVICE_IMAGE \ + "golang:$(sed -n 's/^toolchain go//p' go.mod)" go "${args[@]}"