From dafe8d8d0c626b5a63e180eb2de5e242ecca2248 Mon Sep 17 00:00:00 2001 From: silverwind Date: Mon, 26 Oct 2026 17:25:17 +0000 Subject: [PATCH] fix: keep dry runs off the docker daemon and speed up the tests (#1349) Without a docker daemon six `act/runner` tests failed instead of passing. A dry run still created and removed job networks, the job teardown dropped the dry-run flag along with every other context value, and the container mocks asked the daemon for the runner arch. This also speeds up the tests and CI without dropping assertions. Most of the time went to git and docker round trips, which are now pinned, batched or coalesced, and the dind and e2e steps cache or overlap more of their setup. On CI, `act/runner` went from 77s to 55s and `test-dind` from 108s to 55s. Locally, `act/runner` without docker went from 53s to 7s. Split out of https://gitea.com/gitea/runner/pulls/1348, which adds Windows CI on top. The go cache changes moved to https://gitea.com/gitea/runner/pulls/1350. Written by Claude Code. --------- Co-authored-by: bircni Reviewed-on: https://gitea.com/gitea/runner/pulls/1349 Reviewed-by: bircni Co-authored-by: silverwind --- .dockerignore | 5 + .gitea/workflows/test.yml | 12 +- Dockerfile | 4 +- act/artifactcache/handler_test.go | 22 ++ act/artifactcache/handler_v2_test.go | 5 + act/artifactcache/results_test.go | 1 + act/artifactcache/s3_storage_test.go | 3 + act/common/git/fetch_action_test.go | 23 +- act/common/git/git.go | 21 +- act/common/git/git_test.go | 87 +++-- act/container/docker_images_test.go | 48 +-- act/container/docker_network.go | 16 +- act/container/docker_proxy.go | 77 +++- act/container/docker_proxy_test.go | 142 +++++++- act/container/docker_ssh_test.go | 1 + act/container/host_environment.go | 18 +- act/container/host_environment_test.go | 158 ++++---- act/container/kubernetes_test.go | 19 +- act/container/parse_env_file.go | 16 +- act/container/parse_env_file_test.go | 2 + act/ghcontext/github_context.go | 8 +- act/ghcontext/github_context_test.go | 10 +- act/runner/action.go | 9 +- act/runner/action_composite_test.go | 2 +- act/runner/action_test.go | 67 +++- act/runner/container_mock_test.go | 10 +- act/runner/expression_test.go | 12 +- act/runner/helpers_test.go | 4 + act/runner/job_executor.go | 2 +- act/runner/job_executor_test.go | 10 +- act/runner/job_hooks_test.go | 2 +- act/runner/max_parallel_test.go | 4 +- act/runner/patch_actions_e2e_test.go | 3 + act/runner/patch_actions_test.go | 42 ++- act/runner/reusable_workflow_test.go | 4 +- act/runner/run_context.go | 2 + act/runner/run_context_test.go | 14 +- act/runner/runner.go | 2 +- act/runner/runner_test.go | 22 +- act/runner/step_action_local_test.go | 3 +- act/runner/step_action_remote_test.go | 15 +- act/runner/step_docker_test.go | 4 +- act/runner/step_run_test.go | 4 +- act/runner/step_test.go | 16 +- .../composite_action/action.yml | 15 - e2e/kubernetes_test.go | 43 ++- internal/app/poll/poller_test.go | 277 +++++++------- internal/pkg/report/reporter_test.go | 338 +++++++++--------- scripts/test-dind.sh | 23 +- tools/test-e2e.sh | 19 +- 50 files changed, 1066 insertions(+), 600 deletions(-) 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/...