diff --git a/internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go b/internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go index e331e477253..7cdedc57017 100644 --- a/internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go +++ b/internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go @@ -6,6 +6,8 @@ package handlers import ( "context" + "errors" + "sync/atomic" "testing" "time" @@ -26,7 +28,19 @@ import ( ) 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 { @@ -37,6 +51,7 @@ 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 @@ func (u *mockUpgradeManager) Upgrade(ctx context.Context, version string, source 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 @@ func TestUpgradeHandler(t *testing.T) { 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 @@ func TestUpgradeHandler(t *testing.T) { 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 @@ func TestUpgradeHandler(t *testing.T) { 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,9 +162,14 @@ func TestUpgradeHandlerSameVersion(t *testing.T) { 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(), @@ -109,7 +177,30 @@ func TestUpgradeHandlerSameVersion(t *testing.T) { 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 @@ func TestUpgradeHandlerSameVersion(t *testing.T) { 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 @@ func TestUpgradeHandlerNewVersion(t *testing.T) { 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 @@ func TestUpgradeHandlerNewVersion(t *testing.T) { 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,8 +285,23 @@ func TestUpgradeHandlerNewVersion(t *testing.T) { 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) @@ -165,4 +309,11 @@ func TestUpgradeHandlerNewVersion(t *testing.T) { 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)) } diff --git a/pkg/testing/fixture.go b/pkg/testing/fixture.go index c7338cc63d1..98a0236f4c4 100644 --- a/pkg/testing/fixture.go +++ b/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"` diff --git a/testing/integration/event_logging_test.go b/testing/integration/event_logging_test.go index 4e799e0cecb..b7478837e92 100644 --- a/testing/integration/event_logging_test.go +++ b/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(),