diff --git a/.dockerignore b/.dockerignore index cc2ea19a..9b34a61f 100644 --- a/.dockerignore +++ b/.dockerignore @@ -45,8 +45,13 @@ cpu.out /bin /dist +/.tmp +/.git/hooks +/.git/logs +/.git/rr-cache /.env /.runner +/.runner.lock /config.yaml /Dockerfile .DS_Store diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index f1d4db9e..5d4c3e5e 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -22,17 +22,19 @@ jobs: run: mkdir -p "$DOCKER_CONFIG" && echo '{}' > "$DOCKER_CONFIG/config.json" # Pre-pull act/runner's two largest base images so a slow pull can't dominate `make test`; # the rest (alpine/ubuntu) pull on demand, absorbed by the make-test -timeout. The host - # daemon retains them between runs, so this is usually a fast manifest re-check. + # daemon retains them between runs, and their digest pins skip the registry when it has them. - name: pre-pull test images env: TEST_JOB_IMAGE: node:24-bookworm-slim@sha256:55fb5f5a527ea2c5c4a95f963e48ef90aa38c7b346054fdc8ee9957a2b55d205 # renovate: datasource=docker TEST_SERVICE_IMAGE: nginx:alpine@sha256:a3554c0d0b48177b1413c26e01f695334428020d9df01243fbae3e11cfd17417 # renovate: datasource=docker run: | for image in "$TEST_JOB_IMAGE" "$TEST_SERVICE_IMAGE"; do - for attempt in 1 2 3; do - docker pull "$image" && break - [ "$attempt" = 3 ] || sleep 5 - done + if [ "$(docker image inspect --format '{{.Os}}/{{.Architecture}}' "$image" 2>/dev/null)" != linux/amd64 ]; then + for attempt in 1 2 3; do + docker pull "$image" && break + [ "$attempt" = 3 ] || sleep 5 + done + fi docker tag "$image" "${image%@*}" done - name: lint diff --git a/Dockerfile b/Dockerfile index d6214006..b956a5fe 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,7 +12,9 @@ ENV GOPROXY=${GOPROXY:-} COPY . /opt/src/runner WORKDIR /opt/src/runner -RUN make clean && make build +RUN --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + make clean && make build ### DIND VARIANT # diff --git a/act/artifactcache/handler_test.go b/act/artifactcache/handler_test.go index ff0477cf..ae731c74 100644 --- a/act/artifactcache/handler_test.go +++ b/act/artifactcache/handler_test.go @@ -48,10 +48,12 @@ func signArtifactURL(h *Handler, id int64) string { } func TestHandler_ExternalURL(t *testing.T) { + t.Parallel() assert.Equal(t, "http://[2001:db8::1]:8080", (&Handler{outboundIP: "2001:db8::1", port: 8080}).ExternalURL()) } func TestHandler_Upstream(t *testing.T) { + t.Parallel() var upstream *httptest.Server upstream = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { assert.Equal(t, http.MethodPut, r.Method) @@ -101,6 +103,7 @@ func TestHandler_Upstream(t *testing.T) { } func TestHandler(t *testing.T) { + t.Parallel() const version = "c19da02a2bd7e77277f1ac29ab45c09b7d46a4ee758284e26bb3045ad11d9d20" dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) @@ -470,6 +473,7 @@ func uploadCacheNormally(t *testing.T, base, key, version string, content []byte } func TestHandler_gcCache(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir, Policy: Policy{Retention: testRetention}}) require.NoError(t, err) @@ -572,6 +576,7 @@ func TestHandler_gcCache(t *testing.T) { } func TestHandler_evictPolicy(t *testing.T) { + t.Parallel() now := time.Now() stale := func(d time.Duration) int64 { return now.Add(-d).Unix() } mib := func(n int64) int64 { return n * miB } @@ -669,6 +674,7 @@ func TestHandler_evictPolicy(t *testing.T) { } func TestHandler_evictForFreeSpace(t *testing.T) { + t.Parallel() free := func(n int64) func(string) (uint64, error) { return func(string) (uint64, error) { return uint64(n), nil } } @@ -708,6 +714,7 @@ func TestHandler_evictForFreeSpace(t *testing.T) { } func TestHandler_SweepKeepsEntryWhenBlobSurvives(t *testing.T) { + t.Parallel() cache := &Cache{Key: "stuck", Complete: true, UsedAt: time.Now().Add(-(testRetention + time.Hour)).Unix()} handler := newTestHandler(t, Policy{Retention: testRetention}, cache) @@ -722,6 +729,7 @@ func TestHandler_SweepKeepsEntryWhenBlobSurvives(t *testing.T) { } func TestHandler_FindProtectsFromEviction(t *testing.T) { + t.Parallel() wanted := &Cache{Repo: testRepo, Key: "wanted", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-3 * time.Hour).Unix()} other := &Cache{Repo: testRepo, Key: "other", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-2 * time.Hour).Unix()} newest := &Cache{Repo: testRepo, Key: "newest", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()} @@ -742,6 +750,7 @@ func TestHandler_FindProtectsFromEviction(t *testing.T) { } func TestHandler_PrefixUsesNewestReservationWithinSecond(t *testing.T) { + t.Parallel() created := time.Now().Add(-time.Hour).Unix() older := &Cache{Repo: testRepo, Key: "deps-old", Version: "v", Complete: true, CreatedAt: created} newer := &Cache{Repo: testRepo, Key: "deps-new", Version: "v", Complete: true, CreatedAt: created} @@ -756,6 +765,7 @@ func TestHandler_PrefixUsesNewestReservationWithinSecond(t *testing.T) { } func TestHandler_evictOnCommit(t *testing.T) { + t.Parallel() full := &Cache{Repo: testRepo, Key: "full", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()} handler := newTestHandler(t, Policy{RepoSizeLimit: 4 * miB}, full) @@ -765,6 +775,7 @@ func TestHandler_evictOnCommit(t *testing.T) { } func TestHandler_gcCacheInterval(t *testing.T) { + t.Parallel() cache := &Cache{Key: "temp", UsedAt: time.Now().Add(-time.Hour).Unix()} handler := newTestHandler(t, Policy{SweepInterval: 30 * time.Minute}, cache) @@ -774,6 +785,7 @@ func TestHandler_gcCacheInterval(t *testing.T) { } func TestHandler_touchCacheSkipsFreshCompletedEntries(t *testing.T) { + t.Parallel() fresh := &Cache{Repo: testRepo, Key: "fresh", Version: "v", Complete: true, UsedAt: time.Now().Unix()} pending := &Cache{Repo: testRepo, Key: "pending", Version: "v", UsedAt: time.Now().Add(-time.Second).Unix()} handler := newTestHandler(t, Policy{}, fresh, pending) @@ -840,6 +852,7 @@ func writeBlob(t *testing.T, handler *Handler, id uint64) { } func TestHandler_RejectsBearer(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) @@ -876,6 +889,7 @@ func TestHandler_RejectsBearer(t *testing.T) { } func TestHandler_UnregisterRevokes(t *testing.T) { + t.Parallel() for _, tc := range []struct { name string token string @@ -917,6 +931,7 @@ func TestHandler_UnregisterRevokes(t *testing.T) { } func TestHandler_CrossRepoIsolation(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) @@ -967,6 +982,7 @@ func TestHandler_CrossRepoIsolation(t *testing.T) { } func TestHandler_ArtifactSignature(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) @@ -998,6 +1014,7 @@ func TestHandler_ArtifactSignature(t *testing.T) { } func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") first, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"}) @@ -1014,6 +1031,7 @@ func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) { } func TestHandler_SeesWritesOfOtherProcesses(t *testing.T) { + t.Parallel() handler := newTestHandler(t, Policy{}) insert := func(by *Handler, cache *Cache) { db, err := by.openDB() @@ -1031,6 +1049,7 @@ func TestHandler_SeesWritesOfOtherProcesses(t *testing.T) { } func TestHandler_ArtifactSignatureDownload(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) @@ -1065,6 +1084,7 @@ func TestHandler_ArtifactSignatureDownload(t *testing.T) { } func TestHandler_GC_PerRepoDedup(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) @@ -1109,6 +1129,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) { } func TestHandler_InternalAPI_Disabled(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") handler, err := StartHandler(Options{Dir: dir}) require.NoError(t, err) @@ -1123,6 +1144,7 @@ func TestHandler_InternalAPI_Disabled(t *testing.T) { } func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) { + t.Parallel() dir := filepath.Join(t.TempDir(), "artifactcache") const secret = "internal-secret" handler, err := StartHandler(Options{Dir: dir, InternalSecret: secret}) diff --git a/act/artifactcache/handler_v2_test.go b/act/artifactcache/handler_v2_test.go index 97cd254e..422b2849 100644 --- a/act/artifactcache/handler_v2_test.go +++ b/act/artifactcache/handler_v2_test.go @@ -78,6 +78,7 @@ func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) } func TestCacheServiceV2RoundTrip(t *testing.T) { + t.Parallel() handler := newTestHandler(t, Policy{}) content := []byte("the cached archive") @@ -103,6 +104,7 @@ func TestCacheServiceV2RoundTrip(t *testing.T) { } func TestCacheServiceV2BlockUpload(t *testing.T) { + t.Parallel() handler := newTestHandler(t, Policy{}) created := v2Call(t, handler, testClient, "CreateCacheEntry", map[string]any{"key": "blocks", "version": "v1"}) @@ -150,6 +152,7 @@ func TestCacheServiceV2BlockUpload(t *testing.T) { } func TestCacheServiceV2Lookups(t *testing.T) { + t.Parallel() handler := newTestHandler(t, Policy{}) saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x")) require.Equal(t, true, saved["ok"]) @@ -231,6 +234,7 @@ func TestCacheServiceV2Lookups(t *testing.T) { } func TestCacheServiceV2RefusalsAreTwirp(t *testing.T) { + t.Parallel() handler := newTestHandler(t, Policy{}) tests := []struct { @@ -267,6 +271,7 @@ func TestCacheServiceV2RefusalsAreTwirp(t *testing.T) { } func TestCacheServiceV2ConcurrentSaveAndRetries(t *testing.T) { + t.Parallel() handler := newTestHandler(t, Policy{}) content := []byte("the cached archive") diff --git a/act/artifactcache/results_test.go b/act/artifactcache/results_test.go index 993e7982..2863c8cb 100644 --- a/act/artifactcache/results_test.go +++ b/act/artifactcache/results_test.go @@ -17,6 +17,7 @@ import ( ) func TestFrontResultsService(t *testing.T) { + t.Parallel() var gotHost, gotPath, gotProto string gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { gotHost, gotPath, gotProto = r.Host, r.URL.Path, r.Header.Get("X-Forwarded-Proto") diff --git a/act/artifactcache/s3_storage_test.go b/act/artifactcache/s3_storage_test.go index c449a045..c4b64341 100644 --- a/act/artifactcache/s3_storage_test.go +++ b/act/artifactcache/s3_storage_test.go @@ -20,6 +20,7 @@ import ( ) func TestHandler_S3SharesEntriesAndDropsExpiredOnes(t *testing.T) { + t.Parallel() var objects sync.Map var uploads atomic.Int32 server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { @@ -91,6 +92,7 @@ func (transport s3TestTransport) RoundTrip(req *http.Request) (*http.Response, e } func TestS3StorageUploadsLargeObjectsInPartsAndAbortsOnFailure(t *testing.T) { + t.Parallel() for _, failingPart := range []string{"", "2"} { var parts, aborts atomic.Int32 var completion struct { @@ -130,6 +132,7 @@ func TestS3StorageUploadsLargeObjectsInPartsAndAbortsOnFailure(t *testing.T) { } func TestS3StorageSignsLikeAWS(t *testing.T) { + t.Parallel() storage := &s3Storage{opts: S3Options{Region: "us-east-1", AccessKey: "AKIAIOSFODNN7EXAMPLE", SecretKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}} req := httptest.NewRequest(http.MethodGet, "https://examplebucket.s3.amazonaws.com/?max-keys=2&prefix=J", nil) req.Header.Set("X-Amz-Content-Sha256", "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855") diff --git a/act/common/git/fetch_action_test.go b/act/common/git/fetch_action_test.go index 643055bc..85a68bec 100644 --- a/act/common/git/fetch_action_test.go +++ b/act/common/git/fetch_action_test.go @@ -42,6 +42,7 @@ func TestFetchAction(t *testing.T) { require.NoError(t, gitCmd("-C", remote, "add", ".")) require.NoError(t, gitCmd("-C", remote, "commit", "-m", "initial")) require.NoError(t, gitCmd("-C", remote, "tag", "v1")) + expectedSha := gitRevParse(t, remote, "main") gitPath, err := exec.LookPath("git") require.NoError(t, err) @@ -49,6 +50,7 @@ func TestFetchAction(t *testing.T) { var mu sync.Mutex var archiveRequests []http.Header var mirror *httptest.Server + archiveCache := map[string][]byte{} requests, archives, failures := 0, "", 0 handler := http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { mu.Lock() @@ -85,7 +87,12 @@ func TestFetchAction(t *testing.T) { if archives != "unprefixed" { args = append(args, "--prefix=action/") } - archive, _ := runGit(request.Context(), remote, nil, args...) + archiveKey := strings.Join(args, "\x00") + archive, ok := archiveCache[archiveKey] + if !ok { + archive, _ = runGit(request.Context(), remote, nil, args...) + archiveCache[archiveKey] = archive + } writer.Header().Set("ETag", etag) _, _ = writer.Write(archive) }) @@ -108,7 +115,6 @@ func TestFetchAction(t *testing.T) { } t.Run("shares identical fetches and isolates cancellation, panics and differing inputs", func(t *testing.T) { - expectedSha := gitRevParse(t, remote, "main") for _, scenario := range []string{"shared", "leader cancelled", "leader panics", "different token"} { t.Run(scenario, func(t *testing.T) { serve("prefixed", 0) @@ -181,7 +187,7 @@ func TestFetchAction(t *testing.T) { input := newInput("v1") tree, sha, err := FetchAction(t.Context(), input) require.NoError(t, err) - assert.Equal(t, gitRevParse(t, remote, "v1"), sha) + assert.Equal(t, expectedSha, sha) assert.Equal(t, filepath.Join(input.Dir, sha), tree) assert.NoDirExists(t, filepath.Join(tree, ".git")) info, err := os.Stat(filepath.Join(tree, "run.sh")) @@ -197,9 +203,10 @@ func TestFetchAction(t *testing.T) { require.NoError(t, gitCmd("-C", remote, "commit", "--allow-empty", "-m", "moved")) require.NoError(t, gitCmd("-C", remote, "tag", "--force", "v1")) + expectedSha = gitRevParse(t, remote, "v1") moved, movedSha, err := FetchAction(t.Context(), input) require.NoError(t, err) - assert.Equal(t, gitRevParse(t, remote, "v1"), movedSha) + assert.Equal(t, expectedSha, movedSha) assert.NotEqual(t, tree, moved) assert.FileExists(t, filepath.Join(tree, "action.yml")) @@ -226,7 +233,7 @@ func TestFetchAction(t *testing.T) { t.Run("downloads a pinned commit once and rejects a short one", func(t *testing.T) { serve("prefixed", 0) - sha := gitRevParse(t, remote, "main") + sha := expectedSha input := newInput(sha) for range 2 { tree, got, err := FetchAction(t.Context(), input) @@ -308,7 +315,7 @@ func TestFetchAction(t *testing.T) { input := newInput("main") _, sha, err := FetchAction(t.Context(), input) require.NoError(t, err) - assert.Equal(t, gitRevParse(t, remote, "main"), sha) + assert.Equal(t, expectedSha, sha) assert.NoDirExists(t, filepath.Join(input.Dir, "git")) _, headers := served() require.Len(t, headers, 4) @@ -321,7 +328,7 @@ func TestFetchAction(t *testing.T) { serve(mode, 0) tree, sha, err := FetchAction(t.Context(), newInput("main")) require.NoError(t, err, mode) - assert.Equal(t, gitRevParse(t, remote, "main"), sha, mode) + assert.Equal(t, expectedSha, sha, mode) assert.FileExists(t, filepath.Join(tree, "action.yml"), mode) assert.NoDirExists(t, filepath.Join(tree, ".git"), mode) _, headers := served() @@ -336,7 +343,7 @@ func TestFetchAction(t *testing.T) { input.URL = "https://example.invalid/org/action" _, sha, err := FetchAction(t.Context(), input) require.NoError(t, err) - assert.Equal(t, gitRevParse(t, remote, "main"), sha) + assert.Equal(t, expectedSha, sha) _, headers := served() assert.Len(t, headers, 1) assert.NoDirExists(t, filepath.Join(input.Dir, "git")) diff --git a/act/common/git/git.go b/act/common/git/git.go index 66994c01..889f86c0 100644 --- a/act/common/git/git.go +++ b/act/common/git/git.go @@ -10,6 +10,7 @@ import ( "encoding/base64" "errors" "fmt" + "io" "net/http" "net/url" "os" @@ -320,9 +321,18 @@ func cachedRef(ref string) string { func resolveRef(ctx context.Context, dir, ref string) (string, error) { if !isFullObjectID(ref) { - for _, name := range []string{cachedRef(ref), "refs/tags/" + ref, "refs/remotes/origin/" + ref} { - if sha, err := gitOutput(ctx, dir, "rev-parse", "--verify", name+"^{commit}"); err == nil { - return sha, nil + revisions := []string{cachedRef(ref) + "^{commit}", "refs/tags/" + ref + "^{commit}", "refs/remotes/origin/" + ref + "^{commit}"} + if strings.ContainsAny(ref, "\x00\n") { + for _, revision := range revisions { + if sha, err := gitOutput(ctx, dir, "rev-parse", "--verify", revision); err == nil { + return sha, nil + } + } + } else if output, err := runGitWithStdin(ctx, dir, nil, strings.NewReader(strings.Join(revisions, "\n")+"\n"), "cat-file", "--batch-check=%(objectname)"); err == nil { + for sha := range strings.SplitSeq(string(output), "\n") { + if isFullObjectID(sha) { + return sha, nil + } } } } @@ -427,6 +437,10 @@ func ListFiles(ctx context.Context, dir string) ([]string, error) { } func runGit(ctx context.Context, dir string, input *NewGitCloneExecutorInput, args ...string) ([]byte, error) { + return runGitWithStdin(ctx, dir, input, nil, args...) +} + +func runGitWithStdin(ctx context.Context, dir string, input *NewGitCloneExecutorInput, stdin io.Reader, args ...string) ([]byte, error) { config := []string{"core.autocrlf=false", "core.eol=lf"} // keep action files LF if input != nil { config = append(config, RemoteConfig(input.Token, input.InsecureSkipTLS)...) @@ -438,6 +452,7 @@ func runGit(ctx context.Context, dir string, input *NewGitCloneExecutorInput, ar args = append([]string{"-C", dir}, args...) } cmd := exec.CommandContext(ctx, "git", args...) + cmd.Stdin = stdin cmd.Env = slices.DeleteFunc(os.Environ(), func(entry string) bool { name, _, _ := strings.Cut(entry, "=") return slices.Contains(LocalEnvVars, name) diff --git a/act/common/git/git_test.go b/act/common/git/git_test.go index b3d776ef..ca852edc 100644 --- a/act/common/git/git_test.go +++ b/act/common/git/git_test.go @@ -29,6 +29,31 @@ import ( "github.com/stretchr/testify/require" ) +var gitFixtureDir string + +func TestMain(m *testing.M) { + var err error + gitFixtureDir, err = os.MkdirTemp("", "runner-git-fixture-") + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + code := 1 + if err = gitCmd("init", "--quiet", "--template=", "--bare", "--initial-branch=main", filepath.Join(gitFixtureDir, "remote")); err == nil { + err = gitCmd("init", "--quiet", "--template=", "--initial-branch=main", filepath.Join(gitFixtureDir, "work")) + } + if err == nil { + code = m.Run() + } else { + fmt.Fprintln(os.Stderr, err) + } + if err := os.RemoveAll(gitFixtureDir); err != nil { + fmt.Fprintln(os.Stderr, err) + code = 1 + } + os.Exit(code) +} + func TestFindGitSlug(t *testing.T) { for _, testcase := range []struct { remoteURL string @@ -141,9 +166,8 @@ func TestGitFindRef(t *testing.T) { } { t.Run(name, func(t *testing.T) { dir := filepath.Join(basedir, name) - require.NoError(t, os.MkdirAll(dir, 0o755)) - require.NoError(t, gitCmd("-C", dir, "init", "--initial-branch=master")) - require.NoError(t, os.RemoveAll(filepath.Join(dir, ".git", "hooks"))) + require.NoError(t, os.CopyFS(dir, os.DirFS(filepath.Join(gitFixtureDir, "work")))) + require.NoError(t, os.WriteFile(filepath.Join(dir, ".git", "HEAD"), []byte("ref: refs/heads/master\n"), 0o644)) for _, command := range tt.commands { require.NoError(t, gitCmd(append([]string{"-C", dir}, command...)...)) } @@ -180,6 +204,7 @@ func TestGitCloneExecutor(t *testing.T) { t.Setenv("GIT_CONFIG_GLOBAL", hostConfig) t.Setenv("GIT_DIR", t.TempDir()) + var clones sync.WaitGroup for name, tt := range map[string]struct { Err error Ref string @@ -207,31 +232,34 @@ func TestGitCloneExecutor(t *testing.T) { Ref: fullSha[:4], }, } { - t.Run(name, func(t *testing.T) { - dir := t.TempDir() - clone := NewGitCloneExecutor(NewGitCloneExecutorInput{ - URL: "https://example.invalid/action", - Ref: tt.Ref, - Dir: dir, - }) + clones.Go(func() { + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + clone := NewGitCloneExecutor(NewGitCloneExecutorInput{ + URL: "https://example.invalid/action", + Ref: tt.Ref, + Dir: dir, + }) - err := clone(t.Context()) - if tt.Err != nil { - require.Error(t, err) - assert.Equal(t, tt.Err, err) - assert.Equal(t, tt.Err, clone(t.Context()), "a retry on the cache keeps the error") - return - } - require.NoError(t, err) - marker := filepath.Join(dir, "marker") - require.NoError(t, os.WriteFile(marker, nil, 0o644)) - require.NoError(t, clone(t.Context())) - assert.FileExists(t, marker, "an insteadOf rewrite must not evict the cache") - script, err := os.ReadFile(filepath.Join(dir, "action.sh")) - require.NoError(t, err) - assert.Equal(t, "#!/bin/sh\necho hi\n", string(script), "host core.autocrlf must not convert action files") + err := clone(t.Context()) + if tt.Err != nil { + require.Error(t, err) + assert.Equal(t, tt.Err, err) + assert.Equal(t, tt.Err, clone(t.Context()), "a retry on the cache keeps the error") + return + } + require.NoError(t, err) + marker := filepath.Join(dir, "marker") + require.NoError(t, os.WriteFile(marker, nil, 0o644)) + require.NoError(t, clone(t.Context())) + assert.FileExists(t, marker, "an insteadOf rewrite must not evict the cache") + script, err := os.ReadFile(filepath.Join(dir, "action.sh")) + require.NoError(t, err) + assert.Equal(t, "#!/bin/sh\necho hi\n", string(script), "host core.autocrlf must not convert action files") + }) }) } + clones.Wait() } func TestGitCloneExecutorReclonesWhenOriginURLChanges(t *testing.T) { @@ -476,9 +504,12 @@ func TestGitCloneExecutorPinnedHashIgnoresShadowingRef(t *testing.T) { func newGitRemote(t *testing.T, messages ...string) (string, string) { t.Helper() remoteDir, workDir := t.TempDir(), t.TempDir() - require.NoError(t, gitCmd("init", "--bare", "--initial-branch=main", remoteDir)) - require.NoError(t, gitCmd("clone", remoteDir, workDir)) - require.NoError(t, gitCmd("-C", workDir, "checkout", "-b", "main")) + require.NoError(t, os.CopyFS(remoteDir, os.DirFS(filepath.Join(gitFixtureDir, "remote")))) + require.NoError(t, os.CopyFS(workDir, os.DirFS(filepath.Join(gitFixtureDir, "work")))) + config, err := os.ReadFile(filepath.Join(workDir, ".git", "config")) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(workDir, ".git", "config"), + fmt.Appendf(config, "\n[remote \"origin\"]\n\turl = %s\n\tfetch = +refs/heads/*:refs/remotes/origin/*\n", strconv.Quote(remoteDir)), 0o644)) for _, message := range messages { require.NoError(t, gitCmd("-C", workDir, "commit", "--allow-empty", "-m", message)) } diff --git a/act/container/docker_images_test.go b/act/container/docker_images_test.go index 8942fd46..b219f931 100644 --- a/act/container/docker_images_test.go +++ b/act/container/docker_images_test.go @@ -5,14 +5,17 @@ package container import ( + "bytes" + "context" "fmt" - "os" - "os/exec" "strings" "testing" + "time" + "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common/logging" + mobyclient "github.com/moby/moby/client" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -21,48 +24,47 @@ func init() { logging.SetLevel(logging.LevelDebug) } -// buildScratchImage builds a tiny empty image for the given platform locally (FROM scratch, no -// network or emulation since there is nothing to run) and returns its tag, removing it after -// the test. -func buildScratchImage(t *testing.T, platform string) string { +func loadScratchImage(t *testing.T, platform string) string { t.Helper() - tag := fmt.Sprintf("act-test-exists-%s:latest", strings.TrimPrefix(platform, "linux/")) - cmd := exec.Command("docker", "build", "--platform", platform, "-t", tag, "-") - cmd.Stdin = strings.NewReader("FROM scratch\nLABEL act-test=1\n") - // Force BuildKit: it records the requested architecture in the image config for a - // FROM-scratch build, whereas the classic builder ignores --platform and tags it with the - // host arch, which would break the per-platform existence assertions below. - cmd.Env = append(os.Environ(), "DOCKER_BUILDKIT=1") - out, err := cmd.CombinedOutput() - require.NoError(t, err, string(out)) - t.Cleanup(func() { _ = exec.Command("docker", "rmi", "-f", tag).Run() }) + tag := fmt.Sprintf("act-test-exists-%s-%d:latest", strings.TrimPrefix(platform, "linux/"), time.Now().UnixNano()) + var archive bytes.Buffer + require.NoError(t, writeFilesTar(t.Context(), &archive, 0, 0, + &FileEntry{Name: "config.json", Mode: 0o600, Body: fmt.Sprintf(`{"architecture":%q,"os":"linux","rootfs":{"type":"layers","diff_ids":[]}}`, strings.TrimPrefix(platform, "linux/"))}, + &FileEntry{Name: "manifest.json", Mode: 0o600, Body: fmt.Sprintf(`[{"Config":"config.json","RepoTags":[%q],"Layers":[]}]`, tag)}, + )) + cli, err := GetDockerClient(t.Context()) + require.NoError(t, err) + t.Cleanup(func() { _ = cli.Close() }) + response, err := cli.ImageLoad(t.Context(), &archive) + require.NoError(t, err) + defer response.Close() + require.NoError(t, logDockerResponse(common.Logger(t.Context()), response, false)) + t.Cleanup(func() { + _, _ = cli.ImageRemove(context.WithoutCancel(t.Context()), tag, mobyclient.ImageRemoveOptions{Force: true}) + }) return tag } func TestImageExistsLocally(t *testing.T) { + t.Parallel() requireDocker(t) ctx := t.Context() - // a non-existent image is reported absent missing, err := ImageExistsLocally(ctx, "library/alpine:this-random-tag-will-never-exist", "linux/amd64") assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.False(t, missing) - // Build tiny images for two architectures locally so per-platform existence can be checked - // offline (formerly pulled node:24-bookworm-slim for amd64 and arm64 over the network). - amd64Ref := buildScratchImage(t, "linux/amd64") - arm64Ref := buildScratchImage(t, "linux/arm64") + amd64Ref := loadScratchImage(t, "linux/amd64") + arm64Ref := loadScratchImage(t, "linux/arm64") amd64Exists, err := ImageExistsLocally(ctx, amd64Ref, "linux/amd64") assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.True(t, amd64Exists) - // a non-host architecture image is detected for its own architecture arm64Exists, err := ImageExistsLocally(ctx, arm64Ref, "linux/arm64") assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.True(t, arm64Exists) - // a present image is reported absent for a different platform wrongPlatform, err := ImageExistsLocally(ctx, amd64Ref, "linux/arm64") assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.False(t, wrongPlatform) diff --git a/act/container/docker_network.go b/act/container/docker_network.go index ee072529..d4cd6aa3 100644 --- a/act/container/docker_network.go +++ b/act/container/docker_network.go @@ -85,14 +85,14 @@ networks: } func NewDockerNetworkCreateExecutor(name string, opts NewDockerNetworkCreateExecutorInput) common.Executor { - return func(ctx context.Context) error { + return common.Executor(func(ctx context.Context) error { cli, err := GetDockerClient(ctx) if err != nil { return err } defer cli.Close() return createDockerNetwork(ctx, cli, name, opts) - } + }).IfNot(common.Dryrun) } func createDockerNetwork(ctx context.Context, cli client.APIClient, name string, opts NewDockerNetworkCreateExecutorInput) error { @@ -157,7 +157,7 @@ func isAddressPoolExhausted(err error) bool { } func NewDockerNetworkConnectExecutor(networkName, container string) common.Executor { - return func(ctx context.Context) error { + return common.Executor(func(ctx context.Context) error { cli, err := GetDockerClient(ctx) if err != nil { return err @@ -167,7 +167,7 @@ func NewDockerNetworkConnectExecutor(networkName, container string) common.Execu common.Logger(ctx).Warnf("cannot connect %s to network %s, so jobs on it cannot reach the cache: %v", container, networkName, err) } return nil - } + }).IfNot(common.Dryrun) } func connectNetwork(ctx context.Context, cli client.APIClient, networkName, container string) error { @@ -182,14 +182,14 @@ func connectNetwork(ctx context.Context, cli client.APIClient, networkName, cont } func NewDockerNetworkDisconnectExecutor(networkName, container string) common.Executor { - return func(ctx context.Context) error { + return common.Executor(func(ctx context.Context) error { cli, err := GetDockerClient(ctx) if err != nil { return err } defer cli.Close() return disconnectNetwork(ctx, cli, networkName, container) - } + }).IfNot(common.Dryrun) } func disconnectNetwork(ctx context.Context, cli client.APIClient, networkName, container string) error { @@ -201,14 +201,14 @@ func disconnectNetwork(ctx context.Context, cli client.APIClient, networkName, c } func NewDockerNetworkRemoveExecutor(name string) common.Executor { - return func(ctx context.Context) error { + return common.Executor(func(ctx context.Context) error { cli, err := GetDockerClient(ctx) if err != nil { return err } defer cli.Close() return removeNetwork(ctx, cli, name) - } + }).IfNot(common.Dryrun) } func removeNetwork(ctx context.Context, cli client.APIClient, name string) error { diff --git a/act/container/docker_proxy.go b/act/container/docker_proxy.go index 7e2cdc25..a105abd9 100644 --- a/act/container/docker_proxy.go +++ b/act/container/docker_proxy.go @@ -48,13 +48,20 @@ var ( rawStreamPath = regexp.MustCompile(`^(/v[0-9.]+)?/(containers/[^/]+/attach|exec/[^/]+/start)$`) ) -var dockerProxyDirectories sync.Map +var dockerProxyDirectories, dockerProxyProbes sync.Map type dockerProxyDirectory struct { dir, daemonDir string socket os.FileInfo } +type dockerProxyProbe struct { + directory dockerProxyDirectory + done chan struct{} + seen bool + err error +} + func NewDockerProxy(ctx context.Context, job string) *DockerProxy { if host := os.Getenv("DOCKER_HOST"); runtime.GOOS != "linux" || host != "" && !strings.HasPrefix(host, "unix://") { return nil @@ -79,16 +86,8 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy { cached, _ := dockerProxyDirectories.Load(key) directory, seen := cached.(dockerProxyDirectory) seen = seen && os.SameFile(directory.socket, socket) // daemon replacement invalidates cached paths - dir, daemonDir := directory.dir, directory.daemonDir if !seen { - dir, daemonDir = runnerContainerWorkdir(probeCtx, cli) - seen = daemonDir != "" - if !seen { - if dir, err = filepath.Abs(os.TempDir()); err == nil { - daemonDir = dir - seen, err = daemonSeesDir(probeCtx, cli, dir, daemonDir) - } - } + directory, seen, err = probeDockerProxyDirectory(probeCtx, cli, key, socket) } if err != nil { common.Logger(ctx).Infof("docker proxy probe failed, jobs get the daemon socket directly: %v", err) @@ -101,17 +100,51 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy { if ctx.Err() != nil { return nil } - proxy, err := StartDockerProxy(daemonSocket, dir, job) + proxy, err := StartDockerProxy(daemonSocket, directory.dir, job) if err != nil { dockerProxyDirectories.Delete(key) common.Logger(ctx).Warnf("docker proxy not started, the job gets the daemon socket directly: %v", err) return nil } - dockerProxyDirectories.Store(key, dockerProxyDirectory{dir, daemonDir, socket}) - proxy.Socket = daemonDir + strings.TrimPrefix(proxy.Socket, dir) + dockerProxyDirectories.Store(key, directory) + proxy.Socket = directory.daemonDir + strings.TrimPrefix(proxy.Socket, directory.dir) return proxy } +func probeDockerProxyDirectory(ctx context.Context, cli client.APIClient, key [3]string, socket os.FileInfo) (dockerProxyDirectory, bool, error) { + for { + probe := &dockerProxyProbe{directory: dockerProxyDirectory{socket: socket}, done: make(chan struct{})} + actual, loaded := dockerProxyProbes.LoadOrStore(key, probe) + if loaded { + probe, _ = actual.(*dockerProxyProbe) + if !os.SameFile(probe.directory.socket, socket) { + dockerProxyProbes.CompareAndDelete(key, probe) + continue + } + select { + case <-ctx.Done(): + return dockerProxyDirectory{}, false, ctx.Err() + case <-probe.done: + } + if ctx.Err() == nil && (errors.Is(probe.err, context.Canceled) || errors.Is(probe.err, context.DeadlineExceeded)) { + continue + } + } else { + probe.directory.dir, probe.directory.daemonDir = runnerContainerWorkdir(ctx, cli) + probe.seen = probe.directory.daemonDir != "" + if !probe.seen { + if probe.directory.dir, probe.err = filepath.Abs(key[2]); probe.err == nil { + probe.directory.daemonDir = probe.directory.dir + probe.seen, probe.err = daemonSeesDir(ctx, cli, probe.directory.dir, probe.directory.daemonDir) + } + } + dockerProxyProbes.CompareAndDelete(key, probe) + close(probe.done) + } + return probe.directory, probe.seen, probe.err + } +} + // runnerContainerWorkdir looks the runner's container up by hostname to find the daemon's path to its working directory. func runnerContainerWorkdir(ctx context.Context, cli client.APIClient) (workdir, daemonDir string) { workdir, err := os.Getwd() @@ -162,12 +195,18 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str if len(images.Items) == 0 { return false, errors.New("no image available for the docker proxy probe") } - created, err := cli.ContainerCreate(ctx, client.ContainerCreateOptions{ - Config: &container.Config{Image: images.Items[0].ID, Cmd: []string{"true"}}, - HostConfig: &container.HostConfig{Mounts: []mount.Mount{ - {Type: mount.TypeBind, Source: daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}, // not the marker itself, podman creates a missing bind source where docker rejects it - }}, - }) + var created client.ContainerCreateResult + for _, image := range images.Items { + created, err = cli.ContainerCreate(ctx, client.ContainerCreateOptions{ + Config: &container.Config{Image: image.ID, Cmd: []string{"true"}}, + HostConfig: &container.HostConfig{Mounts: []mount.Mount{ + {Type: mount.TypeBind, Source: daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}, // not the marker itself, podman creates a missing bind source where docker rejects it + }}, + }) + if !cerrdefs.IsNotFound(err) { + break + } + } if cerrdefs.IsInvalidArgument(err) { return false, nil } diff --git a/act/container/docker_proxy_test.go b/act/container/docker_proxy_test.go index 209a6509..0cce800e 100644 --- a/act/container/docker_proxy_test.go +++ b/act/container/docker_proxy_test.go @@ -22,6 +22,7 @@ import ( "strings" "sync/atomic" "testing" + "testing/synctest" "time" cerrdefs "github.com/containerd/errdefs" @@ -486,7 +487,7 @@ func (c *probeClient) ContainerInspect(context.Context, string, mobyclient.Conta } func (c *probeClient) ImageList(context.Context, mobyclient.ImageListOptions) (mobyclient.ImageListResult, error) { - return mobyclient.ImageListResult{Items: []image.Summary{{ID: "probe-image"}}}, nil + return mobyclient.ImageListResult{Items: []image.Summary{{ID: "removed-image"}, {ID: "probe-image"}}}, nil } func (c *probeClient) ContainerCreate(_ context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { @@ -501,6 +502,139 @@ func (c *probeClient) ContainerStatPath(_ context.Context, _ string, opts mobycl return mobyclient.ContainerStatPathResult{}, c.stat(opts.Path) } +func TestDockerProxyDirectoryProbes(t *testing.T) { + type result struct { + seen bool + err error + } + for name, createErr := range map[string]error{"unavailable directory": nil, "daemon error": errors.New("probe failed")} { + t.Run(name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + dir := t.TempDir() + socket, err := os.Stat(t.TempDir()) + require.NoError(t, err) + key := [3]string{"daemon", dir, dir} + release := make(chan struct{}) + var calls, removals atomic.Int32 + visible := false + cli := &probeClient{ + create: func(_opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + calls.Add(1) + <-release + return mobyclient.ContainerCreateResult{ID: "probe"}, createErr + }, + stat: func(_path string) error { + if visible { + return nil + } else { + return cerrdefs.ErrNotFound + } + }, + remove: func(_ctx context.Context, _id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) { + removals.Add(1) + return mobyclient.ContainerRemoveResult{}, nil + }, + } + results := make(chan result, 8) + for range cap(results) { + go func() { + _, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + results <- result{seen, err} + }() + } + synctest.Wait() + assert.EqualValues(t, 1, calls.Load()) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, seen, err := probeDockerProxyDirectory(ctx, cli, key, socket) + require.ErrorIs(t, err, context.Canceled) + assert.False(t, seen) + replacement, err := os.Stat(t.TempDir()) + require.NoError(t, err) + replaced := make(chan result, 1) + go func() { + _, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, replacement) + replaced <- result{seen, err} + }() + synctest.Wait() + assert.EqualValues(t, 2, calls.Load()) + close(release) + for range cap(results) { + result := <-results + require.ErrorIs(t, result.err, createErr) + assert.False(t, result.seen) + } + replacementResult := <-replaced + require.ErrorIs(t, replacementResult.err, createErr) + assert.False(t, replacementResult.seen) + assert.EqualValues(t, 2, calls.Load()) + if createErr == nil { + assert.EqualValues(t, 2, removals.Load()) + } else { + assert.Zero(t, removals.Load()) + } + createErr, visible = nil, true + directory, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + require.NoError(t, err) + assert.True(t, seen) + assert.Equal(t, dir, directory.dir) + assert.Equal(t, dir, directory.daemonDir) + assert.EqualValues(t, 3, calls.Load()) + entries, err := os.ReadDir(dir) + require.NoError(t, err) + assert.Empty(t, entries) + }) + }) + } + t.Run("cancelled probe does not cancel another caller", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + dir := t.TempDir() + socket, err := os.Stat(t.TempDir()) + require.NoError(t, err) + key := [3]string{"daemon", dir, dir} + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + release := make(chan struct{}) + calls := 0 + cli := &probeClient{ + create: func(_opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + calls++ + if calls == 1 { + <-release + return mobyclient.ContainerCreateResult{}, ctx.Err() + } else { + return mobyclient.ContainerCreateResult{ID: "probe"}, nil + } + }, + stat: func(_path string) error { return nil }, + remove: func(_ctx context.Context, _id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) { + return mobyclient.ContainerRemoveResult{}, nil + }, + } + leader, follower := make(chan result, 1), make(chan result, 1) + go func() { + _, seen, err := probeDockerProxyDirectory(ctx, cli, key, socket) + leader <- result{seen, err} + }() + synctest.Wait() + go func() { + _, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + follower <- result{seen, err} + }() + synctest.Wait() + assert.Equal(t, 1, calls) + cancel() + close(release) + leaderResult, followerResult := <-leader, <-follower + require.ErrorIs(t, leaderResult.err, context.Canceled) + assert.False(t, leaderResult.seen) + require.NoError(t, followerResult.err) + assert.True(t, followerResult.seen) + assert.Equal(t, 2, calls) + }) + }) +} + func TestDaemonSeesDir(t *testing.T) { dir := t.TempDir() markers := make(map[string]bool) @@ -520,9 +654,14 @@ func TestDaemonSeesDir(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() removed := false + var attemptedImages []string cli := &probeClient{ create: func(opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { assert.Equal(t, []mount.Mount{{Type: mount.TypeBind, Source: testCase.daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}}, opts.HostConfig.Mounts) + attemptedImages = append(attemptedImages, opts.Config.Image) + if opts.Config.Image == "removed-image" { + return mobyclient.ContainerCreateResult{}, cerrdefs.ErrNotFound + } cancel() if _, err := os.Stat(testCase.daemonDir); err != nil { return mobyclient.ContainerCreateResult{}, cerrdefs.ErrInvalidArgument @@ -550,6 +689,7 @@ func TestDaemonSeesDir(t *testing.T) { } seen, err := daemonSeesDir(ctx, cli, dir, testCase.daemonDir) require.ErrorIs(t, err, testCase.removeErr) + assert.Equal(t, []string{"removed-image", "probe-image"}, attemptedImages) assert.Equal(t, testCase.seen, seen) assert.Equal(t, testCase.removed, removed) entries, err := os.ReadDir(dir) diff --git a/act/container/docker_ssh_test.go b/act/container/docker_ssh_test.go index 510983ad..3bcd4fac 100644 --- a/act/container/docker_ssh_test.go +++ b/act/container/docker_ssh_test.go @@ -30,6 +30,7 @@ func TestSSHArgs(t *testing.T) { } func TestDialCommandStreamsStdioAndReportsExit(t *testing.T) { + t.Parallel() if _, err := exec.LookPath("sh"); err != nil { t.Skip("needs a POSIX shell") } diff --git a/act/container/host_environment.go b/act/container/host_environment.go index e35ede3f..8ed565f9 100644 --- a/act/container/host_environment.go +++ b/act/container/host_environment.go @@ -225,12 +225,16 @@ func setupPty(cmd *exec.Cmd, cmdline string) (*os.File, *os.File, error) { return ppty, tty, nil } -func writeKeepAlive(ppty io.Writer) { - c := 1 - var err error - for c == 1 && err == nil { - c, err = ppty.Write([]byte{4}) - <-time.After(time.Second) +func writeKeepAlive(ctx context.Context, ppty io.Writer) { + for ctx.Err() == nil { + if count, err := ppty.Write([]byte{4}); count != 1 || err != nil { + return + } + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } } } @@ -302,7 +306,7 @@ func (e *HostEnvironment) exec(ctx context.Context, command []string, cmdline st defer finishLog() _, _ = io.Copy(writer, ppty) }() - go writeKeepAlive(ppty) + go writeKeepAlive(logctx, ppty) } if err := cmd.Start(); err != nil { return err diff --git a/act/container/host_environment_test.go b/act/container/host_environment_test.go index be525b32..177b4d4b 100644 --- a/act/container/host_environment_test.go +++ b/act/container/host_environment_test.go @@ -14,6 +14,7 @@ import ( "runtime" "strings" "testing" + "testing/synctest" "time" "gitea.com/gitea/runner/act/common" @@ -79,42 +80,48 @@ func TestGetContainerArchive(t *testing.T) { } func TestHostEnvironmentExecExitCode(t *testing.T) { + t.Parallel() if runtime.GOOS == "windows" { t.Skip("uses POSIX shell") } for name, allocatePTY := range map[string]bool{"without PTY": false, "with PTY": true} { t.Run(name, func(t *testing.T) { - reader, writer := io.Pipe() - defer reader.Close() - e := &HostEnvironment{Path: t.TempDir(), StdOut: writer, AllocatePTY: allocatePTY} - t.Cleanup(func() { assert.NoError(t, e.Remove()(t.Context())) }) + t.Parallel() + synctest.Test(t, func(t *testing.T) { + reader, writer := io.Pipe() + defer reader.Close() + e := &HostEnvironment{Path: t.TempDir(), StdOut: writer, AllocatePTY: allocatePTY} + t.Cleanup(func() { assert.NoError(t, e.Remove()(t.Context())) }) - result := make(chan error, 1) - go func() { - defer writer.Close() - result <- e.Exec([]string{"sh", "-c", "printf 'failed\\n'; exit 3"}, map[string]string{"PATH": os.Getenv("PATH")}, "", "")(t.Context()) - }() - first := make([]byte, 1) - _, err := io.ReadFull(reader, first) - require.NoError(t, err) - select { - case err := <-result: - t.Fatalf("Exec returned before its output was consumed: %v", err) - case <-time.After(100 * time.Millisecond): - } - remaining, err := io.ReadAll(reader) - require.NoError(t, err) - err = <-result - var exitErr ExitCodeError - require.ErrorAs(t, err, &exitErr) - assert.Equal(t, ExitCodeError(3), exitErr) - assert.Equal(t, "Process completed with exit code 3.", err.Error()) - assert.Equal(t, "failed\n", string(first)+string(remaining)) + result := make(chan error, 1) + go func() { + defer writer.Close() + result <- e.Exec([]string{"sh", "-c", "printf 'failed\\n'; exit 3"}, map[string]string{"PATH": os.Getenv("PATH")}, "", "")(t.Context()) + }() + first := make([]byte, 1) + _, err := io.ReadFull(reader, first) + require.NoError(t, err) + synctest.Wait() + select { + case err := <-result: + t.Fatalf("Exec returned before its output was consumed: %v", err) + default: + } + remaining, err := io.ReadAll(reader) + require.NoError(t, err) + err = <-result + var exitErr ExitCodeError + require.ErrorAs(t, err, &exitErr) + assert.Equal(t, ExitCodeError(3), exitErr) + assert.Equal(t, "Process completed with exit code 3.", err.Error()) + assert.Equal(t, "failed\n", string(first)+string(remaining)) + }) }) } } func TestHostEnvironmentAllocatePTY(t *testing.T) { + t.Parallel() if runtime.GOOS == "windows" { t.Skip("uses POSIX shell") } @@ -127,6 +134,7 @@ func TestHostEnvironmentAllocatePTY(t *testing.T) { {name: "on", allocPTY: true, expect: "TTY"}, } { t.Run(tc.name, func(t *testing.T) { + t.Parallel() buf := &bytes.Buffer{} e := &HostEnvironment{ Path: t.TempDir(), @@ -168,63 +176,67 @@ func TestHostEnvironmentRemoveWorkdir(t *testing.T) { } func TestRemoveAllWithContextDoesNotHangOnStuckDelete(t *testing.T) { - release := make(chan struct{}) - stubDone := make(chan struct{}) + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + stubDone := make(chan struct{}) - orig := removeAll - removeAll = func(string) error { - defer close(stubDone) - <-release - return nil - } - t.Cleanup(func() { - close(release) - <-stubDone - removeAll = orig + orig := removeAll + removeAll = func(string) error { + defer close(stubDone) + <-release + return nil + } + t.Cleanup(func() { + close(release) + <-stubDone + removeAll = orig + }) + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + + err := removeAllWithContext(ctx, t.TempDir()) + require.ErrorIs(t, err, context.DeadlineExceeded) }) - - ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) - defer cancel() - - err := removeAllWithContext(ctx, t.TempDir()) - require.ErrorIs(t, err, context.DeadlineExceeded) } func TestHostEnvironmentRemoveDoesNotHang(t *testing.T) { for _, phase := range []string{"CleanUp callback", "path removal"} { t.Run(phase, func(t *testing.T) { - origTimeout := hostCleanupTimeout - hostCleanupTimeout = 100 * time.Millisecond - t.Cleanup(func() { hostCleanupTimeout = origTimeout }) - release, stubDone := make(chan struct{}), make(chan struct{}) - block := func() { - defer close(stubDone) - <-release - } - e := &HostEnvironment{Path: filepath.Join(t.TempDir(), "misc", "hostexecutor"), StdOut: os.Stdout} - if phase == "CleanUp callback" { - e.CleanUp = block - } else { - origRemoveAll := removeAll - removeAll = func(string) error { - block() - return nil + synctest.Test(t, func(t *testing.T) { + origTimeout := hostCleanupTimeout + hostCleanupTimeout = 100 * time.Millisecond + t.Cleanup(func() { hostCleanupTimeout = origTimeout }) + release, stubDone := make(chan struct{}), make(chan struct{}) + block := func() { + defer close(stubDone) + <-release + } + e := &HostEnvironment{Path: filepath.Join(t.TempDir(), "misc", "hostexecutor"), StdOut: os.Stdout} + if phase == "CleanUp callback" { + e.CleanUp = block + } else { + origRemoveAll := removeAll + removeAll = func(string) error { + block() + return nil + } + t.Cleanup(func() { removeAll = origRemoveAll }) + } + t.Cleanup(func() { + close(release) + <-stubDone + }) + require.NoError(t, os.MkdirAll(e.Path, 0o700)) + done := make(chan error, 1) + go func() { done <- e.Remove()(common.WithLogger(t.Context(), logging.Default())) }() + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(10 * time.Second): + t.Fatalf("Remove() hung on a stuck %s", phase) } - t.Cleanup(func() { removeAll = origRemoveAll }) - } - t.Cleanup(func() { - close(release) - <-stubDone }) - require.NoError(t, os.MkdirAll(e.Path, 0o700)) - done := make(chan error, 1) - go func() { done <- e.Remove()(common.WithLogger(t.Context(), logging.Default())) }() - select { - case err := <-done: - require.NoError(t, err) - case <-time.After(10 * time.Second): - t.Fatalf("Remove() hung on a stuck %s", phase) - } }) } } diff --git a/act/container/kubernetes_test.go b/act/container/kubernetes_test.go index 33dca99b..4b48a67f 100644 --- a/act/container/kubernetes_test.go +++ b/act/container/kubernetes_test.go @@ -89,7 +89,7 @@ func execFailure(t *testing.T, code int, status string) http.HandlerFunc { } } -func fakeKubernetes(t *testing.T, podStatus string) (string, *fakeCluster) { +func fakeKubernetes(t *testing.T, podStatus string, env ...string) (string, *fakeCluster) { cluster := &fakeCluster{requests: map[string][]string{}} server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Header.Get("Authorization") != "Bearer test-token" { @@ -109,7 +109,7 @@ func fakeKubernetes(t *testing.T, podStatus string) (string, *fakeCluster) { if failure := cluster.nextExecFailure(); failure != nil { failure(w, r) } else { - serveLocalExec(t, w, r) + serveLocalExec(t, w, r, env) } case strings.HasSuffix(r.URL.Path, "/log"): _, _ = io.WriteString(w, "log of "+r.URL.Query().Get("container")) @@ -132,7 +132,7 @@ users: [{name: test, user: {tokenFile: token}}] return kubeconfig, cluster } -func serveLocalExec(t *testing.T, w http.ResponseWriter, r *http.Request) { +func serveLocalExec(t *testing.T, w http.ResponseWriter, r *http.Request, env []string) { conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{execProtocol}}) if err != nil { t.Error(err) @@ -154,7 +154,7 @@ func serveLocalExec(t *testing.T, w http.ResponseWriter, r *http.Request) { stdin.Write(frame[1:]) } cmd := exec.Command(command[0], command[1:]...) - cmd.Env = append(os.Environ(), "COPYFILE_DISABLE=1") + cmd.Env = append(append(os.Environ(), "COPYFILE_DISABLE=1"), env...) var stdout, stderr bytes.Buffer cmd.Stdin, cmd.Stdout, cmd.Stderr = &stdin, &stdout, &stderr status := `{"status":"Success"}` @@ -183,15 +183,17 @@ func TestKubernetesServiceHostnames(t *testing.T) { } func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { + t.Parallel() if runtime.GOOS == "windows" { t.Skip("uses POSIX shell") } ctx := t.Context() - kubeconfig, cluster := fakeKubernetes(t, `{"spec":{"containers":[{},{}]},"status":{"phase":"Running","containerStatuses":[ - {"name":"job","ready":true,"state":{"running":{}}}, - {"name":"my-db","ready":true,"state":{"running":{}}}]}}`) workdir, err := filepath.EvalSymlinks(t.TempDir()) require.NoError(t, err) + path := workdir + ":" + os.Getenv("PATH") + kubeconfig, cluster := fakeKubernetes(t, `{"spec":{"containers":[{},{}]},"status":{"phase":"Running","containerStatuses":[ + {"name":"job","ready":true,"state":{"running":{}}}, + {"name":"my-db","ready":true,"state":{"running":{}}}]}}`, "PATH="+path) options := KubernetesOptions{Kubeconfig: kubeconfig, RunnerUUID: "runner-uuid", PodTemplates: []map[string]any{ {"spec": map[string]any{ "automountServiceAccountToken": true, @@ -252,7 +254,6 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { assert.Contains(t, cluster.requests["POST /api/v1/namespaces/ci/pods"][0], `"args":["postgres","-D","$$(PGDATA)"]`) require.NoError(t, os.WriteFile(filepath.Join(workdir, "env"), []byte("#!/bin/sh\necho \"$@\" > \"$0-args\"\nexec /usr/bin/env \"$@\"\n"), 0o755)) - t.Setenv("PATH", workdir+":"+os.Getenv("PATH")) cluster.failExec( execFailure(t, http.StatusInternalServerError, `{"message":"error dialing backend: remote error: tls: internal error"}`), execFailure(t, 0, `{"status":"Failure","reason":"InternalError","message":"Internal error occurred: error sending request: remote error: tls: internal error"}`), @@ -300,7 +301,7 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { pathEnv := map[string]string{} require.NoError(t, job.UpdateFromImageEnv(&pathEnv)(ctx)) - assert.Equal(t, os.Getenv("PATH"), pathEnv["PATH"]) + assert.Equal(t, path, pathEnv["PATH"]) info, err := service.Inspect(ctx) require.NoError(t, err) diff --git a/act/container/parse_env_file.go b/act/container/parse_env_file.go index d7c4cf6b..ee02fba9 100644 --- a/act/container/parse_env_file.go +++ b/act/container/parse_env_file.go @@ -7,6 +7,7 @@ package container import ( "archive/tar" "bufio" + "bytes" "context" "fmt" "io" @@ -39,12 +40,19 @@ func parseEnvFile(e Container, srcPath string, env *map[string]string, windows b s := bufio.NewScanner(decoded) // Default 64 KiB max token size is too small for realistic env-file lines; allow up to 16 MiB. s.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) + scanned := 0 s.Split(func(data []byte, atEOF bool) (int, []byte, error) { - advance, token, err := bufio.ScanLines(data, atEOF) - if token != nil { - return advance, data[:advance], err + if index := bytes.IndexByte(data[scanned:], '\n'); index >= 0 { + advance := scanned + index + 1 + scanned = 0 + return advance, data[:advance], nil } - return advance, token, err + scanned = len(data) + if atEOF && len(data) > 0 { + scanned = 0 + return len(data), data, nil + } + return 0, nil, nil }) for s.Scan() { line := trimEnvLineEnding(s.Text(), windows) diff --git a/act/container/parse_env_file_test.go b/act/container/parse_env_file_test.go index 700b5e99..e28974fa 100644 --- a/act/container/parse_env_file_test.go +++ b/act/container/parse_env_file_test.go @@ -22,6 +22,7 @@ func newTestHostEnv(t *testing.T) (*HostEnvironment, string) { } func TestParseEnvFile(t *testing.T) { + t.Parallel() largeValue := strings.Repeat("x", 2*1024*1024) encodeUTF16 := func(endianness unicode.Endianness) string { content, err := unicode.UTF16(endianness, unicode.UseBOM).NewEncoder().Bytes([]byte("FOO=bar\r\nMULTI<config_getCacheServiceVersion();"}, {"esbuild", gateEsbuild, "var vu={};const getCacheServiceVersion=()=>XK(),getCacheServiceURL=()=>YK();"}, } { t.Run(tc.name, func(t *testing.T) { - // Unpatched, a Gitea host is taken for GHES: v1, whose branch already reads the - // cache URL. The patch has to move the version without moving that. - assert.Equal(t, "v1 http://cache:8088/", eval(t, tc.bundle, tc.prelude, "http://cache:8088/")) - patched, ok := patchedBundle([]byte(tc.bundle)) require.True(t, ok) - assert.Equal(t, "v2 http://cache:8088/", eval(t, string(patched), tc.prelude, "http://cache:8088/")) - assert.Equal(t, "v2 https://gitea.example", eval(t, string(patched), tc.prelude, ""), + evaluate := func(bundle string) string { + return "(() => {" + tc.prelude + bundle + "\nconsole.log(getCacheServiceVersion()+' '+getCacheServiceURL());})();\n" + } + cmd := exec.CommandContext(t.Context(), "node", "-e", evaluate(tc.bundle)+evaluate(string(patched))+ + "process.env.ACTIONS_CACHE_URL='';\n"+evaluate(string(patched))) + cmd.Env = append(os.Environ(), + "ACTIONS_CACHE_SERVICE_V2=true", + "ACTIONS_CACHE_URL=http://cache:8088/", + "ACTIONS_RESULTS_URL=https://gitea.example", + "GITHUB_SERVER_URL=https://gitea.example", + ) + out, err := cmd.CombinedOutput() + require.NoError(t, err, "%s", out) + results := strings.Split(strings.TrimSuffix(string(out), "\n"), "\n") + require.Len(t, results, 3) + assert.Equal(t, "v1 http://cache:8088/", results[0]) + assert.Equal(t, "v2 http://cache:8088/", results[1]) + assert.Equal(t, "v2 https://gitea.example", results[2], "with no cache server the results URL is still the fallback") }) } @@ -177,6 +174,7 @@ func TestPatchedBundleBehavesInNode(t *testing.T) { // the gate inside a double-quoted literal. Rewriting the call rather than emptying its argument // would end that string early and leave the bundle unparseable. func TestPatchedBundleSurvivesInsideAStringLiteral(t *testing.T) { + t.Parallel() requireHostTools(t, "node") escaped := strings.ReplaceAll(gateTSC, `"`, `\"`) diff --git a/act/runner/reusable_workflow_test.go b/act/runner/reusable_workflow_test.go index ee12d543..41489400 100644 --- a/act/runner/reusable_workflow_test.go +++ b/act/runner/reusable_workflow_test.go @@ -27,6 +27,7 @@ import ( // ref (branch/tag) must reflect the new tip on every invocation, not stay pinned // to the cache populated on the first run. func TestReusableWorkflowCachedBranchRefRefreshes(t *testing.T) { + t.Parallel() if _, err := exec.LookPath("git"); err != nil { t.Skip("git not available in PATH") } @@ -52,7 +53,7 @@ func TestReusableWorkflowCachedBranchRefRefreshes(t *testing.T) { gitMust(t, workDir, "push", "-u", "origin", "master") rc := &RunContext{ - Config: &Config{}, + Config: &Config{Env: testCheckoutEnv()}, Run: &model.Run{ JobID: "j1", Workflow: &model.Workflow{ @@ -149,6 +150,7 @@ func TestCloneRemoteReusableWorkflowSendsInstanceTokenAndClientCert(t *testing.T require.NoError(t, os.WriteFile(keyFile, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}), 0o600)) rc := &RunContext{ Config: &Config{ + Env: testCheckoutEnv(), GitHubInstance: server.URL, Secrets: map[string]string{"GITEA_TOKEN": "token"}, InsecureSkipTLS: true, diff --git a/act/runner/run_context.go b/act/runner/run_context.go index a0883f1f..7b1cfcee 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -114,6 +114,7 @@ type RunContext struct { stepEnv map[string]string unsecureCommandErr error // refused ::set-env::/::add-path::, turned into a step failure unsecureCommandMu sync.Mutex + checkoutPins sync.Map // git lookups of contexts no run has pinned } // serviceContainer pairs a service container with the workflow id that keys job.services. @@ -1539,6 +1540,7 @@ func (rc *RunContext) getGithubContext(ctx context.Context) *model.GithubContext ghc.SetBaseAndHeadRef() repoPath := rc.Config.Workdir + ctx = ghcontext.WithPinnedCheckout(ctx, &rc.checkoutPins) ghcontext.SetRepositoryAndOwner(ctx, ghc, repoPath) if ghc.Ref == "" { ghcontext.SetRef(ctx, ghc, repoPath) diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go index 66232bc1..06ce5426 100644 --- a/act/runner/run_context_test.go +++ b/act/runner/run_context_test.go @@ -43,6 +43,7 @@ func TestRunContext_EvalBool(t *testing.T) { rc := &RunContext{ Config: &Config{ Workdir: ".", + Env: testCheckoutEnv(), }, Env: map[string]string{ "SOMETHING_TRUE": "true", @@ -200,7 +201,7 @@ func startJobContainerInputs(t *testing.T, workflowYAML string, cfg *Config) []* cfg.Workdir = "/tmp" cfg.ContainerNetworkMode = "host" - cfg.Env = map[string]string{} + cfg.Env = testCheckoutEnv() cfg.Secrets = map[string]string{} rc := &RunContext{ @@ -481,6 +482,7 @@ func TestRunContext_GetBindsAndMounts(t *testing.T) { require.NoError(t, err) rc := newRunContext(&Config{ + Env: testCheckoutEnv(), ValidVolumes: []string{"**"}, Secrets: map[string]string{ "MAME": "/host/mame/roms", @@ -777,7 +779,7 @@ func TestInterpolateOutputsIsPerMatrixComboKeepingNeedsOutputs(t *testing.T) { require.NoError(t, job.RawOutputs.Encode(job.Outputs)) needed := &model.Job{Outputs: map[string]string{"o": "from-gitea"}} run := &model.Run{JobID: "j", Workflow: &model.Workflow{Name: "w", Jobs: map[string]*model.Job{"j": job, "needed": needed}}} - r := &runnerImpl{config: &Config{}} + r := &runnerImpl{config: &Config{PresetGitHubContext: &model.GithubContext{}}} ctx := t.Context() _, err := r.newRunContext(ctx, &model.Run{JobID: "needed", Workflow: run.Workflow}, nil) @@ -804,7 +806,7 @@ func TestInterpolateOutputsFromExpression(t *testing.T) { job := &model.Job{RawOutputs: rawOutputs} run := &model.Run{JobID: "j", Workflow: &model.Workflow{Name: "w", Jobs: map[string]*model.Job{"j": job}}} - rc, err := (&runnerImpl{config: &Config{}}).newRunContext(t.Context(), run, nil) + rc, err := (&runnerImpl{config: &Config{PresetGitHubContext: &model.GithubContext{}}}).newRunContext(t.Context(), run, nil) require.NoError(t, err) require.NoError(t, rc.interpolateOutputs()(t.Context())) @@ -956,6 +958,7 @@ func TestGetGithubContextRef(t *testing.T) { Config: &Config{ EventName: data.event, Workdir: "", + Env: map[string]string{"GITHUB_REPOSITORY": "owner/repo", "SHA_REF": "1234567890"}, }, Run: &model.Run{ Workflow: &model.Workflow{ @@ -973,7 +976,7 @@ func TestGetGithubContextRef(t *testing.T) { func createIfTestRunContext(jobs map[string]*model.Job) *RunContext { rc := &RunContext{ - Config: &Config{Workdir: ".", PlatformPicker: func([]string) string { return "ubuntu-latest" }}, + Config: &Config{Workdir: ".", PresetGitHubContext: &model.GithubContext{}, PlatformPicker: func([]string) string { return "ubuntu-latest" }}, Run: &model.Run{ JobID: "job1", Workflow: &model.Workflow{ @@ -1210,7 +1213,7 @@ func TestRunContext_cleanupFailedStart(t *testing.T) { workflow, err := model.ReadWorkflow(strings.NewReader("jobs: {job: {services: {postgres: {image: postgres:latest}}}}")) require.NoError(t, err) rc := &RunContext{ - Config: &Config{ForcePull: true, ContainerNetworkMode: "host", Workdir: "/workspace"}, + Config: &Config{ForcePull: true, ContainerNetworkMode: "host", Workdir: "/workspace", Env: testCheckoutEnv()}, Run: &model.Run{JobID: "job", Workflow: workflow}, platformImage: "node:20", } @@ -1562,6 +1565,7 @@ func TestRunContextGetRunnerContext(t *testing.T) { } func TestRunContextUpdateExtraPath(t *testing.T) { + t.Parallel() longPath := strings.Repeat("x", 64*1024+1) for _, testcase := range []struct { name string diff --git a/act/runner/runner.go b/act/runner/runner.go index cfb4532b..de1b2e51 100644 --- a/act/runner/runner.go +++ b/act/runner/runner.go @@ -324,7 +324,7 @@ func (runner *runnerImpl) NewPlanExecutor(plan *model.Plan) common.Executor { executor := common.NewPipelineExecutor(stagePipeline...).Then(handleFailure(plan)) return func(ctx context.Context) error { - return executor(container.WithRunnerArch(ghcontext.WithPinnedCheckout(ctx))) + return executor(container.WithRunnerArch(ghcontext.WithPinnedCheckout(ctx, &sync.Map{}))) } } diff --git a/act/runner/runner_test.go b/act/runner/runner_test.go index c59b77d5..b466de40 100644 --- a/act/runner/runner_test.go +++ b/act/runner/runner_test.go @@ -13,11 +13,13 @@ import ( "path/filepath" "runtime" "strings" + "sync" "testing" "time" "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common/logging" + "gitea.com/gitea/runner/act/ghcontext" "gitea.dev/actionslib/pkg/model" "github.com/joho/godotenv" @@ -26,12 +28,13 @@ import ( ) var ( - baseImage = "node:24-bookworm-slim" - platforms map[string]string - logLevel = logging.LevelDebug - workdir = "testdata" - secrets map[string]string - testRunID = strings.ToLower(rand.Text()[:8]) // keeps docker names apart from concurrent test processes on one daemon + baseImage = "node:24-bookworm-slim" + platforms map[string]string + logLevel = logging.LevelDebug + workdir = "testdata" + secrets map[string]string + testRunID = strings.ToLower(rand.Text()[:8]) // keeps docker names apart from concurrent test processes on one daemon + testCheckoutPins sync.Map ) func mapPlatformPicker(platforms map[string]string) func([]string) string { @@ -158,12 +161,12 @@ func TestNewPlanExecutorRequiresRunsOnBeforeRunningAnyJob(t *testing.T) { require.NoError(t, err) plan, err := model.CombineWorkflowPlanner(workflow).PlanJob("test") require.NoError(t, err) - require.ErrorContains(t, (&runnerImpl{config: &Config{}}).NewPlanExecutor(plan)(t.Context()), `job "test" requires a non-empty 'runs-on'`, runsOn) + require.ErrorContains(t, (&runnerImpl{config: &Config{Env: testCheckoutEnv()}}).NewPlanExecutor(plan)(t.Context()), `job "test" requires a non-empty 'runs-on'`, runsOn) assert.Empty(t, workflow.Jobs["first"].Result, runsOn) } workflow := &model.Workflow{Jobs: map[string]*model.Job{"build": {Result: "success"}}} - require.NoError(t, (&runnerImpl{config: &Config{}}).NewPlanExecutor(&model.Plan{Stages: []*model.Stage{{Runs: []*model.Run{{Workflow: workflow, JobID: "build"}}}}})(t.Context())) + require.NoError(t, (&runnerImpl{config: &Config{Env: testCheckoutEnv()}}).NewPlanExecutor(&model.Plan{Stages: []*model.Stage{{Runs: []*model.Run{{Workflow: workflow, JobID: "build"}}}}})(t.Context())) assert.Equal(t, "success", workflow.Jobs["build"].Result) } @@ -219,6 +222,7 @@ type TestJobFileInfo struct { } func (j *TestJobFileInfo) runTest(ctx context.Context, t *testing.T, cfg *Config) { + ctx = ghcontext.WithPinnedCheckout(ctx, &testCheckoutPins) fmt.Printf("::group::%s\n", j.workflowPath) //nolint:forbidigo // pre-existing issue from nektos/act logging.SetLevel(logLevel) @@ -481,7 +485,7 @@ func TestRunEventHostEnvironment(t *testing.T) { } func TestDryrunEvent(t *testing.T) { - t.Parallel() + t.Setenv("DOCKER_HOST", "://") // Dryrun plans without containers or network (shells and local actions only). ctx := common.WithDryrun(t.Context(), true) diff --git a/act/runner/step_action_local_test.go b/act/runner/step_action_local_test.go index 49d9c849..98fc841a 100644 --- a/act/runner/step_action_local_test.go +++ b/act/runner/step_action_local_test.go @@ -57,6 +57,7 @@ func TestStepActionLocalTest(t *testing.T) { ExprEval: &expressionEvaluator{}, Config: &Config{ Workdir: "/tmp", + Env: testCheckoutEnv(), }, Run: &model.Run{ JobID: "1", @@ -109,7 +110,7 @@ func TestStepActionPost(t *testing.T) { ctx := t.Context() cm := &containerMock{} rc := &RunContext{ - Config: &Config{GitHubInstance: "https://github.com", ActionCacheDir: t.TempDir()}, + Config: &Config{GitHubInstance: "https://github.com", ActionCacheDir: t.TempDir(), PresetGitHubContext: &model.GithubContext{}}, JobContainer: cm, Run: &model.Run{JobID: "1", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"1": {}}}}, StepResults: map[string]*model.StepResult{"step": {Conclusion: test.status}}, diff --git a/act/runner/step_action_remote_test.go b/act/runner/step_action_remote_test.go index 3326e802..ea0197c1 100644 --- a/act/runner/step_action_remote_test.go +++ b/act/runner/step_action_remote_test.go @@ -51,10 +51,12 @@ func TestShortSHAActionRejected(t *testing.T) { require.NoError(t, os.WriteFile(filepath.Join(workflowDir, "push.yml"), []byte(workflow), 0o644)) runner, err := New(&Config{ + Env: testCheckoutEnv(), Workdir: workflowDir, EventName: "push", GitHubInstance: "github.com", DefaultActionInstance: actionRoot, + ActionCacheDir: t.TempDir(), ContainerMaxLifetime: time.Hour, PlatformPicker: func([]string) string { return baseImage }, }) @@ -94,8 +96,9 @@ func TestStepActionRemote(t *testing.T) { sar := &stepActionRemote{ RunContext: &RunContext{ Config: &Config{ - GitHubInstance: "github.com", - ActionCacheDir: t.TempDir(), + GitHubInstance: "github.com", + ActionCacheDir: t.TempDir(), + PresetGitHubContext: &model.GithubContext{}, }, Run: &model.Run{ JobID: "1", @@ -134,7 +137,7 @@ func TestStepActionRemote(t *testing.T) { t.Run("refreshes composite inputs without leaking environment", func(t *testing.T) { step := &stepActionRemote{ Step: &model.Step{Uses: "org/composite@v1", With: map[string]string{"SHARED": "first"}}, - RunContext: &RunContext{Config: &Config{ActionCacheDir: t.TempDir()}, Run: &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}}, JobContainer: &jobContainerMock{}}, + RunContext: &RunContext{Config: &Config{ActionCacheDir: t.TempDir(), PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}}, JobContainer: &jobContainerMock{}}, action: &model.Action{Inputs: map[string]model.Input{"shared": {}}}, env: map[string]string{}, remoteAction: mustNewRemoteAction(t, "org/composite@v1"), @@ -161,7 +164,7 @@ func TestStepActionRemotePrepare(t *testing.T) { }) newJob := func(instance string) *RunContext { return &RunContext{ - Config: &Config{GitHubInstance: instance, ActionCacheDir: t.TempDir()}, + Config: &Config{GitHubInstance: instance, ActionCacheDir: t.TempDir(), PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "1", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"1": {}}}}, JobContainer: &jobContainerMock{}, } @@ -222,7 +225,7 @@ func TestStepActionRemotePreCleanup(t *testing.T) { sar := &stepActionRemote{ Step: &model.Step{ID: "step", Uses: "remote/action@v1", If: yaml.Node{Value: "false"}}, RunContext: &RunContext{ - Config: &Config{}, + Config: &Config{PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}}, StepResults: map[string]*model.StepResult{}, JobContainer: cm, @@ -405,6 +408,7 @@ func TestStepActionRemotePreResolvesDownloadedCommit(t *testing.T) { Step: &model.Step{Uses: "actions/setup-go@main"}, RunContext: &RunContext{ Config: &Config{ + Env: testCheckoutEnv(), GitHubInstance: "https://gitea.example.com", DefaultActionInstance: instance, ActionCacheDir: t.TempDir(), @@ -484,6 +488,7 @@ func TestStepActionRemoteCloneTokenSurvivesNilSecrets(t *testing.T) { Step: &model.Step{Uses: "org/repo@v1"}, RunContext: &RunContext{ Config: &Config{ + Env: testCheckoutEnv(), GitHubInstance: tt.gitHubInstance, DefaultActionInstance: tt.defaultActionInstance, ActionCacheDir: t.TempDir(), diff --git a/act/runner/step_docker_test.go b/act/runner/step_docker_test.go index ae2a92a9..7c15330f 100644 --- a/act/runner/step_docker_test.go +++ b/act/runner/step_docker_test.go @@ -36,6 +36,7 @@ func TestStepDockerMain(t *testing.T) { RunContext: &RunContext{ StepResults: map[string]*model.StepResult{}, Config: &Config{ + PresetGitHubContext: &model.GithubContext{}, Secrets: map[string]string{ "DOCKER_USERNAME": "docker-user", "DOCKER_PASSWORD": "docker-password", @@ -116,7 +117,8 @@ func TestStepDockerNewStepContainer(t *testing.T) { RunContext: &RunContext{ StepResults: map[string]*model.StepResult{}, Config: &Config{ - AllocatePTY: test.allocPTY, + PresetGitHubContext: &model.GithubContext{}, + AllocatePTY: test.allocPTY, PlatformPicker: func(_platforms []string) string { return test.platform }, diff --git a/act/runner/step_run_test.go b/act/runner/step_run_test.go index 90e3c20a..af6d328f 100644 --- a/act/runner/step_run_test.go +++ b/act/runner/step_run_test.go @@ -37,7 +37,7 @@ func TestStepRun(t *testing.T) { RunContext: &RunContext{ StepResults: map[string]*model.StepResult{}, ExprEval: &expressionEvaluator{}, - Config: &Config{}, + Config: &Config{Env: testCheckoutEnv()}, Run: &model.Run{ JobID: "1", Workflow: &model.Workflow{ @@ -167,7 +167,7 @@ func TestStepRunShellParity(t *testing.T) { sr := &stepRun{ RunContext: &RunContext{ - Config: &Config{}, + Config: &Config{PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "1", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"1": {}}}}, JobContainer: jobContainer, }, diff --git a/act/runner/step_test.go b/act/runner/step_test.go index 79086153..db524580 100644 --- a/act/runner/step_test.go +++ b/act/runner/step_test.go @@ -128,9 +128,7 @@ func (sm *stepMock) getEnv() *map[string]string { func TestSetupEnv(t *testing.T) { rc := &RunContext{ Config: &Config{ - Env: map[string]string{ - "GITHUB_RUN_ID": "runId", - }, + Env: mergeMaps(testCheckoutEnv(), map[string]string{"GITHUB_RUN_ID": "runId"}), }, Run: &model.Run{ JobID: "1", @@ -224,7 +222,7 @@ jobs: for _, value := range []string{"first", "second"} { t.Run(value, func(t *testing.T) { t.Parallel() - rc, err := (&runnerImpl{config: &Config{}}).newRunContext(t.Context(), &model.Run{Workflow: workflow, JobID: "test"}, map[string]any{"env": `{"WORKFLOW":"` + value + `${{ github.job }}","PRIORITY":"workflow"}`, "priority": "${{ github.job }}", "defaults": `{"shell":"` + value + `"}`}) + rc, err := (&runnerImpl{config: &Config{PresetGitHubContext: &model.GithubContext{}}}).newRunContext(t.Context(), &model.Run{Workflow: workflow, JobID: "test"}, map[string]any{"env": `{"WORKFLOW":"` + value + `${{ github.job }}","PRIORITY":"workflow"}`, "priority": "${{ github.job }}", "defaults": `{"shell":"` + value + `"}`}) require.NoError(t, err) rc.workflowCallInputs = map[string]any{"args": `{"name":"` + value + `${{ github.job }}","fetch-depth":1234567}`} enabled, err := rc.isEnabled(t.Context()) @@ -252,7 +250,7 @@ jobs: t.Run("rejects inputs that still contain an expression", func(t *testing.T) { workflow, err := model.ReadWorkflow(strings.NewReader("jobs:\n test:\n steps:\n - uses: ./act\n with: ${{ inputs.args }}\n")) require.NoError(t, err) - rc, err := (&runnerImpl{config: &Config{}}).newRunContext(t.Context(), &model.Run{Workflow: workflow, JobID: "test"}, nil) + rc, err := (&runnerImpl{config: &Config{PresetGitHubContext: &model.GithubContext{}}}).newRunContext(t.Context(), &model.Run{Workflow: workflow, JobID: "test"}, nil) require.NoError(t, err) rc.workflowCallInputs = map[string]any{"args": "${{ inputs.unresolved }}"} require.ErrorContains(t, setupInputs(t.Context(), &stepRun{RunContext: rc, Step: workflow.Jobs["test"].Steps[0].Clone(), env: map[string]string{}}), "with:") @@ -288,7 +286,7 @@ func TestIsStepEnabled(t *testing.T) { t.Run(condition.expression, func(t *testing.T) { step := &stepRun{ RunContext: &RunContext{ - Config: &Config{}, + Config: &Config{PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}}, StepResults: map[string]*model.StepResult{}, }, @@ -308,7 +306,7 @@ func TestIsStepEnabled(t *testing.T) { t.Run("env and step inputs are unavailable at every stage", func(t *testing.T) { step := &stepRun{ RunContext: &RunContext{ - Config: &Config{}, + Config: &Config{PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}}, Env: map[string]string{"INPUT_FORGED": "leaked"}, }, @@ -340,7 +338,7 @@ func TestIsContinueOnError(t *testing.T) { t.Run(test.name, func(t *testing.T) { step := &stepRun{ RunContext: &RunContext{ - Config: &Config{}, + Config: &Config{PresetGitHubContext: &model.GithubContext{}}, Run: &model.Run{JobID: "job", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"job": {}}}}, }, Step: &model.Step{}, @@ -363,7 +361,7 @@ func TestRunStepExecutorDoesNotLeakRefusalToNextStep(t *testing.T) { cm.On("UpdateFromEnv", mock.Anything, mock.Anything).Return(noopExecutor) rc := &RunContext{ - Config: &Config{Env: map[string]string{}}, + Config: &Config{Env: testCheckoutEnv()}, Run: &model.Run{ JobID: "1", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"1": {}}}, diff --git a/act/runner/testdata/uses-composite/composite_action/action.yml b/act/runner/testdata/uses-composite/composite_action/action.yml index e79736c3..d5b76e6a 100644 --- a/act/runner/testdata/uses-composite/composite_action/action.yml +++ b/act/runner/testdata/uses-composite/composite_action/action.yml @@ -53,33 +53,18 @@ runs: if [ "${{ inputs.test_input_required }}" != "test_input_required_value" ]; then exit 1 fi - shell: bash - - - run: | if [ "${{ inputs.test_input_optional }}" != "test_input_optional_value" ]; then exit 1 fi - shell: bash - - - run: | if [ "${{ inputs.test_input_optional_with_default_overriden }}" != "test_input_optional_with_default_overriden" ]; then exit 1 fi - shell: bash - - - run: | if [ "${{ inputs.test_input_required_with_default }}" != "test_input_optional_value" ]; then exit 1 fi - shell: bash - - - run: | if [ "${{ inputs.test_input_required_with_default_overriden }}" != "test_input_required_with_default_overriden" ]; then exit 1 fi - shell: bash - - - run: | if [ -z "$GITHUB_ACTION_PATH" ]; then exit 1 fi diff --git a/e2e/kubernetes_test.go b/e2e/kubernetes_test.go index 0e447326..1df48ac5 100644 --- a/e2e/kubernetes_test.go +++ b/e2e/kubernetes_test.go @@ -6,9 +6,11 @@ package e2e import ( + "bufio" "bytes" "context" "fmt" + "io" "net" "net/netip" "net/url" @@ -30,9 +32,7 @@ import ( func testKubernetes(t *testing.T) { t.Parallel() ctx := t.Context() - k3sID, kubernetes := startK3s(t) - - images := preloadImages(t, k3sID, os.Getenv("E2E_JOB_IMAGE"), os.Getenv("SERVICE_IMAGE")) + k3sID, kubernetes, images := startK3s(t) api, repo := newScenario(t) if err := api.CreateVariable(ctx, repo, "E2E_SERVICE_IMAGE", images[1]); err != nil { @@ -51,7 +51,7 @@ func testKubernetes(t *testing.T) { } } -func startK3s(t *testing.T) (string, *config.Kubernetes) { +func startK3s(t *testing.T) (string, *config.Kubernetes, []string) { t.Helper() ctx := t.Context() cli := fixture.cli @@ -93,6 +93,13 @@ func startK3s(t *testing.T) (string, *config.Kubernetes) { t.Fatalf("start k3s container: %v", 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) + } + 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 { @@ -139,7 +146,33 @@ func startK3s(t *testing.T) (string, *config.Kubernetes) { "hostnames": []any{giteaURL.Hostname()}, }}}} } - return created.ID, kubernetes + return created.ID, kubernetes, images +} + +func waitForContainerLog(ctx context.Context, id, message string) error { + logs, err := fixture.cli.ContainerLogs(ctx, id, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true}) + if err != nil { + return err + } + defer logs.Close() + + reader, writer := io.Pipe() + defer reader.Close() + go func() { + _, err := stdcopy.StdCopy(writer, writer, logs) + _ = writer.CloseWithError(err) + }() + + scanner := bufio.NewScanner(reader) + for scanner.Scan() { + if strings.Contains(scanner.Text(), message) { + return nil + } + } + if err := scanner.Err(); err != nil { + return err + } + return fmt.Errorf("container exited before logging %q", message) } func preloadImages(t *testing.T, k3sID string, images ...string) []string { diff --git a/internal/app/poll/poller_test.go b/internal/app/poll/poller_test.go index 88deafb1..27a3dc4e 100644 --- a/internal/app/poll/poller_test.go +++ b/internal/app/poll/poller_test.go @@ -9,6 +9,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "gitea.com/gitea/runner/act/common/logging/logtest" @@ -216,33 +217,35 @@ func TestPollerReady(t *testing.T) { } func TestPollerPausesAndResumesForLocalAvailability(t *testing.T) { - var fetches atomic.Int64 - cli := mocks.NewClient(t) - cli.On("FetchTask", mock.Anything, mock.Anything).Maybe().Run(func(mock.Arguments) { - fetches.Add(1) - }).Return(&runnerv1.FetchTaskResponse{}, nil) + synctest.Test(t, func(t *testing.T) { + var fetches atomic.Int64 + cli := mocks.NewClient(t) + cli.On("FetchTask", mock.Anything, mock.Anything).Maybe().Run(func(mock.Arguments) { + fetches.Add(1) + }).Return(&runnerv1.FetchTaskResponse{}, nil) - cfg, err := config.LoadDefault("") - require.NoError(t, err) - cfg.Runner.FetchInterval = 10 * time.Millisecond - cfg.Runner.FetchIntervalMax = 10 * time.Millisecond - runner := &availabilityRunner{reason: "low disk space"} - poller := New(cfg, cli, runner) + cfg, err := config.LoadDefault("") + require.NoError(t, err) + cfg.Runner.FetchInterval = 10 * time.Millisecond + cfg.Runner.FetchIntervalMax = 10 * time.Millisecond + runner := &availabilityRunner{reason: "low disk space"} + poller := New(cfg, cli, runner) - var wg sync.WaitGroup - wg.Go(poller.Poll) - time.Sleep(40 * time.Millisecond) - assert.Zero(t, fetches.Load(), "an unavailable runner must not fetch a task") + var wg sync.WaitGroup + wg.Go(poller.Poll) + <-time.After(40 * time.Millisecond) + assert.Zero(t, fetches.Load(), "an unavailable runner must not fetch a task") - runner.ready.Store(true) - require.Eventually(t, func() bool { - return fetches.Load() > 0 - }, time.Second, 10*time.Millisecond, "polling should resume after local recovery") + runner.ready.Store(true) + require.Eventually(t, func() bool { + return fetches.Load() > 0 + }, time.Second, 10*time.Millisecond, "polling should resume after local recovery") - shutdownCtx, cancel := context.WithTimeout(t.Context(), time.Second) - defer cancel() - require.NoError(t, poller.Shutdown(shutdownCtx)) - wg.Wait() + shutdownCtx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + require.NoError(t, poller.Shutdown(shutdownCtx)) + wg.Wait() + }) } func (m *mockRunner) Run(ctx context.Context, _ *runnerv1.Task) error { @@ -276,36 +279,38 @@ func TestPollerCallsOnIdle(t *testing.T) { name = "PollOnce" } t.Run(name, func(t *testing.T) { - cli := mocks.NewClient(t) - cli.On("FetchTask", mock.Anything, mock.Anything).Return(&runnerv1.FetchTaskResponse{}, nil) + synctest.Test(t, func(t *testing.T) { + cli := mocks.NewClient(t) + cli.On("FetchTask", mock.Anything, mock.Anything).Return(&runnerv1.FetchTaskResponse{}, nil) - cfg, err := config.LoadDefault("") - require.NoError(t, err) - if !pollOnce { - cfg.Runner.Capacity = 1 - } - cfg.Runner.FetchInterval = 10 * time.Millisecond - cfg.Runner.FetchIntervalMax = 10 * time.Millisecond - - runner := &idleAwareRunner{} - poller := New(cfg, cli, runner) - var wg sync.WaitGroup - wg.Go(func() { - if pollOnce { - poller.PollOnce(func() {}) - } else { - poller.Poll() + cfg, err := config.LoadDefault("") + require.NoError(t, err) + if !pollOnce { + cfg.Runner.Capacity = 1 } + cfg.Runner.FetchInterval = 10 * time.Millisecond + cfg.Runner.FetchIntervalMax = 10 * time.Millisecond + + runner := &idleAwareRunner{} + poller := New(cfg, cli, runner) + var wg sync.WaitGroup + wg.Go(func() { + if pollOnce { + poller.PollOnce(func() {}) + } else { + poller.Poll() + } + }) + + require.Eventually(t, func() bool { + return runner.idleCalls.Load() > 0 + }, time.Second, 10*time.Millisecond) + + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + require.NoError(t, poller.Shutdown(ctx)) + wg.Wait() }) - - require.Eventually(t, func() bool { - return runner.idleCalls.Load() > 0 - }, time.Second, 10*time.Millisecond) - - ctx, cancel := context.WithTimeout(t.Context(), time.Second) - defer cancel() - require.NoError(t, poller.Shutdown(ctx)) - wg.Wait() }) } } @@ -323,109 +328,113 @@ func TestPollerPollOnceCallsOnTaskFetchedBeforeRunningIt(t *testing.T) { } func TestPoller_ConcurrencyLimitedByCapacity(t *testing.T) { - const ( - capacity = 3 - totalTasks = 6 - taskDelay = 50 * time.Millisecond - ) + synctest.Test(t, func(t *testing.T) { + const ( + capacity = 3 + totalTasks = 6 + taskDelay = 50 * time.Millisecond + ) - var ( - tasksReturned atomic.Int64 - fetchConcur atomic.Int64 - maxFetchConcur atomic.Int64 - ) + var ( + tasksReturned atomic.Int64 + fetchConcur atomic.Int64 + maxFetchConcur atomic.Int64 + ) - cli := mocks.NewClient(t) - cli.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { - atomicMax(&maxFetchConcur, fetchConcur.Add(1)) - defer fetchConcur.Add(-1) + cli := mocks.NewClient(t) + cli.On("FetchTask", mock.Anything, mock.Anything).Return( + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { + atomicMax(&maxFetchConcur, fetchConcur.Add(1)) + defer fetchConcur.Add(-1) - n := tasksReturned.Add(1) - if n <= totalTasks { - return &runnerv1.FetchTaskResponse{ - Task: &runnerv1.Task{Id: n}, - }, nil - } - return &runnerv1.FetchTaskResponse{}, nil - }, - ) + n := tasksReturned.Add(1) + if n <= totalTasks { + return &runnerv1.FetchTaskResponse{ + Task: &runnerv1.Task{Id: n}, + }, nil + } + return &runnerv1.FetchTaskResponse{}, nil + }, + ) - runner := &mockRunner{delay: taskDelay} + runner := &mockRunner{delay: taskDelay} - cfg, err := config.LoadDefault("") - require.NoError(t, err) - cfg.Runner.Capacity = capacity - cfg.Runner.FetchInterval = 10 * time.Millisecond - cfg.Runner.FetchIntervalMax = 10 * time.Millisecond + cfg, err := config.LoadDefault("") + require.NoError(t, err) + cfg.Runner.Capacity = capacity + cfg.Runner.FetchInterval = 10 * time.Millisecond + cfg.Runner.FetchIntervalMax = 10 * time.Millisecond - poller := New(cfg, cli, runner) + poller := New(cfg, cli, runner) - var wg sync.WaitGroup - wg.Go(poller.Poll) + var wg sync.WaitGroup + wg.Go(poller.Poll) - require.Eventually(t, func() bool { - return runner.totalCompleted.Load() >= totalTasks - }, 2*time.Second, 10*time.Millisecond, "all tasks should complete") + require.Eventually(t, func() bool { + return runner.totalCompleted.Load() >= totalTasks + }, 2*time.Second, 10*time.Millisecond, "all tasks should complete") - ctx, cancel := context.WithTimeout(t.Context(), time.Second) - defer cancel() - err = poller.Shutdown(ctx) - require.NoError(t, err) - wg.Wait() + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + err = poller.Shutdown(ctx) + require.NoError(t, err) + wg.Wait() - assert.LessOrEqual(t, runner.maxConcurrent.Load(), int64(capacity), - "concurrent running tasks must not exceed capacity") - assert.GreaterOrEqual(t, runner.maxConcurrent.Load(), int64(2), - "with 6 tasks and capacity 3, at least 2 should overlap") - assert.Equal(t, int64(1), maxFetchConcur.Load(), - "FetchTask must never be called concurrently (single poller)") - assert.Equal(t, int64(totalTasks), runner.totalCompleted.Load(), - "all tasks should have been executed") + assert.LessOrEqual(t, runner.maxConcurrent.Load(), int64(capacity), + "concurrent running tasks must not exceed capacity") + assert.GreaterOrEqual(t, runner.maxConcurrent.Load(), int64(2), + "with 6 tasks and capacity 3, at least 2 should overlap") + assert.Equal(t, int64(1), maxFetchConcur.Load(), + "FetchTask must never be called concurrently (single poller)") + assert.Equal(t, int64(totalTasks), runner.totalCompleted.Load(), + "all tasks should have been executed") + }) } func TestPoller_ShutdownForcesJobsOnTimeout(t *testing.T) { - var served atomic.Bool - cli := mocks.NewClient(t) - cli.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { - if served.CompareAndSwap(false, true) { - return &runnerv1.FetchTaskResponse{ - Task: &runnerv1.Task{Id: 1}, - }, nil - } - return &runnerv1.FetchTaskResponse{}, nil - }, - ) + synctest.Test(t, func(t *testing.T) { + var served atomic.Bool + cli := mocks.NewClient(t) + cli.On("FetchTask", mock.Anything, mock.Anything).Return( + func(_ context.Context, _ *runnerv1.FetchTaskRequest) (*runnerv1.FetchTaskResponse, error) { + if served.CompareAndSwap(false, true) { + return &runnerv1.FetchTaskResponse{ + Task: &runnerv1.Task{Id: 1}, + }, nil + } + return &runnerv1.FetchTaskResponse{}, nil + }, + ) - runner := &mockRunner{delay: 30 * time.Second} + runner := &mockRunner{delay: 30 * time.Second} - cfg, err := config.LoadDefault("") - require.NoError(t, err) - cfg.Runner.Capacity = 1 - cfg.Runner.FetchInterval = 10 * time.Millisecond - cfg.Runner.FetchIntervalMax = 10 * time.Millisecond + cfg, err := config.LoadDefault("") + require.NoError(t, err) + cfg.Runner.Capacity = 1 + cfg.Runner.FetchInterval = 10 * time.Millisecond + cfg.Runner.FetchIntervalMax = 10 * time.Millisecond - poller := New(cfg, cli, runner) + poller := New(cfg, cli, runner) - var wg sync.WaitGroup - wg.Go(poller.Poll) + var wg sync.WaitGroup + wg.Go(poller.Poll) - require.Eventually(t, func() bool { - return runner.running.Load() == 1 - }, time.Second, 10*time.Millisecond, "task should start running") + require.Eventually(t, func() bool { + return runner.running.Load() == 1 + }, time.Second, 10*time.Millisecond, "task should start running") - ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) - defer cancel() - start := time.Now() - err = poller.Shutdown(ctx) - elapsed := time.Since(start) + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + start := time.Now() + err = poller.Shutdown(ctx) + elapsed := time.Since(start) - require.ErrorIs(t, err, context.DeadlineExceeded) - assert.Less(t, elapsed, 5*time.Second, - "Shutdown must not block on the parked task; shutdownJobs() must run on timeout") + require.ErrorIs(t, err, context.DeadlineExceeded) + assert.Less(t, elapsed, 5*time.Second, + "Shutdown must not block on the parked task; shutdownJobs() must run on timeout") - wg.Wait() - assert.Equal(t, int64(1), runner.totalCompleted.Load(), - "the parked task must be cancelled and unwound") + wg.Wait() + assert.Equal(t, int64(1), runner.totalCompleted.Load(), + "the parked task must be cancelled and unwound") + }) } diff --git a/internal/pkg/report/reporter_test.go b/internal/pkg/report/reporter_test.go index 0367a546..64c1e9d0 100644 --- a/internal/pkg/report/reporter_test.go +++ b/internal/pkg/report/reporter_test.go @@ -18,6 +18,7 @@ import ( "testing/synctest" "time" + "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common/logging" "gitea.com/gitea/runner/act/common/logging/logtest" "gitea.com/gitea/runner/act/runner" @@ -198,34 +199,36 @@ func TestReporter_LogFlush(t *testing.T) { {"batch size", 10 * time.Second, 5, []string{"log line 0", "log line 1", "log line 2", "log line 3", "log line 4"}}, } { t.Run(tc.name, func(t *testing.T) { - var updateLogCalls atomic.Int64 - client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Return( - func(_ctx context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { - updateLogCalls.Add(1) - return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil - }, - ) - client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(&runnerv1.UpdateTaskResponse{}, nil) + synctest.Test(t, func(t *testing.T) { + var updateLogCalls atomic.Int64 + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Return( + func(_ctx context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + updateLogCalls.Add(1) + return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil + }, + ) + client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(&runnerv1.UpdateTaskResponse{}, nil) - cfg, _ := config.LoadDefault("") - cfg.Runner.LogReportInterval = 10 * time.Second - cfg.Runner.LogReportMaxLatency = tc.maxLatency - cfg.Runner.LogReportBatchSize = tc.batchSize - reporter := newTestReporter(t, client, cfg) - reporter.ResetSteps(newTestSteps(1)) - reporter.RunDaemon() - defer func() { require.NoError(t, reporter.Close("")) }() + cfg, _ := config.LoadDefault("") + cfg.Runner.LogReportInterval = 10 * time.Second + cfg.Runner.LogReportMaxLatency = tc.maxLatency + cfg.Runner.LogReportBatchSize = tc.batchSize + reporter := newTestReporter(t, client, cfg) + reporter.ResetSteps(newTestSteps(1)) + reporter.RunDaemon() + defer func() { require.NoError(t, reporter.Close("")) }() - for _, message := range tc.messages { - require.NoError(t, fire(reporter, logEntry{ - Message: message, - Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}, - })) - } - assert.Eventually(t, func() bool { - return updateLogCalls.Load() > 0 - }, 500*time.Millisecond, 10*time.Millisecond, "logs must flush before the periodic timers") + for _, message := range tc.messages { + require.NoError(t, fire(reporter, logEntry{ + Message: message, + Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}, + })) + } + assert.Eventually(t, func() bool { + return updateLogCalls.Load() > 0 + }, 500*time.Millisecond, 10*time.Millisecond, "logs must flush before the periodic timers") + }) }) } } @@ -313,43 +316,45 @@ func TestReporter_StateChangedRestoredOnError(t *testing.T) { } func TestReporter_StateNotifyFlush(t *testing.T) { - var updateTaskCalls atomic.Int64 + synctest.Test(t, func(t *testing.T) { + var updateTaskCalls atomic.Int64 - client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { - return &runnerv1.UpdateLogResponse{ - AckIndex: req.Index + int64(len(req.Rows)), - }, nil - }, - ) - client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { - updateTaskCalls.Add(1) - return &runnerv1.UpdateTaskResponse{}, nil - }, - ) + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return( + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil + }, + ) + client.On("UpdateTask", mock.Anything, mock.Anything).Return( + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + updateTaskCalls.Add(1) + return &runnerv1.UpdateTaskResponse{}, nil + }, + ) - cfg, _ := config.LoadDefault("") - cfg.Runner.StateReportInterval = 10 * time.Second - cfg.Runner.LogReportInterval = 10 * time.Second + cfg, _ := config.LoadDefault("") + cfg.Runner.StateReportInterval = 10 * time.Second + cfg.Runner.LogReportInterval = 10 * time.Second - reporter := newTestReporter(t, client, cfg) - reporter.ResetSteps(newTestSteps(1)) - reporter.RunDaemon() - defer func() { - _ = reporter.Close("") - }() + reporter := newTestReporter(t, client, cfg) + reporter.ResetSteps(newTestSteps(1)) + reporter.RunDaemon() + defer func() { + _ = reporter.Close("") + }() - require.NoError(t, fire(reporter, logEntry{ - Message: "step starting", - Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}, - })) + require.NoError(t, fire(reporter, logEntry{ + Message: "step starting", + Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}, + })) - assert.Eventually(t, func() bool { - return updateTaskCalls.Load() > 0 - }, 500*time.Millisecond, 10*time.Millisecond, - "step transition should have triggered immediate state flush via stateNotify") + assert.Eventually(t, func() bool { + return updateTaskCalls.Load() > 0 + }, 500*time.Millisecond, 10*time.Millisecond, + "step transition should have triggered immediate state flush via stateNotify") + }) } func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) { @@ -401,33 +406,35 @@ func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) { } func TestReporter_StateHeartbeat(t *testing.T) { - var updateTaskCalls atomic.Int64 + synctest.Test(t, func(t *testing.T) { + var updateTaskCalls atomic.Int64 - client := mocks.NewClient(t) - client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { - updateTaskCalls.Add(1) - return &runnerv1.UpdateTaskResponse{}, nil - }, - ) + client := mocks.NewClient(t) + client.On("UpdateTask", mock.Anything, mock.Anything).Return( + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + updateTaskCalls.Add(1) + return &runnerv1.UpdateTaskResponse{}, nil + }, + ) - cfg, _ := config.LoadDefault("") - cfg.Runner.StateReportInterval = 50 * time.Millisecond - reporter := newTestReporter(t, client, cfg) - reporter.ResetSteps(newTestSteps(1)) + cfg, _ := config.LoadDefault("") + cfg.Runner.StateReportInterval = 50 * time.Millisecond + reporter := newTestReporter(t, client, cfg) + reporter.ResetSteps(newTestSteps(1)) - reporter.stateMu.Lock() - reporter.stateChanged = true - reporter.stateMu.Unlock() - require.NoError(t, reporter.ReportState(false)) - require.Equal(t, int64(1), updateTaskCalls.Load()) + reporter.stateMu.Lock() + reporter.stateChanged = true + reporter.stateMu.Unlock() + require.NoError(t, reporter.ReportState(false)) + require.Equal(t, int64(1), updateTaskCalls.Load()) - require.NoError(t, reporter.ReportState(false)) - assert.Equal(t, int64(1), updateTaskCalls.Load(), "no-op ReportState within stateReportInterval must skip") + require.NoError(t, reporter.ReportState(false)) + assert.Equal(t, int64(1), updateTaskCalls.Load(), "no-op ReportState within stateReportInterval must skip") - time.Sleep(2 * cfg.Runner.StateReportInterval) - require.NoError(t, reporter.ReportState(false)) - assert.Equal(t, int64(2), updateTaskCalls.Load(), "ReportState must heartbeat after stateReportInterval even with no state change") + <-time.After(2 * cfg.Runner.StateReportInterval) + require.NoError(t, reporter.ReportState(false)) + assert.Equal(t, int64(2), updateTaskCalls.Load(), "ReportState must heartbeat after stateReportInterval even with no state change") + }) } func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) { @@ -556,58 +563,60 @@ func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) { } func TestReporter_StopHeartbeats(t *testing.T) { - var updateTaskCalls atomic.Int64 + synctest.Test(t, func(t *testing.T) { + var updateTaskCalls atomic.Int64 - client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return( - func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { - return &runnerv1.UpdateLogResponse{ - AckIndex: req.Index + int64(len(req.Rows)), - }, nil - }, - ) - client.On("UpdateTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { - updateTaskCalls.Add(1) - return &runnerv1.UpdateTaskResponse{}, nil - }, - ) + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Maybe().Return( + func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil + }, + ) + client.On("UpdateTask", mock.Anything, mock.Anything).Return( + func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + updateTaskCalls.Add(1) + return &runnerv1.UpdateTaskResponse{}, nil + }, + ) - cfg, err := config.LoadDefault("") - require.NoError(t, err) - cfg.Runner.StateReportInterval = 20 * time.Millisecond - cfg.Runner.LogReportInterval = time.Hour + cfg, err := config.LoadDefault("") + require.NoError(t, err) + cfg.Runner.StateReportInterval = 20 * time.Millisecond + cfg.Runner.LogReportInterval = time.Hour - reporter := newTestReporter(t, client, cfg) - reporter.ResetSteps(newTestSteps(1)) - reporter.RunDaemon() + reporter := newTestReporter(t, client, cfg) + reporter.ResetSteps(newTestSteps(1)) + reporter.RunDaemon() - reporter.stateMu.Lock() - reporter.stateChanged = true - reporter.state.Result = runnerv1.Result_RESULT_SUCCESS - reporter.state.StoppedAt = timestamppb.Now() - reporter.stateMu.Unlock() + reporter.stateMu.Lock() + reporter.stateChanged = true + reporter.state.Result = runnerv1.Result_RESULT_SUCCESS + reporter.state.StoppedAt = timestamppb.Now() + reporter.stateMu.Unlock() - require.Eventually(t, func() bool { - return updateTaskCalls.Load() >= 1 - }, time.Second, 5*time.Millisecond, "daemon must send at least one UpdateTask before StopHeartbeats") + require.Eventually(t, func() bool { + return updateTaskCalls.Load() >= 1 + }, time.Second, 5*time.Millisecond, "daemon must send at least one UpdateTask before StopHeartbeats") - beforeStop := updateTaskCalls.Load() - reporter.StopHeartbeats() + beforeStop := updateTaskCalls.Load() + reporter.StopHeartbeats() - select { - case <-reporter.daemon: - case <-time.After(time.Second): - t.Fatal("StopHeartbeats must stop the daemon loop") - } + select { + case <-reporter.daemon: + case <-time.After(time.Second): + t.Fatal("StopHeartbeats must stop the daemon loop") + } - time.Sleep(3 * cfg.Runner.StateReportInterval) - assert.Equal(t, beforeStop, updateTaskCalls.Load(), - "UpdateTask must not be called after StopHeartbeats") + <-time.After(3 * cfg.Runner.StateReportInterval) + assert.Equal(t, beforeStop, updateTaskCalls.Load(), + "UpdateTask must not be called after StopHeartbeats") - require.NoError(t, reporter.Close("")) - assert.Greater(t, updateTaskCalls.Load(), beforeStop, - "Close() must still send a final UpdateTask after StopHeartbeats") + require.NoError(t, reporter.Close("")) + assert.Greater(t, updateTaskCalls.Load(), beforeStop, + "Close() must still send a final UpdateTask after StopHeartbeats") + }) } func TestAppendLogRow(t *testing.T) { @@ -786,48 +795,53 @@ func TestReporter_AckIndexBeyondBuffer(t *testing.T) { } func TestReporter_CloseWithStuckDaemon(t *testing.T) { - release := make(chan struct{}) - entered := make(chan struct{}) - var once sync.Once + synctest.Test(t, func(t *testing.T) { + release := make(chan struct{}) + entered := make(chan struct{}) + var once sync.Once + output := logging.Output() + logging.SetOutput(common.NewLineWriter(func(line string) bool { + if strings.Contains(line, "No Response from RunDaemon") { + close(release) + } + return true + })) + t.Cleanup(func() { logging.SetOutput(output) }) - client := mocks.NewClient(t) - client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { - once.Do(func() { - close(entered) - <-release + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) { + once.Do(func() { + close(entered) + <-release + }) + return &runnerv1.UpdateLogResponse{ + AckIndex: req.Index + int64(len(req.Rows)), + }, nil }) - return &runnerv1.UpdateLogResponse{ - AckIndex: req.Index + int64(len(req.Rows)), - }, nil + client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { + return &runnerv1.UpdateTaskResponse{}, nil + }).Maybe() + + cfg, _ := config.LoadDefault("") + cfg.Runner.LogReportInterval = 10 * time.Millisecond + reporter := newTestReporter(t, client, cfg) + reporter.daemonWait = time.Millisecond + reporter.RunDaemon() + reporter.ResetSteps(newTestSteps(1)) + + require.NoError(t, fire(reporter, logEntry{ + Message: "hello", + Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}, + Level: logging.LevelInfo, + })) + + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("daemon never reached UpdateLog") + } + require.NoError(t, reporter.Close("")) }) - client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) { - return &runnerv1.UpdateTaskResponse{}, nil - }).Maybe() - - cfg, _ := config.LoadDefault("") - cfg.Runner.LogReportInterval = 10 * time.Millisecond - reporter := newTestReporter(t, client, cfg) - reporter.daemonWait = time.Millisecond - reporter.RunDaemon() - reporter.ResetSteps(newTestSteps(1)) - - require.NoError(t, fire(reporter, logEntry{ - Message: "hello", - Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}, - Level: logging.LevelInfo, - })) - - select { - case <-entered: - case <-time.After(5 * time.Second): - t.Fatal("daemon never reached UpdateLog") - } - go func() { - time.Sleep(10 * reporter.daemonWait) - close(release) - }() - - require.NoError(t, reporter.Close("")) } func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) { diff --git a/scripts/test-dind.sh b/scripts/test-dind.sh index 7d93ee83..abe00dd2 100755 --- a/scripts/test-dind.sh +++ b/scripts/test-dind.sh @@ -33,7 +33,7 @@ host_docker="${DOCKER_HOST:-$(docker context inspect --format '{{.Endpoints.dock test_dir="" cleanup() { - docker -H "$host_docker" rm -fv "$name" "$name-s6" >/dev/null 2>&1 || true + docker -H "$host_docker" rm -fv "$name" "$name-s6-failure" "$name-s6-stop" >/dev/null 2>&1 || true if [ -n "$test_dir" ]; then rm -rf "$test_dir" fi @@ -52,7 +52,7 @@ else fi # Override the image entrypoint (s6) and run only dockerd, exposed over insecure TCP. # We are testing the daemon the image ships, not the runner supervision tree. - daemon_args=(-e DOCKER_TLS_CERTDIR= --entrypoint dockerd-entrypoint.sh "$image" --host=tcp://0.0.0.0:2375) + daemon_args=(-e DOCKER_TLS_CERTDIR= --entrypoint dockerd-entrypoint.sh "$image" --host=tcp://0.0.0.0:2375 --tls=false) if [ $# -eq 0 ]; then default_tests=true set -- -count=1 -race -run '^TestImageExistsLocally$' ./act/container/ @@ -87,6 +87,11 @@ fi # Create the dind container on the host daemon first, then repoint DOCKER_HOST at it: exporting # DOCKER_HOST before `docker run` would make this `docker run` target the not-yet-existent dind. docker -H "$host_docker" run -d --privileged --name "$name" "${run_args[@]}" "${daemon_args[@]}" >/dev/null +if [ "$default_tests" = true ]; then + echo "==> Starting ${target} supervision checks" + docker -H "$host_docker" run -d --privileged --name "$name-s6-failure" -e DOCKER_TLS_CERTDIR= -e RUNNER_STATE_FILE=/etc/hostname "$image" >/dev/null + docker -H "$host_docker" run -d --privileged --name "$name-s6-stop" -e DOCKER_TLS_CERTDIR= "$image" >/dev/null +fi export DOCKER_HOST="$daemon_host" echo "==> Waiting for daemon" @@ -148,7 +153,6 @@ if [ "$default_tests" = true ]; then docker -H "$host_docker" exec -e DOCKER_HOST="$socket" "$name" docker run --rm -v "${socket#unix://}:/var/run/docker.sock" -v /tmp/gitea-runner-proxy-test:/data -w /data \ -e ACT_TEST_DOCKER_PROXY=proxy -e ACT_TEST_IMAGE="$job_image" "$job_image" ./runner.test -test.v -test.run '^TestDockerProxyMountedJob$' -test.timeout 3m - supervised="$name-s6" supervised_exit_is() { if [ "$(docker -H "$host_docker" inspect -f '{{.State.ExitCode}}' "$supervised")" != "$1" ] || ! docker -H "$host_docker" logs "$supervised" 2>&1 | grep -F "$2" >/dev/null; then @@ -158,15 +162,14 @@ if [ "$default_tests" = true ]; then fi } echo "==> Checking that a runner failure sets the ${target} container exit code" - docker -H "$host_docker" run -d --privileged --name "$supervised" -e DOCKER_TLS_CERTDIR= -e RUNNER_STATE_FILE=/etc/hostname "$image" >/dev/null - for _ in $(seq 1 120); do - [ "$(docker -H "$host_docker" inspect -f '{{.State.Running}}' "$supervised")" = false ] && break - sleep 1 - done + supervised="$name-s6-failure" + if ! timeout 120 docker -H "$host_docker" wait "$supervised" >/dev/null; then + docker -H "$host_docker" logs "$supervised" >&2 + exit 1 + fi supervised_exit_is 1 'registration file not found' - docker -H "$host_docker" rm -fv "$supervised" >/dev/null echo "==> Checking that stopping a started ${target} runner exits 0" - docker -H "$host_docker" run -d --privileged --name "$supervised" -e DOCKER_TLS_CERTDIR= "$image" >/dev/null + supervised="$name-s6-stop" for _ in $(seq 1 120); do docker -H "$host_docker" logs "$supervised" 2>&1 | grep -F 'is missing or not a regular file' >/dev/null && break sleep 1 diff --git a/tools/test-e2e.sh b/tools/test-e2e.sh index d4b4f104..80173397 100755 --- a/tools/test-e2e.sh +++ b/tools/test-e2e.sh @@ -7,9 +7,20 @@ set -euo pipefail : "${E2E_K3S_IMAGE:?E2E_K3S_IMAGE is required}" : "${SERVICE_IMAGE:?SERVICE_IMAGE is required}" -docker pull "$E2E_JOB_IMAGE" -docker pull "$SERVICE_IMAGE" -docker pull "$E2E_GITEA_IMAGE" -docker pull "$E2E_K3S_IMAGE" +platform=$(docker version --format '{{.Server.Os}}/{{.Server.Arch}}') +pulls=() +for image in "$E2E_JOB_IMAGE" "$SERVICE_IMAGE" "$E2E_GITEA_IMAGE" "$E2E_K3S_IMAGE"; do + if [[ "$image" == *@sha256:* ]] && [[ "$(docker image inspect --format '{{.Os}}/{{.Architecture}}' "$image" 2>/dev/null)" == "$platform" ]]; then + continue + fi + docker pull "$image" & + pulls+=("$!") +done + +result=0 +for pull in "${pulls[@]}"; do + wait "$pull" || result=$? +done +((result == 0)) || exit "$result" exec "${GO:-go}" test -tags e2e -count=1 -parallel "$E2E_CONCURRENCY" -timeout 20m -v ./e2e/...