Refactor snapshot retrieval to use ContainerOrchestratorUtility

Simplify the getSnapshotsForPVC function by delegating snapshot retrieval
to the ContainerOrchestratorUtility interface instead of directly using
the snapshot client. This improves testability and maintains consistency
with the existing architecture.

Changes:
- Remove direct dependency on external-snapshotter client in util.go
- Update getSnapshotsForPVC to use ContainerOrchestratorUtility.GetSnapshotsForPVC
- Remove rest.Config parameter from getSnapshotsForPVC function signature
- Add comprehensive unit tests for getSnapshotsForPVC covering:
  * Uninitialized ContainerOrchestratorUtility scenario
  * PVC with no snapshots
  * PVC with multiple snapshots
- Implement GetSnapshotsForPVC in FakeK8SOrchestrator for testing
- Add TODO comment for future refactoring of pvcToSnapshotsMap key type

Benefits:
- Improved testability through dependency injection
- Reduced coupling to external snapshot client
- Consistent error handling
- Better alignment with existing orchestrator abstraction pattern

Author: kolluria

pkg/common/unittestcommon/utils.go

@@ -517,8 +517,11 @@ func (c *FakeK8SOrchestrator) SetCSINodeTopologyInstances(instances []interface{
 }
 
 func (c *FakeK8SOrchestrator) GetSnapshotsForPVC(ctx context.Context, pvcName, namespace string) []string {
-	//TODO implement me
-	panic("implement me")
+	if strings.Contains(pvcName, "no-snap") {
+		return []string{}
+	}
+
+	return []string{"snap1", "snap2", "snap3"}
 }
 
 // configFromVCSim starts a vcsim instance and returns config for use against the

pkg/csi/service/common/commonco/k8sorchestrator/k8sorchestrator.go

@@ -251,6 +251,7 @@ func (n namespacedName) String() string {
 // pvcToSnapshotsMap maps a PVC to its snapshots.
 // Key is the namespaced name of the PVC and value is a map.
 // The key of the inner map is the namespaced name of the snapshot.
+// TODO: change the key of the inner map to string
 type pvcToSnapshotsMap struct {
 	*sync.RWMutex
 	items map[namespacedName]map[namespacedName]struct{}

pkg/syncer/cnsoperator/controller/cnsunregistervolume/util.go

@@ -21,7 +21,6 @@ import (
 	"errors"
 	"strings"
 
-	snapshotclient "github.com/kubernetes-csi/external-snapshotter/client/v8/clientset/versioned"
 	vmoperatortypes "github.com/vmware-tanzu/vm-operator/api/v1alpha2"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -111,7 +110,7 @@ func _getVolumeUsageInfo(ctx context.Context, pvcName string, pvcNamespace strin
 		return &volumeUsageInfo, nil
 	}
 
-	volumeUsageInfo.snapshots, volumeUsageInfo.isInUse, err = getSnapshotsForPVC(ctx, pvcName, pvcNamespace, *cfg)
+	volumeUsageInfo.snapshots, volumeUsageInfo.isInUse, err = getSnapshotsForPVC(ctx, pvcName, pvcNamespace)
 	if err != nil {
 		return nil, err
 	}
@@ -222,35 +221,16 @@ func getPodsForPVC(ctx context.Context, pvcName string, pvcNamespace string,
 }
 
 // getSnapshotsForPVC returns a list of snapshots that are created for the specified PVC.
-func getSnapshotsForPVC(ctx context.Context, pvcName string, pvcNamespace string,
-	cfg rest.Config) ([]string, bool, error) {
+func getSnapshotsForPVC(ctx context.Context, pvcName string, pvcNamespace string) ([]string, bool, error) {
 	log := logger.GetLogger(ctx)
-	c, err := snapshotclient.NewForConfig(&cfg)
-	if err != nil {
-		log.Warnf("Failed to create snapshot client for PVC %q in namespace %q. Error: %q",
-			pvcName, pvcNamespace, err.Error())
-		return nil, false, errors.New("failed to create snapshot client")
-	}
-
-	// TODO: check if we can use informer cache
-	list, err := c.SnapshotV1().VolumeSnapshots(pvcNamespace).List(ctx, metav1.ListOptions{})
-	if err != nil {
-		log.Warnf("Failed to list VolumeSnapshots in namespace %q for PVC %q. Error: %q",
-			pvcNamespace, pvcName, err.Error())
-		return nil, false, errors.New("failed to list VolumeSnapshots")
-	}
-
-	var snapshots []string
-	for _, snap := range list.Items {
-		if snap.Spec.Source.PersistentVolumeClaimName == nil ||
-			*snap.Spec.Source.PersistentVolumeClaimName != pvcName {
-			continue
-		}
-
-		snapshots = append(snapshots, snap.Name)
+	if commonco.ContainerOrchestratorUtility == nil {
+		err := errors.New("ContainerOrchestratorUtility is not initialized")
+		log.Warn(err)
+		return nil, false, err
 	}
 
-	return snapshots, len(snapshots) > 0, nil
+	snaps := commonco.ContainerOrchestratorUtility.GetSnapshotsForPVC(ctx, pvcName, pvcNamespace)
+	return snaps, len(snaps) > 0, nil
 }
 
 // getGuestClustersForPVC returns a list of guest clusters that are using the specified PVC.

pkg/syncer/cnsoperator/controller/cnsunregistervolume/util_test.go

@@ -0,0 +1,57 @@
+package cnsunregistervolume
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/common/unittestcommon"
+	"sigs.k8s.io/vsphere-csi-driver/v3/pkg/csi/service/common/commonco"
+)
+
+func TestGetSnapshotsForPVC(t *testing.T) {
+	containerOrchestratorUtilityOriginal := commonco.ContainerOrchestratorUtility
+	defer func() {
+		commonco.ContainerOrchestratorUtility = containerOrchestratorUtilityOriginal
+	}()
+
+	t.Run("WhenContainerOrchestratorIsNotInitialised", func(tt *testing.T) {
+		// Setup
+		commonco.ContainerOrchestratorUtility = nil
+		expErr := errors.New("ContainerOrchestratorUtility is not initialized")
+
+		// Execute
+		_, _, err := getSnapshotsForPVC(context.Background(), "", "")
+
+		// Assert
+		assert.Equal(tt, expErr, err)
+	})
+
+	t.Run("WhenPVCHasNoSnapshots", func(tt *testing.T) {
+		// Setup
+		commonco.ContainerOrchestratorUtility = &unittestcommon.FakeK8SOrchestrator{}
+
+		// Execute
+		snaps, inUse, err := getSnapshotsForPVC(context.Background(), "no-snaps", "")
+
+		// Assert
+		assert.NoError(tt, err)
+		assert.False(tt, inUse)
+		assert.Empty(tt, snaps)
+	})
+
+	t.Run("WhenPVCHasSnapshots", func(tt *testing.T) {
+		// Setup
+		commonco.ContainerOrchestratorUtility = &unittestcommon.FakeK8SOrchestrator{}
+		expSnaps := []string{"snap1", "snap2", "snap3"}
+
+		// Execute
+		snaps, inUse, err := getSnapshotsForPVC(context.Background(), "with-snaps", "")
+
+		// Assert
+		assert.NoError(tt, err)
+		assert.True(tt, inUse)
+		assert.Equal(tt, expSnaps, snaps)
+	})
+}