From 61f0cfa951b11562b9050e48945fd1784b87fc47 Mon Sep 17 00:00:00 2001 From: silverwind Date: Tue, 28 Jul 2026 14:34:04 +0000 Subject: [PATCH] test: speed up (#1121) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Full suite 178s → 96s, and a run's log 12MB → 2KB. `act/runner` was 177s of the 178s, serialised behind one docker daemon. - Its fixtures now run in parallel, bounded by a slot count instead of `go test -parallel`, with a per-test container name prefix and a pinned `MaxParallel`. - Fixtures asserting a job failure leaked their container and network (`AutoRemove` was at the act-CLI default), filling the daemon's address pool over time. - Replaced sleeps used as synchronisation in the parallel-executor and cache-handler tests. - Dropped duplicate coverage: two files re-testing `NewParallelExecutor`, a test asserting on its own semaphore, and `TestDockerActionForcePullForceRebuild`, whose config `runTest` discarded. - `fmt-check`/`security-check` move from `make test` to a `checks` target; `security-check` no longer installs `xgo` and `gxz`. - Test flags follow gitea: `GOTEST_FLAGS ?= -race -timeout 20m -parallel 8`, with `-cover`/`-coverprofile` left in the target. Dropped `-v`, since a failing package still prints its full output without it. Coverage unchanged at 73.4%. Reviewed-on: https://gitea.com/gitea/runner/pulls/1121 Reviewed-by: bircni Co-authored-by: silverwind --- .gitea/workflows/test.yml | 2 + Makefile | 11 +- act/artifactcache/handler_test.go | 30 ++- act/common/executor_max_parallel_test.go | 89 ------- act/common/executor_parallel_advanced_test.go | 221 ------------------ act/common/executor_test.go | 78 ++++--- act/runner/helpers_test.go | 14 -- act/runner/job_executor_test.go | 2 + act/runner/runner_max_parallel_test.go | 109 --------- act/runner/runner_test.go | 67 +++--- 10 files changed, 117 insertions(+), 506 deletions(-) delete mode 100644 act/common/executor_max_parallel_test.go delete mode 100644 act/common/executor_parallel_advanced_test.go delete mode 100644 act/runner/runner_max_parallel_test.go diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index e65011a1..7c32f3b6 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -33,6 +33,8 @@ jobs: done - name: lint run: make lint + - name: checks + run: make checks - name: build run: make build - name: test diff --git a/Makefile b/Makefile index 4c7deeb4..d41c6b9e 100644 --- a/Makefile +++ b/Makefile @@ -21,6 +21,8 @@ DOCKER_ROOTLESS_REF := $(DOCKER_IMAGE):$(DOCKER_TAG)-dind-rootless GOLANGCI_LINT_PACKAGE ?= github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.12.2 GOVULNCHECK_PACKAGE ?= golang.org/x/vuln/cmd/govulncheck@v1.3.0 +GOTEST_FLAGS ?= -race -timeout 20m -parallel 8 + STATIC ?= EXTLDFLAGS ?= ifneq ($(STATIC),) @@ -110,6 +112,9 @@ deps-tools: ## install tool dependencies $(GO) install $(GOVULNCHECK_PACKAGE) & \ wait +.PHONY: checks +checks: tidy-check fmt-check security-check ## run the non-lint source checks + .PHONY: lint lint: lint-go lint-go-windows ## lint everything @@ -131,7 +136,7 @@ lint-pr-title: ## lint PR title against Conventional Commits (set PR_TITLE=...) @node ./tools/lint-pr-title.ts .PHONY: security-check -security-check: deps-tools +security-check: GOEXPERIMENT= $(GO) run $(GOVULNCHECK_PACKAGE) -show color ./... || true .PHONY: tidy @@ -148,8 +153,8 @@ tidy-check: tidy fi .PHONY: test -test: fmt-check security-check ## test everything (integration tests self-skip without docker/network) - @$(GO) test -race -timeout 20m -v -cover -coverprofile coverage.txt ./... && echo "\n==>\033[32m Ok\033[m\n" || exit 1 +test: ## test everything (integration tests self-skip without docker/network) + @$(GO) test $(GOTEST_FLAGS) -cover -coverprofile coverage.txt ./... && echo "\n==>\033[32m Ok\033[m\n" || exit 1 .PHONY: coverage-report coverage-report: ## turn coverage.txt from `make test` into .tmp/coverage.md diff --git a/act/artifactcache/handler_test.go b/act/artifactcache/handler_test.go index 71e84e64..012bee70 100644 --- a/act/artifactcache/handler_test.go +++ b/act/artifactcache/handler_test.go @@ -445,13 +445,6 @@ func TestHandler(t *testing.T) { require.Equal(t, 404, resp.StatusCode) }) - t.Run("get with not exist id", func(t *testing.T) { - resp, err := testClient.Get(signArtifactURL(handler, 100)) - require.NoError(t, err) - defer resp.Body.Close() - require.Equal(t, 404, resp.StatusCode) - }) - t.Run("get with multiple keys", func(t *testing.T) { version := "c19da02a2bd7e77277f1ac29ab45c09b7d46a4ee758284e26bb3045ad11d9d20" key := strings.ToLower(t.Name()) @@ -469,7 +462,8 @@ func TestHandler(t *testing.T) { _, err := rand.Read(contents[i]) require.NoError(t, err) uploadCacheNormally(t, base, keys[i], version, contents[i]) - time.Sleep(time.Second) // ensure CreatedAt of caches are different + // ensure CreatedAt of caches are different, in upload order + backdateCache(t, handler, keys[i], time.Duration(len(contents)-i)*time.Second) } reqKeys := strings.Join([]string{ @@ -554,7 +548,8 @@ func TestHandler(t *testing.T) { _, err := rand.Read(contents[i]) require.NoError(t, err) uploadCacheNormally(t, base, keys[i], version, contents[i]) - time.Sleep(time.Second) // ensure CreatedAt of caches are different + // ensure CreatedAt of caches are different, in upload order + backdateCache(t, handler, keys[i], time.Duration(len(contents)-i)*time.Second) } reqKeys := strings.Join([]string{ @@ -607,7 +602,8 @@ func TestHandler(t *testing.T) { _, err := rand.Read(contents[i]) require.NoError(t, err) uploadCacheNormally(t, base, keys[i], version, contents[i]) - time.Sleep(time.Second) // ensure CreatedAt of caches are different + // ensure CreatedAt of caches are different, in upload order + backdateCache(t, handler, keys[i], time.Duration(len(contents)-i)*time.Second) } reqKeys := strings.Join([]string{ @@ -646,6 +642,20 @@ func TestHandler(t *testing.T) { }) } +// backdateCache rewrites a cache's CreatedAt. It has one-second resolution, so age-ordering +// tests set it directly instead of sleeping a second between uploads. +func backdateCache(t *testing.T, handler *Handler, key string, age time.Duration) { + db, err := handler.openDB() + require.NoError(t, err) + defer db.Close() + + var caches []*Cache + require.NoError(t, db.Find(&caches, bolthold.Where("Key").Eq(key))) + require.Len(t, caches, 1) + caches[0].CreatedAt = time.Now().Add(-age).Unix() + require.NoError(t, db.Update(caches[0].ID, caches[0])) +} + func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) { //nolint:unparam // pre-existing issue from nektos/act var id uint64 { diff --git a/act/common/executor_max_parallel_test.go b/act/common/executor_max_parallel_test.go deleted file mode 100644 index 7b3004d7..00000000 --- a/act/common/executor_max_parallel_test.go +++ /dev/null @@ -1,89 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package common - -import ( - "context" - "sync" - "sync/atomic" - "testing" - "time" - - "github.com/stretchr/testify/assert" -) - -// Simple fast test that verifies max-parallel: 2 limits concurrency -func TestMaxParallel2Quick(t *testing.T) { - ctx := context.Background() - - var currentRunning atomic.Int32 - var maxSimultaneous atomic.Int32 - - executors := make([]Executor, 4) - for i := range 4 { - executors[i] = func(ctx context.Context) error { - current := currentRunning.Add(1) - - // Update max if needed - for { - maxValue := maxSimultaneous.Load() - if current <= maxValue || maxSimultaneous.CompareAndSwap(maxValue, current) { - break - } - } - - time.Sleep(10 * time.Millisecond) - currentRunning.Add(-1) - return nil - } - } - - err := NewParallelExecutor(2, executors...)(ctx) - - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - assert.LessOrEqual(t, maxSimultaneous.Load(), int32(2), - "Should not exceed max-parallel: 2") -} - -// Test that verifies max-parallel: 1 enforces sequential execution -func TestMaxParallel1Sequential(t *testing.T) { - ctx := context.Background() - - var currentRunning atomic.Int32 - var maxSimultaneous atomic.Int32 - var executionOrder []int - var orderMutex sync.Mutex - - executors := make([]Executor, 5) - for i := range 5 { - taskID := i - executors[i] = func(ctx context.Context) error { - current := currentRunning.Add(1) - - // Track execution order - orderMutex.Lock() - executionOrder = append(executionOrder, taskID) - orderMutex.Unlock() - - // Update max if needed - for { - maxValue := maxSimultaneous.Load() - if current <= maxValue || maxSimultaneous.CompareAndSwap(maxValue, current) { - break - } - } - - time.Sleep(20 * time.Millisecond) - currentRunning.Add(-1) - return nil - } - } - - err := NewParallelExecutor(1, executors...)(ctx) - - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - assert.Equal(t, int32(1), maxSimultaneous.Load(), - "max-parallel: 1 should only run 1 task at a time") - assert.Len(t, executionOrder, 5, "All 5 tasks should have executed") -} diff --git a/act/common/executor_parallel_advanced_test.go b/act/common/executor_parallel_advanced_test.go deleted file mode 100644 index 9e69927d..00000000 --- a/act/common/executor_parallel_advanced_test.go +++ /dev/null @@ -1,221 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package common - -import ( - "context" - "sync" - "sync/atomic" - "testing" - "time" - - "github.com/stretchr/testify/assert" -) - -// TestMaxParallelJobExecution tests actual job execution with max-parallel -func TestMaxParallelJobExecution(t *testing.T) { - t.Run("MaxParallel=1 Sequential", func(t *testing.T) { - var currentRunning atomic.Int32 - var maxConcurrent int32 - var executionOrder []int - var mu sync.Mutex - - executors := make([]Executor, 5) - for i := range 5 { - taskID := i - executors[i] = func(ctx context.Context) error { - current := currentRunning.Add(1) - - // Track max concurrent - for { - maxValue := atomic.LoadInt32(&maxConcurrent) - if current <= maxValue || atomic.CompareAndSwapInt32(&maxConcurrent, maxValue, current) { - break - } - } - - mu.Lock() - executionOrder = append(executionOrder, taskID) - mu.Unlock() - - time.Sleep(10 * time.Millisecond) - currentRunning.Add(-1) - return nil - } - } - - ctx := context.Background() - err := NewParallelExecutor(1, executors...)(ctx) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - - assert.Equal(t, int32(1), maxConcurrent, "Should never exceed 1 concurrent execution") - assert.Len(t, executionOrder, 5, "All tasks should execute") - }) - - t.Run("MaxParallel=3 Limited", func(t *testing.T) { - var currentRunning atomic.Int32 - var maxConcurrent int32 - - executors := make([]Executor, 10) - for i := range 10 { - executors[i] = func(ctx context.Context) error { - current := currentRunning.Add(1) - - for { - maxValue := atomic.LoadInt32(&maxConcurrent) - if current <= maxValue || atomic.CompareAndSwapInt32(&maxConcurrent, maxValue, current) { - break - } - } - - time.Sleep(20 * time.Millisecond) - currentRunning.Add(-1) - return nil - } - } - - ctx := context.Background() - err := NewParallelExecutor(3, executors...)(ctx) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - - assert.LessOrEqual(t, int(maxConcurrent), 3, "Should never exceed 3 concurrent executions") - assert.GreaterOrEqual(t, int(maxConcurrent), 1, "Should have at least 1 concurrent execution") - }) - - t.Run("MaxParallel=0 Uses1Worker", func(t *testing.T) { - var maxConcurrent int32 - var currentRunning atomic.Int32 - - executors := make([]Executor, 5) - for i := range 5 { - executors[i] = func(ctx context.Context) error { - current := currentRunning.Add(1) - - for { - maxValue := atomic.LoadInt32(&maxConcurrent) - if current <= maxValue || atomic.CompareAndSwapInt32(&maxConcurrent, maxValue, current) { - break - } - } - - time.Sleep(10 * time.Millisecond) - currentRunning.Add(-1) - return nil - } - } - - ctx := context.Background() - // When maxParallel is 0 or negative, it defaults to 1 - err := NewParallelExecutor(0, executors...)(ctx) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - - assert.Equal(t, int32(1), maxConcurrent, "Should use 1 worker when max-parallel is 0") - }) -} - -// TestMaxParallelWithErrors tests error handling with max-parallel -func TestMaxParallelWithErrors(t *testing.T) { - t.Run("OneTaskFailsOthersContinue", func(t *testing.T) { - var successCount int32 - - executors := make([]Executor, 5) - for i := range 5 { - taskID := i - executors[i] = func(ctx context.Context) error { - if taskID == 2 { - return assert.AnError - } - atomic.AddInt32(&successCount, 1) - return nil - } - } - - ctx := context.Background() - err := NewParallelExecutor(2, executors...)(ctx) - - // Should return the error from task 2 - assert.Error(t, err) //nolint:testifylint // pre-existing issue from nektos/act - - // Other tasks should still execute - assert.Equal(t, int32(4), successCount, "4 tasks should succeed") - }) - - t.Run("ContextCancellation", func(t *testing.T) { - ctx, cancel := context.WithCancel(context.Background()) - - var startedCount int32 - executors := make([]Executor, 10) - for i := range 10 { - executors[i] = func(ctx context.Context) error { - atomic.AddInt32(&startedCount, 1) - time.Sleep(100 * time.Millisecond) - return nil - } - } - - // Cancel after a short delay - go func() { - time.Sleep(30 * time.Millisecond) - cancel() - }() - - err := NewParallelExecutor(3, executors...)(ctx) - assert.Error(t, err) //nolint:testifylint // pre-existing issue from nektos/act - assert.ErrorIs(t, err, context.Canceled) //nolint:testifylint // pre-existing issue from nektos/act - - // Not all tasks should start due to cancellation (but timing may vary) - // Just verify cancellation occurred - t.Logf("Started %d tasks before cancellation", startedCount) - }) -} - -// TestMaxParallelResourceSharing tests resource sharing scenarios -func TestMaxParallelResourceSharing(t *testing.T) { - t.Run("SharedResourceWithMutex", func(t *testing.T) { - var sharedCounter int - var mu sync.Mutex - - executors := make([]Executor, 100) - for i := range 100 { - executors[i] = func(ctx context.Context) error { - mu.Lock() - sharedCounter++ - mu.Unlock() - return nil - } - } - - ctx := context.Background() - err := NewParallelExecutor(10, executors...)(ctx) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - - assert.Equal(t, 100, sharedCounter, "All tasks should increment counter") - }) - - t.Run("ChannelCommunication", func(t *testing.T) { - resultChan := make(chan int, 50) - - executors := make([]Executor, 50) - for i := range 50 { - taskID := i - executors[i] = func(ctx context.Context) error { - resultChan <- taskID - return nil - } - } - - ctx := context.Background() - err := NewParallelExecutor(5, executors...)(ctx) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - - close(resultChan) - - results := make(map[int]bool) - for result := range resultChan { - results[result] = true - } - - assert.Len(t, results, 50, "All task IDs should be received") - }) -} diff --git a/act/common/executor_test.go b/act/common/executor_test.go index ed894d80..fd45684b 100644 --- a/act/common/executor_test.go +++ b/act/common/executor_test.go @@ -9,9 +9,9 @@ import ( "errors" "reflect" "strings" + "sync" "sync/atomic" "testing" - "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -82,44 +82,45 @@ func TestNewConditionalExecutor(t *testing.T) { assert.Equal(1, falseCount) } -func TestNewParallelExecutor(t *testing.T) { - assert := assert.New(t) +// concurrencyProbe returns an executor recording the peak number of concurrent copies. Copies +// block until wantActive are in flight so the peak is exact without sleeping, and later copies +// find the gate already open so the last one still finishes with no partner left. +func concurrencyProbe(wantActive int32) (exec Executor, count, maxActive *atomic.Int32) { + var counted, active, peak atomic.Int32 + var once sync.Once + reached := make(chan struct{}) - ctx := context.Background() - - var count, activeCount, maxCount atomic.Int32 - emptyWorkflow := NewPipelineExecutor(func(ctx context.Context) error { - count.Add(1) - - active := activeCount.Add(1) + return func(ctx context.Context) error { + counted.Add(1) + running := active.Add(1) for { - m := maxCount.Load() - if active <= m || maxCount.CompareAndSwap(m, active) { + seen := peak.Load() + if running <= seen || peak.CompareAndSwap(seen, running) { break } } - time.Sleep(2 * time.Second) - activeCount.Add(-1) - + if running >= wantActive { + once.Do(func() { close(reached) }) + } + <-reached + active.Add(-1) return nil - }) + }, &counted, &peak +} - err := NewParallelExecutor(2, emptyWorkflow, emptyWorkflow, emptyWorkflow)(ctx) +func TestNewParallelExecutor(t *testing.T) { + ctx := context.Background() - assert.Equal(int32(3), count.Load(), "should run all 3 executors") - assert.Equal(int32(2), maxCount.Load(), "should run at most 2 executors in parallel") - assert.NoError(err) //nolint:testifylint // pre-existing issue from nektos/act + exec, count, maxActive := concurrencyProbe(2) + require.NoError(t, NewParallelExecutor(2, exec, exec, exec)(ctx)) + assert.Equal(t, int32(3), count.Load(), "should run all 3 executors") + assert.Equal(t, int32(2), maxActive.Load(), "should run at most 2 executors in parallel") - // Reset to test running the executor with 0 parallelism - count.Store(0) - activeCount.Store(0) - maxCount.Store(0) - - errSingle := NewParallelExecutor(0, emptyWorkflow, emptyWorkflow, emptyWorkflow)(ctx) - - assert.Equal(int32(3), count.Load(), "should run all 3 executors") - assert.Equal(int32(1), maxCount.Load(), "should run at most 1 executors in parallel") - assert.NoError(errSingle) + // parallelism below 1 falls back to a single worker + exec, count, maxActive = concurrencyProbe(1) + require.NoError(t, NewParallelExecutor(0, exec, exec, exec)(ctx)) + assert.Equal(t, int32(3), count.Load(), "should run all 3 executors") + assert.Equal(t, int32(1), maxActive.Load(), "should run at most 1 executor in parallel") } func TestNewParallelExecutorEmpty(t *testing.T) { @@ -173,6 +174,23 @@ func TestNewParallelExecutorCanceled(t *testing.T) { assert.Error(errExpected, err) //nolint:testifylint // pre-existing issue from nektos/act } +func TestNewParallelExecutorRunsRemainingAfterFailure(t *testing.T) { + var successCount atomic.Int32 + executors := make([]Executor, 5) + for i := range executors { + executors[i] = func(ctx context.Context) error { + if i == 2 { + return errors.New("fake error") + } + successCount.Add(1) + return nil + } + } + + require.Error(t, NewParallelExecutor(2, executors...)(context.Background())) + assert.Equal(t, int32(4), successCount.Load(), "a failing executor must not stop the others") +} + func TestExecutorConditionalsAndFinally(t *testing.T) { ctx := context.Background() var calls []string diff --git a/act/runner/helpers_test.go b/act/runner/helpers_test.go index b773cb92..40434dfa 100644 --- a/act/runner/helpers_test.go +++ b/act/runner/helpers_test.go @@ -5,11 +5,9 @@ package runner import ( "context" - "net" "os/exec" "runtime" "testing" - "time" "gitea.com/gitea/runner/act/container" @@ -42,18 +40,6 @@ func requireDocker(t *testing.T) { } } -// requireNetwork skips the test unless github.com is reachable. A few tests exercise behaviour -// that inherently needs the network (force-pulling an image, resolving a remote short-sha ref); -// gating lets the rest of the suite run offline without these failing. -func requireNetwork(t *testing.T) { - t.Helper() - conn, err := net.DialTimeout("tcp", "github.com:443", 3*time.Second) - if err != nil { - t.Skipf("skipping: network unavailable: %v", err) - } - _ = conn.Close() -} - // requireHostTools skips the test unless every named executable is on PATH. Used by the // self-hosted (host environment) suite, which runs steps directly on the host. func requireHostTools(t *testing.T, tools ...string) { diff --git a/act/runner/job_executor_test.go b/act/runner/job_executor_test.go index aa723721..6247c6eb 100644 --- a/act/runner/job_executor_test.go +++ b/act/runner/job_executor_test.go @@ -33,6 +33,7 @@ import ( ) func TestJobExecutor(t *testing.T) { + t.Parallel() // Dryrun only checks syntax/planning; all cases resolve locally, so this runs offline. tables := []TestJobFileInfo{ {workdir, "uses-and-run-in-one-step", "push", "Invalid run/uses syntax for job:test step:Test", platforms, secrets}, @@ -46,6 +47,7 @@ func TestJobExecutor(t *testing.T) { ctx := common.WithDryrun(context.Background(), true) for _, table := range tables { t.Run(table.workflowPath, func(t *testing.T) { + t.Parallel() table.runTest(ctx, t, &Config{}) }) } diff --git a/act/runner/runner_max_parallel_test.go b/act/runner/runner_max_parallel_test.go deleted file mode 100644 index 76b17d5a..00000000 --- a/act/runner/runner_max_parallel_test.go +++ /dev/null @@ -1,109 +0,0 @@ -// Copyright 2026 The Gitea Authors. All rights reserved. -// SPDX-License-Identifier: MIT - -package runner - -import ( - "sync" - "testing" - "time" - - "github.com/stretchr/testify/assert" -) - -// TestMaxParallelConfig tests that MaxParallel config is properly set -func TestMaxParallelConfig(t *testing.T) { - t.Run("MaxParallel set to 2", func(t *testing.T) { - config := &Config{ - Workdir: "testdata", - MaxParallel: 2, - } - - runner, err := New(config) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - assert.NotNil(t, runner) - - // Verify config is properly stored - runnerImpl, ok := runner.(*runnerImpl) - assert.True(t, ok) - assert.Equal(t, 2, runnerImpl.config.MaxParallel) - }) - - t.Run("MaxParallel set to 0 (no limit)", func(t *testing.T) { - config := &Config{ - Workdir: "testdata", - MaxParallel: 0, - } - - runner, err := New(config) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - assert.NotNil(t, runner) - - runnerImpl, ok := runner.(*runnerImpl) - assert.True(t, ok) - assert.Equal(t, 0, runnerImpl.config.MaxParallel) - }) - - t.Run("MaxParallel not set (defaults to 0)", func(t *testing.T) { - config := &Config{ - Workdir: "testdata", - } - - runner, err := New(config) - assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act - assert.NotNil(t, runner) - - runnerImpl, ok := runner.(*runnerImpl) - assert.True(t, ok) - assert.Equal(t, 0, runnerImpl.config.MaxParallel) - }) -} - -// TestMaxParallelConcurrencyTracking tests that max-parallel actually limits concurrent execution -func TestMaxParallelConcurrencyTracking(t *testing.T) { - // This is a unit test for the parallel executor logic - // We test that when MaxParallel is set, it limits the number of workers - - var mu sync.Mutex - var maxConcurrent int - var currentConcurrent int - - // Create a function that tracks concurrent execution - trackingFunc := func() { - mu.Lock() - currentConcurrent++ - if currentConcurrent > maxConcurrent { - maxConcurrent = currentConcurrent - } - mu.Unlock() - - // Simulate work - time.Sleep(50 * time.Millisecond) - - mu.Lock() - currentConcurrent-- - mu.Unlock() - } - - // Run multiple tasks with limited parallelism - maxConcurrent = 0 - currentConcurrent = 0 - - // This simulates what NewParallelExecutor does with a semaphore - var wg sync.WaitGroup - semaphore := make(chan struct{}, 2) // Limit to 2 concurrent - - for range 6 { - wg.Go(func() { - semaphore <- struct{}{} // Acquire - defer func() { <-semaphore }() // Release - trackingFunc() - }) - } - - wg.Wait() - - // With a semaphore of 2, max concurrent should be <= 2 - assert.LessOrEqual(t, maxConcurrent, 2, "Maximum concurrent executions should not exceed limit") - assert.GreaterOrEqual(t, maxConcurrent, 1, "Should have at least 1 concurrent execution") -} diff --git a/act/runner/runner_test.go b/act/runner/runner_test.go index 61244f4b..922c0f02 100644 --- a/act/runner/runner_test.go +++ b/act/runner/runner_test.go @@ -13,6 +13,7 @@ import ( "path" "path/filepath" "runtime" + "slices" "strings" "testing" "time" @@ -163,6 +164,12 @@ func TestGraphEvent(t *testing.T) { assert.Empty(t, plan.Stages) } +// these two build the same action Dockerfiles into one image tag, so they cannot overlap +var sharedImageWorkflows = []string{"local-action-dockerfile", "local-action-via-composite-dockerfile"} + +// bounds concurrent plans: each job holds a network, and the daemon's address pool is finite +var planSlots = make(chan struct{}, 4) + type TestJobFileInfo struct { workdir string workflowPath string @@ -182,12 +189,19 @@ func (j *TestJobFileInfo) runTest(ctx context.Context, t *testing.T, cfg *Config fullWorkflowPath := filepath.Join(workdir, j.workflowPath) runnerConfig := &Config{ - Workdir: workdir, - BindWorkdir: false, - EventName: j.eventName, - EventPath: cfg.EventPath, - Platforms: j.platforms, - ReuseContainers: false, + Workdir: workdir, + BindWorkdir: false, + EventName: j.eventName, + EventPath: cfg.EventPath, + Platforms: j.platforms, + // fixtures reuse workflow and job names, so parallel tests would collide without this + ContainerNamePrefix: strings.ReplaceAll(t.Name(), "/", "-"), + ReuseContainers: false, + // as the shipped runner does, else a fixture asserting a job failure keeps its + // container, and its network, on the daemon forever + AutoRemove: true, + // 0 would run jobs runtime.NumCPU()-wide, making the network peak machine-dependent + MaxParallel: 2, ForceRebuild: true, Env: cfg.Env, Secrets: cfg.Secrets, @@ -210,7 +224,11 @@ func (j *TestJobFileInfo) runTest(ctx context.Context, t *testing.T, cfg *Config plan, err := planner.PlanEvent(j.eventName) assert.True(t, (err == nil) != (plan == nil), "PlanEvent should return either a plan or an error") //nolint:testifylint // pre-existing issue from nektos/act if err == nil && plan != nil { - err = runner.NewPlanExecutor(plan)(ctx) + err = func() error { + planSlots <- struct{}{} + defer func() { <-planSlots }() + return runner.NewPlanExecutor(plan)(ctx) + }() if j.errorMessage == "" { assert.NoError(t, err, fullWorkflowPath) //nolint:testifylint // pre-existing issue from nektos/act } else { @@ -227,6 +245,7 @@ type TestConfig struct { func TestRunEvent(t *testing.T) { requireDocker(t) + t.Parallel() ctx := context.Background() @@ -315,6 +334,9 @@ func TestRunEvent(t *testing.T) { // host /proc bind mounts are Linux-Docker-only requireLinuxDocker(t) } + if !slices.Contains(sharedImageWorkflows, table.workflowPath) { + t.Parallel() + } config := &Config{ Secrets: table.secrets, @@ -445,6 +467,7 @@ func TestRunEventHostEnvironment(t *testing.T) { } func TestDryrunEvent(t *testing.T) { + t.Parallel() // Dryrun plans without containers or network (shells and local actions only). ctx := common.WithDryrun(context.Background(), true) @@ -464,6 +487,7 @@ func TestDryrunEvent(t *testing.T) { for _, table := range tables { t.Run(table.workflowPath, func(t *testing.T) { + t.Parallel() table.runTest(ctx, t, &Config{}) }) } @@ -474,33 +498,11 @@ func TestDryrunEvent(t *testing.T) { // workflow's outputs via `needs`). func TestReusableWorkflowCaller(t *testing.T) { requireDocker(t) + t.Parallel() table := TestJobFileInfo{workdir, "uses-workflow", "push", "", platforms, map[string]string{"secret": "keep_it_private"}} table.runTest(context.Background(), t, &Config{Secrets: table.secrets}) } -func TestDockerActionForcePullForceRebuild(t *testing.T) { - requireDocker(t) - requireNetwork(t) // force-pulls a docker action image - - ctx := context.Background() - - config := &Config{ - ForcePull: true, - ForceRebuild: true, - } - - tables := []TestJobFileInfo{ - {workdir, "local-action-dockerfile", "push", "", platforms, secrets}, - {workdir, "local-action-via-composite-dockerfile", "push", "", platforms, secrets}, - } - - for _, table := range tables { - t.Run(table.workflowPath, func(t *testing.T) { - table.runTest(ctx, t, config) - }) - } -} - type maskJobLoggerFactory struct { Output bytes.Buffer } @@ -513,6 +515,7 @@ func (f *maskJobLoggerFactory) WithJobLogger() *log.Logger { } func TestMaskValues(t *testing.T) { + t.Parallel() assertNoSecret := func(text, secret string) { //nolint:unparam // pre-existing issue from nektos/act found := strings.Contains(text, "composite secret") if found { @@ -543,6 +546,7 @@ func TestMaskValues(t *testing.T) { func TestRunEventSecrets(t *testing.T) { requireDocker(t) + t.Parallel() workflowPath := "secrets" tjfi := TestJobFileInfo{ @@ -598,6 +602,7 @@ func TestRunWithService(t *testing.T) { } func TestRunActionInputs(t *testing.T) { + t.Parallel() requireDocker(t) workflowPath := "input-from-cli" @@ -617,6 +622,7 @@ func TestRunActionInputs(t *testing.T) { } func TestRunEventPullRequest(t *testing.T) { + t.Parallel() requireDocker(t) workflowPath := "pull-request" @@ -633,6 +639,7 @@ func TestRunEventPullRequest(t *testing.T) { } func TestRunMatrixWithUserDefinedInclusions(t *testing.T) { + t.Parallel() requireDocker(t) workflowPath := "matrix-with-user-inclusions"