mirror of
https://gitea.com/gitea/act_runner.git
synced 2026-11-03 05:45:12 +08:00
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 <zettat123@gmail.com> Co-authored-by: bircni <bircni@icloud.com> Reviewed-on: https://gitea.com/gitea/runner/pulls/1347 Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com>
This commit is contained in:
@@ -59,8 +59,10 @@ Each lives in `internal/pkg/action/<name>` 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 `<workflow>@<ref>` and a log row is `<timestamp>Z <payload>`.
|
||||
|
||||
224
e2e/cli_test.go
224
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")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
33
e2e/host_test.go
Normal file
33
e2e/host_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
127
e2e/reconnect_test.go
Normal file
127
e2e/reconnect_test.go
Normal file
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
22
e2e/reusable_test.go
Normal file
22
e2e/reusable_test.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
113
e2e/scheduling_test.go
Normal file
113
e2e/scheduling_test.go
Normal file
@@ -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
|
||||
}
|
||||
17
e2e/testdata/workflows/capacity.yml
vendored
Normal file
17
e2e/testdata/workflows/capacity.yml
vendored
Normal file
@@ -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)"
|
||||
7
e2e/testdata/workflows/cli_labels.yml
vendored
Normal file
7
e2e/testdata/workflows/cli_labels.yml
vendored
Normal file
@@ -0,0 +1,7 @@
|
||||
name: cli-labels
|
||||
on: push
|
||||
jobs:
|
||||
hello:
|
||||
runs-on: e2e-override
|
||||
steps:
|
||||
- run: echo e2e-override-ran
|
||||
11
e2e/testdata/workflows/cli_once.yml
vendored
Normal file
11
e2e/testdata/workflows/cli_once.yml
vendored
Normal file
@@ -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
|
||||
10
e2e/testdata/workflows/cli_shutdown.yml
vendored
Normal file
10
e2e/testdata/workflows/cli_shutdown.yml
vendored
Normal file
@@ -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
|
||||
10
e2e/testdata/workflows/host.yml
vendored
Normal file
10
e2e/testdata/workflows/host.yml
vendored
Normal file
@@ -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
|
||||
15
e2e/testdata/workflows/reconnect.yml
vendored
Normal file
15
e2e/testdata/workflows/reconnect.yml
vendored
Normal file
@@ -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
|
||||
10
e2e/testdata/workflows/reconnect_idle.yml
vendored
Normal file
10
e2e/testdata/workflows/reconnect_idle.yml
vendored
Normal file
@@ -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
|
||||
20
e2e/testdata/workflows/reusable_callee.yml
vendored
Normal file
20
e2e/testdata/workflows/reusable_callee.yml
vendored
Normal file
@@ -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
|
||||
12
e2e/testdata/workflows/reusable_caller.yml
vendored
Normal file
12
e2e/testdata/workflows/reusable_caller.yml
vendored
Normal file
@@ -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"
|
||||
15
e2e/testdata/workflows/route.yml
vendored
Normal file
15
e2e/testdata/workflows/route.yml
vendored
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user