fix: Integration testing leaked processes on DualCluster

tests/integration/commontestutils/skrcontextimpl/dual_cluster.go

@@ -10,27 +10,34 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
+	"fmt"
 	"github.com/kyma-project/lifecycle-manager/internal/event"
 	"github.com/kyma-project/lifecycle-manager/internal/remote"
 )
 
 var (
 	ErrEmptyRestConfig  = errors.New("rest.Config is nil")
-	errSkrEnvNotStarted = errors.New("SKR envtest environment not started")
+	ErrSkrEnvNotStarted = errors.New("SKR envtest environment not started")
 )
 
+type Stopper interface {
+	Stop() error
+}
+
 type DualClusterFactory struct {
 	clients sync.Map
 	scheme  *machineryruntime.Scheme
 	event   event.Event
 	skrEnv  *envtest.Environment
+	skrEnvs sync.Map
 }
 
 func NewDualClusterFactory(scheme *machineryruntime.Scheme, event event.Event) *DualClusterFactory {
 	return &DualClusterFactory{
 		clients: sync.Map{},
 		scheme:  scheme,
 		event:   event,
+		skrEnvs: sync.Map{},
 	}
 }
 
@@ -40,46 +47,63 @@ func (f *DualClusterFactory) Init(_ context.Context, kyma types.NamespacedName)
 		return nil
 	}
 
-	f.skrEnv = &envtest.Environment{
+	skrEnv := &envtest.Environment{
 		ErrorIfCRDPathMissing: true,
 		// Scheme: scheme,
 	}
-	cfg, err := f.GetSkrEnv().Start()
+
+	// Start the envtest and record the returned cfg
+	cfg, err := skrEnv.Start()
 	if err != nil {
 		return err
 	}
 	if cfg == nil {
+		// cleanup fast - if start returned nil cfg
+		_ = skrEnv.Stop()
 		return ErrEmptyRestConfig
 	}
 
 	var authUser *envtest.AuthenticatedUser
-	authUser, err = f.GetSkrEnv().AddUser(envtest.User{
+	authUser, err = skrEnv.AddUser(envtest.User{
 		Name:   "skr-admin-account",
 		Groups: []string{"system:masters"},
 	}, cfg)
 	if err != nil {
+		_ = skrEnv.Stop()
 		return err
 	}
 
 	skrClient, err := client.New(authUser.Config(), client.Options{Scheme: f.scheme})
+	if err != nil {
+		_ = skrEnv.Stop()
+		return err
+	}
 	newClient := remote.NewClientWithConfig(skrClient, authUser.Config())
 	f.clients.Store(kyma.Name, newClient)
 
+	f.skrEnv = skrEnv
+	// track this envtest so Stop() can stop all started envs
+	f.skrEnvs.Store(kyma.Name, skrEnv)
+
 	return err
 }
 
 func (f *DualClusterFactory) Get(kyma types.NamespacedName) (*remote.SkrContext, error) {
 	value, ok := f.clients.Load(kyma.Name)
 	if !ok {
-		return nil, errSkrEnvNotStarted
+		return nil, ErrSkrEnvNotStarted
 	}
 	skrClient, ok := value.(*remote.ConfigAndClient)
 	if !ok {
-		return nil, errSkrEnvNotStarted
+		return nil, ErrSkrEnvNotStarted
 	}
 	return remote.NewSkrContext(skrClient, f.event), nil
 }
 
+func (f *DualClusterFactory) StoreEnv(name string, env interface{}) {
+	f.skrEnvs.Store(name, env)
+}
+
 func (f *DualClusterFactory) InvalidateCache(_ types.NamespacedName) {
 	// no-op
 }
@@ -89,8 +113,33 @@ func (f *DualClusterFactory) GetSkrEnv() *envtest.Environment {
 }
 
 func (f *DualClusterFactory) Stop() error {
-	if f.skrEnv == nil {
+	var errs []error
+
+	f.skrEnvs.Range(func(key, value any) bool {
+		name, _ := key.(string)
+		if stopper, ok := value.(Stopper); ok && stopper != nil {
+			if err := stopper.Stop(); err != nil {
+				if name != "" {
+					errs = append(errs, fmt.Errorf("stop %s: %w", name, err))
+				} else {
+					errs = append(errs, fmt.Errorf("stop <unknown>: %w", err))
+				}
+			}
+		}
+
+		// remove entries so we don't double-stop later
+		f.skrEnvs.Delete(key)
+		if name != "" {
+			f.clients.Delete(name)
+		}
+		return true
+	})
+
+	// Clear skrEnv
+	f.skrEnv = nil
+
+	if len(errs) == 0 {
 		return nil
 	}
-	return f.skrEnv.Stop()
+	return fmt.Errorf("errors stopping envtests: %w", errors.Join(errs...))
 }

tests/integration/commontestutils/skrcontextimpl/dual_cluster_test.go

@@ -0,0 +1,190 @@
+package skrcontextimpl_test
+
+import (
+	"context"
+	"errors"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	machineryruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+
+	testskrcontext "github.com/kyma-project/lifecycle-manager/tests/integration/commontestutils/skrcontextimpl"
+)
+
+func TestDualClusterFactory(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "DualCluster Factory Suite")
+}
+
+func newFactory() *testskrcontext.DualClusterFactory {
+	scheme := machineryruntime.NewScheme()
+	return testskrcontext.NewDualClusterFactory(scheme, nil)
+}
+
+func Test_GetBeforeInit(t *testing.T) {
+	dualFactory := newFactory()
+
+	_, err := dualFactory.Get(types.NamespacedName{Name: "kymaUninitialized"})
+
+	require.Error(t, err)
+	assert.ErrorIs(t, err, testskrcontext.ErrSkrEnvNotStarted)
+}
+
+func Test_StopWithErrors(t *testing.T) {
+	dualFactory := newFactory()
+	envPrimary := &fakeEnv{name: "primary-env", stopErr: errors.New("primary stop failure")}
+	envSecondary := &fakeEnv{name: "secondary-env", stopErr: errors.New("secondary stop failure")}
+	envTertiary := &fakeEnv{name: "tertiary-env"}
+	dualFactory.StoreEnv("primary-env", envPrimary)
+	dualFactory.StoreEnv("secondary-env", envSecondary)
+	dualFactory.StoreEnv("tertiary-env", envTertiary)
+
+	err := dualFactory.Stop()
+
+	require.Error(t, err)
+	msg := err.Error()
+	assert.Contains(t, msg, "primary stop failure")
+	assert.Contains(t, msg, "secondary stop failure")
+	assert.Nil(t, dualFactory.GetSkrEnv())
+	assert.True(t, envPrimary.stopCalled)
+	assert.True(t, envSecondary.stopCalled)
+	assert.True(t, envTertiary.stopCalled)
+}
+
+func Test_StopIdempotent(t *testing.T) {
+	dualFactory := newFactory()
+	fakeEnv := &fakeEnv{name: "test-env"}
+	dualFactory.StoreEnv("test-env", fakeEnv)
+
+	require.NoError(t, dualFactory.Stop())
+
+	assert.True(t, fakeEnv.stopCalled)
+}
+
+func Test_StopClearsAllEntries(t *testing.T) {
+	dualFactory := newFactory()
+	for range make([]struct{}, 5) {
+		fakeEnv := &fakeEnv{name: "test-env"}
+		dualFactory.StoreEnv("test-env", fakeEnv)
+	}
+
+	require.NoError(t, dualFactory.Stop())
+
+	assert.Nil(t, dualFactory.GetSkrEnv())
+	_, err := dualFactory.Get(types.NamespacedName{Name: "test-env"})
+	assert.Error(t, err)
+}
+
+func Test_ConcurrentStopCalls(t *testing.T) {
+	dualFactory := newFactory()
+	for range make([]struct{}, 10) {
+		fakeEnv := &fakeEnv{name: "test-env"}
+		dualFactory.StoreEnv("test-env", fakeEnv)
+	}
+	var waitGroup sync.WaitGroup
+	errors := make(chan error, 5)
+
+	for range make([]struct{}, 5) {
+		waitGroup.Add(1)
+		go func() {
+			defer waitGroup.Done()
+			errors <- dualFactory.Stop()
+		}()
+	}
+	waitGroup.Wait()
+	close(errors)
+
+	for err := range errors {
+		assert.NoError(t, err)
+	}
+}
+
+func Test_LeakPrevention_VerifyNoGoroutineLeaks(t *testing.T) {
+	before := runtime.NumGoroutine()
+	for range make([]struct{}, 3) {
+		dualFactory := newFactory()
+		for range make([]struct{}, 5) {
+			fakeEnv := &leakyFakeEnv{
+				name:                 "test-env",
+				shouldSpawnGoroutine: true,
+			}
+			dualFactory.StoreEnv("test-env", fakeEnv)
+		}
+		require.NoError(t, dualFactory.Stop())
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	// Force garbage collection to clean up any lingering references
+	runtime.GC()
+	runtime.GC()
+	time.Sleep(50 * time.Millisecond)
+
+	after := runtime.NumGoroutine()
+	assert.LessOrEqual(t, after, before+2,
+		"Expected no significant goroutine leaks. Before: %d, After: %d", before, after)
+}
+
+func Test_LeakPrevention_VerifyStopperInterfaceHandling(t *testing.T) {
+	dualFactory := newFactory()
+	normalStopper := &fakeEnv{name: "normal"}
+	errorStopper := &fakeEnv{name: "error", stopErr: errors.New("stop error")}
+	leakyStopper := &leakyFakeEnv{name: "leaky", shouldSpawnGoroutine: true}
+	dualFactory.StoreEnv("normal", normalStopper)
+	dualFactory.StoreEnv("error", errorStopper)
+	dualFactory.StoreEnv("leaky", leakyStopper)
+	dualFactory.StoreEnv("non-stopper", "this is just a string")
+
+	err := dualFactory.Stop()
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "stop error")
+	assert.True(t, normalStopper.stopCalled)
+	assert.True(t, errorStopper.stopCalled)
+	assert.True(t, leakyStopper.stopCalled)
+}
+
+type fakeEnv struct {
+	name       string
+	stopCalled bool
+	stopErr    error
+}
+
+func (f *fakeEnv) Stop() error {
+	f.stopCalled = true
+	return f.stopErr
+}
+
+type leakyFakeEnv struct {
+	name                 string
+	stopCalled           bool
+	stopErr              error
+	shouldSpawnGoroutine bool
+	cancel               context.CancelFunc
+}
+
+func (l *leakyFakeEnv) Stop() error {
+	l.stopCalled = true
+
+	if l.shouldSpawnGoroutine && l.cancel == nil {
+		// Simulate starting a background goroutine (like envtest)
+		ctx, cancel := context.WithCancel(context.Background())
+		l.cancel = cancel
+
+		go func() {
+			<-ctx.Done() // Wait for cancellation
+		}()
+	}
+
+	// Clean up the goroutine
+	if l.cancel != nil {
+		l.cancel()
+	}
+
+	return l.stopErr
+}