test: add test for MutableCSINodeAllocatableCount feature

Signed-off-by: Penghao <[email protected]>

Author: Phaow

test/extended/storage/helpers.go

@@ -2,6 +2,8 @@ package storage
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"time"
 
 	g "github.com/onsi/ginkgo/v2"
@@ -10,6 +12,7 @@ import (
 	imageregistry "github.com/openshift/client-go/imageregistry/clientset/versioned"
 	clusteroperatorhelpers "github.com/openshift/library-go/pkg/config/clusteroperator/v1helpers"
 	exutil "github.com/openshift/origin/test/extended/util"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	e2e "k8s.io/kubernetes/test/e2e/framework"
 	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
@@ -65,3 +68,94 @@ func skipIfNotS3Storage(oc *exutil.CLI) {
 		e2eskipper.Skipf("No S3 storage detected")
 	}
 }
+
+// GetSchedulableLinuxWorkerNodes returns a list of nodes that match the requirements:
+// os: linux, role: worker, status: ready, schedulable
+func GetSchedulableLinuxWorkerNodes(oc *exutil.CLI) ([]v1.Node, error) {
+	var nodes, workers []v1.Node
+	linuxNodes, err := oc.AdminKubeClient().CoreV1().Nodes().List(context.Background(), metav1.ListOptions{LabelSelector: "kubernetes.io/os=linux"})
+	// get schedulable linux worker nodes
+	for _, node := range linuxNodes.Items {
+		if _, ok := node.Labels["node-role.kubernetes.io/worker"]; ok && !node.Spec.Unschedulable {
+			workers = append(workers, node)
+		}
+	}
+	// get ready nodes
+	for _, worker := range workers {
+		for _, con := range worker.Status.Conditions {
+			if con.Type == "Ready" && con.Status == "True" {
+				nodes = append(nodes, worker)
+				break
+			}
+		}
+	}
+	return nodes, err
+}
+
+// getCSINodeAllocatableCount gets the allocatable count for a specific CSI driver from a CSINode
+func getCSINodeAllocatableCountByDriver(ctx context.Context, oc *exutil.CLI, nodeName, driverName string) int32 {
+	csiNode, err := oc.AdminKubeClient().StorageV1().CSINodes().Get(ctx, nodeName, metav1.GetOptions{})
+	o.Expect(err).NotTo(o.HaveOccurred(), fmt.Sprintf("failed to get CSINode for node %s", nodeName))
+
+	for _, driver := range csiNode.Spec.Drivers {
+		if driver.Name == driverName {
+			if driver.Allocatable != nil && driver.Allocatable.Count != nil {
+				return *driver.Allocatable.Count
+			}
+		}
+	}
+	e2e.Failf("CSI driver %s not found in CSINode %s", driverName, nodeName)
+	return 0
+}
+
+// getAWSInstanceIDFromNode extracts the AWS instance ID from a node's providerID
+func getAWSInstanceIDFromNode(ctx context.Context, oc *exutil.CLI, nodeName string) string {
+	node, err := oc.AdminKubeClient().CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
+	o.Expect(err).NotTo(o.HaveOccurred(), fmt.Sprintf("failed to get node %s", nodeName))
+
+	providerID := node.Spec.ProviderID
+	o.Expect(providerID).NotTo(o.BeEmpty(), "node providerID should not be empty")
+
+	parts := strings.Split(providerID, "/")
+	o.Expect(len(parts)).To(o.BeNumerically(">=", 5), "invalid AWS providerID format")
+
+	instanceID := parts[len(parts)-1]
+	o.Expect(instanceID).To(o.HavePrefix("i-"), "instance ID should start with 'i-'")
+
+	return instanceID
+}
+
+// getAttachedVolumeCountFromVolumeAttachments returns how many VolumeAttachments
+// currently target the given node.
+func getAttachedVolumeCountFromVolumeAttachments(ctx context.Context, oc *exutil.CLI, nodeName string) int32 {
+	vaList, err := oc.AdminKubeClient().
+		StorageV1().
+		VolumeAttachments().
+		List(ctx, metav1.ListOptions{})
+	o.Expect(err).NotTo(o.HaveOccurred(), "failed to list VolumeAttachments")
+
+	var count int32 = 0
+	for _, va := range vaList.Items {
+		if va.Spec.NodeName == nodeName && va.Status.Attached {
+			count++
+		}
+	}
+	return count
+}
+
+// GetCSIStorageClassByProvisioner finds a StorageClass that uses the CSI driver provisioner
+// it will return the name of the first matched StorageClass if found, otherwise it will fail the test
+func GetCSIStorageClassByProvisioner(ctx context.Context, oc *exutil.CLI, provisioner string) string {
+	storageClasses, err := oc.AdminKubeClient().StorageV1().StorageClasses().List(ctx, metav1.ListOptions{})
+	o.Expect(err).NotTo(o.HaveOccurred(), "failed to list StorageClasses")
+
+	for _, sc := range storageClasses.Items {
+		if sc.Provisioner == ebsCSIDriverName {
+			e2e.Logf("Found CSI StorageClass: %s", sc.Name)
+			return sc.Name
+		}
+	}
+
+	e2e.Failf("No StorageClass found with provisioner %s", ebsCSIDriverName)
+	return ""
+}