sync changes from PR Azure#4817

src/aks-preview/azext_aks_preview/managed_cluster_decorator.py

@@ -813,6 +813,8 @@ def get_disk_driver(self) -> Optional[ManagedClusterStorageProfileDiskCSIDriver]
         """
         enable_disk_driver = self.raw_param.get("enable_disk_driver")
         disable_disk_driver = self.raw_param.get("disable_disk_driver")
+        if not enable_disk_driver and not disable_disk_driver:
+            return None
         profile = self.models.ManagedClusterStorageProfileDiskCSIDriver()
 
         if enable_disk_driver and disable_disk_driver:
@@ -842,6 +844,8 @@ def get_file_driver(self) -> Optional[ManagedClusterStorageProfileFileCSIDriver]
         """
         enable_file_driver = self.raw_param.get("enable_file_driver")
         disable_file_driver = self.raw_param.get("disable_file_driver")
+        if not enable_file_driver and not disable_file_driver:
+            return None
         profile = self.models.ManagedClusterStorageProfileFileCSIDriver()
 
         if enable_file_driver and disable_file_driver:
@@ -871,6 +875,8 @@ def get_snapshot_controller(self) -> Optional[ManagedClusterStorageProfileSnapsh
         """
         enable_snapshot_controller = self.raw_param.get("enable_snapshot_controller")
         disable_snapshot_controller = self.raw_param.get("disable_snapshot_controller")
+        if not enable_snapshot_controller and not disable_snapshot_controller:
+            return None
         profile = self.models.ManagedClusterStorageProfileSnapshotController()
 
         if enable_snapshot_controller and disable_snapshot_controller:

src/aks-preview/azext_aks_preview/tests/latest/test_managed_cluster_decorator.py

@@ -1237,30 +1237,64 @@ def test_get_snapshot_controller(self):
             ctx_1.get_snapshot_controller()
 
     def test_get_storage_profile(self):
-        # default
+        # create
         ctx_1 = AKSPreviewManagedClusterContext(
             self.cmd,
-            AKSManagedClusterParamDict({}),
+            AKSManagedClusterParamDict({
+                "disable_disk_driver": True,
+            }),
             self.models,
             decorator_mode=DecoratorMode.UPDATE,
         )
-        storage_profile = self.models.ManagedClusterStorageProfile(
+        mc_1 = self.models.ManagedCluster(
+            location="test_location",
+        )
+        ctx_1.attach_mc(mc_1)
+        ground_truth_storage_profile_1 = self.models.ManagedClusterStorageProfile(
+            disk_csi_driver=self.models.ManagedClusterStorageProfileDiskCSIDriver(
+                enabled=False,
+            ),
+            file_csi_driver=None,
+            snapshot_controller=None,
+        )
+        self.assertEqual(ctx_1.get_storage_profile(), ground_truth_storage_profile_1)
+
+        # update
+        ctx_2 = AKSPreviewManagedClusterContext(
+            self.cmd,
+            AKSManagedClusterParamDict({
+                "enable_file_driver": True,
+                "disable_snapshot_controller": True,
+            }),
+            self.models,
+            decorator_mode=DecoratorMode.UPDATE,
+        )
+        storage_profile_2 = self.models.ManagedClusterStorageProfile(
             disk_csi_driver=self.models.ManagedClusterStorageProfileDiskCSIDriver(
                 enabled=True,
             ),
             file_csi_driver=self.models.ManagedClusterStorageProfileFileCSIDriver(
-                enabled=True,
+                enabled=False,
             ),
             snapshot_controller=self.models.ManagedClusterStorageProfileSnapshotController(
                 enabled=True,
             ),
         )
-        mc = self.models.ManagedCluster(
+        mc_2 = self.models.ManagedCluster(
             location="test_location",
-            storage_profile=storage_profile,
+            storage_profile=storage_profile_2,
         )
-        ctx_1.attach_mc(mc)
-        self.assertEqual(ctx_1.get_storage_profile(), storage_profile)
+        ctx_2.attach_mc(mc_2)
+        ground_truth_storage_profile_2 = self.models.ManagedClusterStorageProfile(
+            disk_csi_driver=None,
+            file_csi_driver=self.models.ManagedClusterStorageProfileFileCSIDriver(
+                enabled=True,
+            ),
+            snapshot_controller=self.models.ManagedClusterStorageProfileSnapshotController(
+                enabled=False,
+            ),
+        )
+        self.assertEqual(ctx_2.get_storage_profile(), ground_truth_storage_profile_2)
 
     def test_get_user_assignd_identity_from_mc(self):
         ctx_1 = AKSPreviewManagedClusterContext(
@@ -1870,15 +1904,9 @@ def test_construct_mc_profile_preview(self):
         )
         identity_1 = self.models.ManagedClusterIdentity(type="SystemAssigned")
         storage_profile_1 = self.models.ManagedClusterStorageProfile(
-            disk_csi_driver=self.models.ManagedClusterStorageProfileDiskCSIDriver(
-                enabled=True,
-            ),
-            file_csi_driver=self.models.ManagedClusterStorageProfileFileCSIDriver(
-                enabled=True,
-            ),
-            snapshot_controller=self.models.ManagedClusterStorageProfileSnapshotController(
-                enabled=True,
-            ),
+            disk_csi_driver=None,
+            file_csi_driver=None,
+            snapshot_controller=None,
         )
         ground_truth_mc_1 = self.models.ManagedCluster(
             location="test_location",
@@ -2889,9 +2917,9 @@ def test_update_mc_profile_preview(self):
             )
         }
         ground_truth_storage_profile_1 = self.models.ManagedClusterStorageProfile(
-            disk_csi_driver=self.models.ManagedClusterStorageProfileDiskCSIDriver(),
-            file_csi_driver=self.models.ManagedClusterStorageProfileFileCSIDriver(),
-            snapshot_controller=self.models.ManagedClusterStorageProfileSnapshotController(),
+            disk_csi_driver=None,
+            file_csi_driver=None,
+            snapshot_controller=None,
         )
         ground_truth_mc_1 = self.models.ManagedCluster(
             location="test_location",