[8.15](backport #5409) [Integration Test Framework] fix createTempDir and flaky tests

internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go

@@ -6,6 +6,8 @@
 
 import (
 	"context"
+	"errors"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -26,17 +28,30 @@
 )
 
 type mockUpgradeManager struct {
+<<<<<<< HEAD
 	msgChan chan string
+=======
+	UpgradeFn func(
+		ctx context.Context,
+		version string,
+		sourceURI string,
+		action *fleetapi.ActionUpgrade,
+		details *details.Details,
+		skipVerifyOverride bool,
+		skipDefaultPgp bool,
+		pgpBytes ...string) (reexec.ShutdownCallbackFn, error)
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }
 
 func (u *mockUpgradeManager) Upgradeable() bool {
 	return true
 }
 
 func (u *mockUpgradeManager) Reload(rawConfig *config.Config) error {
 	return nil
 }
 
+<<<<<<< HEAD
 func (u *mockUpgradeManager) Upgrade(ctx context.Context, version string, sourceURI string, action *fleetapi.ActionUpgrade, details *details.Details, skipVerifyOverride bool, skipDefaultPgp bool, pgpBytes ...string) (_ reexec.ShutdownCallbackFn, err error) {
 	select {
 	case <-time.After(2 * time.Second):
@@ -46,6 +61,27 @@
 		u.msgChan <- "canceled " + version
 		return nil, ctx.Err()
 	}
+=======
+func (u *mockUpgradeManager) Upgrade(
+	ctx context.Context,
+	version string,
+	sourceURI string,
+	action *fleetapi.ActionUpgrade,
+	details *details.Details,
+	skipVerifyOverride bool,
+	skipDefaultPgp bool,
+	pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+	return u.UpgradeFn(
+		ctx,
+		version,
+		sourceURI,
+		action,
+		details,
+		skipVerifyOverride,
+		skipDefaultPgp,
+		pgpBytes...)
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }
 
 func (u *mockUpgradeManager) Ack(ctx context.Context, acker acker.Acker) error {
@@ -65,7 +101,11 @@
 	log, _ := logger.New("", false)
 
 	agentInfo := &info.AgentInfo{}
+<<<<<<< HEAD
 	msgChan := make(chan string)
+=======
+	upgradeCalledChan := make(chan struct{})
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 
 	// Create and start the coordinator
 	c := coordinator.New(
@@ -75,7 +115,25 @@
 		agentInfo,
 		component.RuntimeSpecs{},
 		nil,
+<<<<<<< HEAD
 		&mockUpgradeManager{msgChan: msgChan},
+=======
+		&mockUpgradeManager{
+			UpgradeFn: func(
+				ctx context.Context,
+				version string,
+				sourceURI string,
+				action *fleetapi.ActionUpgrade,
+				details *details.Details,
+				skipVerifyOverride bool,
+				skipDefaultPgp bool,
+				pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+				upgradeCalledChan <- struct{}{}
+				return nil, nil
+			},
+		},
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 		nil, nil, nil, nil, nil, false)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
@@ -86,8 +144,13 @@
 	ack := noopacker.New()
 	err := u.Handle(ctx, &a, ack)
 	require.NoError(t, err)
-	msg := <-msgChan
-	require.Equal(t, "completed 8.3.0", msg)
+
+	// Make sure this test does not dead lock or wait for too long
+	select {
+	case <-time.Tick(50 * time.Millisecond):
+		t.Fatal("mockUpgradeManager.Upgrade was not called")
+	case <-upgradeCalledChan:
+	}
 }
 
 func TestUpgradeHandlerSameVersion(t *testing.T) {
@@ -99,17 +162,45 @@
 	log, _ := logger.New("", false)
 
 	agentInfo := &info.AgentInfo{}
+<<<<<<< HEAD
 	msgChan := make(chan string)
+=======
+	upgradeCalledChan := make(chan struct{})
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 
 	// Create and start the Coordinator
+	upgradeCalled := atomic.Bool{}
 	c := coordinator.New(
 		log,
 		configuration.DefaultConfiguration(),
 		logger.DefaultLogLevel,
 		agentInfo,
 		component.RuntimeSpecs{},
 		nil,
+<<<<<<< HEAD
 		&mockUpgradeManager{msgChan: msgChan},
+=======
+		&mockUpgradeManager{
+			UpgradeFn: func(
+				ctx context.Context,
+				version string,
+				sourceURI string,
+				action *fleetapi.ActionUpgrade,
+				details *details.Details,
+				skipVerifyOverride bool,
+				skipDefaultPgp bool,
+				pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+				if upgradeCalled.CompareAndSwap(false, true) {
+					upgradeCalledChan <- struct{}{}
+					return nil, nil
+				}
+				err := errors.New("mockUpgradeManager.Upgrade called more than once")
+				t.Error(err.Error())
+				return nil, err
+			},
+		},
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 		nil, nil, nil, nil, nil, false)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
@@ -122,8 +213,18 @@
 	err2 := u.Handle(ctx, &a, ack)
 	require.NoError(t, err1)
 	require.NoError(t, err2)
+<<<<<<< HEAD
 	msg := <-msgChan
 	require.Equal(t, "completed 8.3.0", msg)
+=======
+
+	// Make sure this test does not dead lock or wait for too long
+	select {
+	case <-time.Tick(50 * time.Millisecond):
+		t.Fatal("mockUpgradeManager.Upgrade was not called")
+	case <-upgradeCalledChan:
+	}
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }
 
 func TestUpgradeHandlerNewVersion(t *testing.T) {
@@ -133,9 +234,13 @@
 	defer cancel()
 
 	log, _ := logger.New("", false)
+	upgradeCalledChan := make(chan string)
 
 	agentInfo := &info.AgentInfo{}
+<<<<<<< HEAD
 	msgChan := make(chan string)
+=======
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 
 	// Create and start the Coordinator
 	c := coordinator.New(
@@ -145,7 +250,31 @@
 		agentInfo,
 		component.RuntimeSpecs{},
 		nil,
+<<<<<<< HEAD
 		&mockUpgradeManager{msgChan: msgChan},
+=======
+		&mockUpgradeManager{
+			UpgradeFn: func(
+				ctx context.Context,
+				version string,
+				sourceURI string,
+				action *fleetapi.ActionUpgrade,
+				details *details.Details,
+				skipVerifyOverride bool,
+				skipDefaultPgp bool,
+				pgpBytes ...string) (reexec.ShutdownCallbackFn, error) {
+
+				defer func() {
+					upgradeCalledChan <- version
+				}()
+				if version == "8.2.0" {
+					return nil, errors.New("upgrade to 8.2.0 will always fail")
+				}
+
+				return nil, nil
+			},
+		},
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 		nil, nil, nil, nil, nil, false)
 	//nolint:errcheck // We don't need the termination state of the Coordinator
 	go c.Run(ctx)
@@ -156,13 +285,35 @@
 	a2 := fleetapi.ActionUpgrade{Data: fleetapi.ActionUpgradeData{
 		Version: "8.5.0", SourceURI: "http://localhost"}}
 	ack := noopacker.New()
+
+	checkMsg := func(c <-chan string, expected, errMsg string) {
+		t.Helper()
+		// Make sure this test does not dead lock or wait for too long
+		// For some reason < 1s sometimes makes the test fail.
+		select {
+		case <-time.Tick(1300 * time.Millisecond):
+			t.Fatal("timed out waiting for Upgrade to return")
+		case msg := <-c:
+			require.Equal(t, expected, msg, errMsg)
+		}
+	}
+
+	// Send both upgrade actions, a1 will error before a2 succeeds
 	err1 := u.Handle(ctx, &a1, ack)
 	require.NoError(t, err1)
+<<<<<<< HEAD
 	time.Sleep(1 * time.Second)
 	err2 := u.Handle(ctx, &a2, ack)
 	require.NoError(t, err2)
 	msg1 := <-msgChan
 	require.Equal(t, "canceled 8.2.0", msg1)
 	msg2 := <-msgChan
 	require.Equal(t, "completed 8.5.0", msg2)
+=======
+	checkMsg(upgradeCalledChan, "8.2.0", "first call must be with version 8.2.0")
+
+	err2 := u.Handle(ctx, &a2, ack)
+	require.NoError(t, err2)
+	checkMsg(upgradeCalledChan, "8.5.0", "second call to Upgrade must be with version 8.5.0")
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 }

pkg/testing/fixture.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/details"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/install"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/control"
 	"github.com/elastic/elastic-agent/pkg/control/v2/client"
@@ -1196,6 +1197,34 @@ func performConfigure(ctx context.Context, c client.Client, cfg string, timeout
 	return nil
 }
 
+<<<<<<< HEAD
+=======
+// createTempDir creates a temporary directory that will be
+// removed after the tests passes. If the test fails, the
+// directory is kept for further investigation.
+//
+// If the test is run with -v and fails the temporary directory is logged
+func createTempDir(t *testing.T) string {
+	tempDir, err := os.MkdirTemp("", strings.ReplaceAll(t.Name(), "/", "-"))
+	if err != nil {
+		t.Fatalf("failed to make temp directory: %s", err)
+	}
+
+	cleanup := func() {
+		if !t.Failed() {
+			if err := install.RemovePath(tempDir); err != nil {
+				t.Errorf("could not remove temp dir '%s': %s", tempDir, err)
+			}
+		} else {
+			t.Logf("Temporary directory %q preserved for investigation/debugging", tempDir)
+		}
+	}
+	t.Cleanup(cleanup)
+
+	return tempDir
+}
+
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 type AgentStatusOutput struct {
 	Info struct {
 		ID           string `json:"id"`

testing/integration/event_logging_test.go

@@ -75,7 +75,10 @@ func TestEventLogFile(t *testing.T) {
 		Local: true,
 		Sudo:  false,
 	})
+<<<<<<< HEAD
 
+=======
+>>>>>>> 1242e7186a ([Integration Test Framework] fix createTempDir and flaky tests (#5409))
 	ctx, cancel := testcontext.WithDeadline(
 		t,
 		context.Background(),