[AKS] az aks update: Fix bug where autoscale profile boolean values could be incorrectly serialized as strings

src/azure-cli/azure/cli/command_modules/acs/managed_cluster_decorator.py

@@ -452,7 +452,12 @@ def __validate_cluster_autoscaler_profile(
             valid_keys = list(
                 k.replace("_", "-") for k in attribute_list(self.models.ManagedClusterPropertiesAutoScalerProfile())
             )
-            for key in cluster_autoscaler_profile.keys():
+            boolean_keys = {
+                "daemonset-eviction-for-empty-nodes",
+                "daemonset-eviction-for-occupied-nodes",
+                "ignore-daemonsets-utilization",
+            }
+            for key, value in cluster_autoscaler_profile.items():
                 if not key:
                     raise InvalidArgumentValueError("Empty key specified for cluster-autoscaler-profile")
                 if key not in valid_keys:
@@ -461,6 +466,13 @@ def __validate_cluster_autoscaler_profile(
                             key, ", ".join(valid_keys)
                         )
                     )
+                if key in boolean_keys and not isinstance(value, bool):
+                    if not isinstance(value, str) or value.lower() not in ("true", "false"):
+                        raise InvalidArgumentValueError(
+                            "Value '{}' for cluster-autoscaler-profile key '{}' must be either 'true' or 'false'."
+                            .format(value, key)
+                        )
+                    cluster_autoscaler_profile[key] = value.lower() == "true"
         return cluster_autoscaler_profile
 
     # pylint: disable=no-self-use
@@ -4819,7 +4831,10 @@ def get_node_os_upgrade_channel(self) -> Union[str, None]:
         # this parameter does not need validation
         return node_os_upgrade_channel
 
-    def _get_cluster_autoscaler_profile(self, read_only: bool = False) -> Union[Dict[str, str], None]:
+    def _get_cluster_autoscaler_profile(
+        self,
+        read_only: bool = False
+    ) -> Union[Dict[str, Union[str, bool]], None]:
         """Internal function to dynamically obtain the value of cluster_autoscaler_profile according to the context.
 
         This function will call function "__validate_cluster_autoscaler_profile" to parse and verify the parameter
@@ -4849,7 +4864,7 @@ def _get_cluster_autoscaler_profile(self, read_only: bool = False) -> Union[Dict
         # dynamic completion for update mode only
         if not read_only and self.decorator_mode == DecoratorMode.UPDATE:
             if cluster_autoscaler_profile and self.mc and self.mc.auto_scaler_profile:
-                # shallow copy should be enough for string-to-string dictionary
+                # shallow copy is enough for a dictionary of scalar values
                 copy_of_raw_dict = dict(self.mc.auto_scaler_profile)
                 new_options_dict = dict(cluster_autoscaler_profile.items())
                 copy_of_raw_dict.update(new_options_dict)
@@ -4858,7 +4873,7 @@ def _get_cluster_autoscaler_profile(self, read_only: bool = False) -> Union[Dict
         # this parameter does not need validation
         return cluster_autoscaler_profile
 
-    def get_cluster_autoscaler_profile(self) -> Union[Dict[str, str], None]:
+    def get_cluster_autoscaler_profile(self) -> Union[Dict[str, Union[str, bool]], None]:
         """Dynamically obtain the value of cluster_autoscaler_profile according to the context.
 
         This function will call function "__validate_cluster_autoscaler_profile" to parse and verify the parameter

src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_managed_cluster_decorator.py

@@ -249,6 +249,40 @@ def test_validate_cluster_autoscaler_profile(self):
         with self.assertRaises(InvalidArgumentValueError):
             ctx._AKSManagedClusterContext__validate_cluster_autoscaler_profile(s10)
 
+        # SDK boolean values
+        s11 = [
+            "daemonset-eviction-for-empty-nodes=true",
+            "daemonset-eviction-for-occupied-nodes=FALSE",
+            "ignore-daemonsets-utilization=True",
+            "balance-similar-node-groups=true",
+        ]
+        t11 = ctx._AKSManagedClusterContext__validate_cluster_autoscaler_profile(s11)
+        g11 = {
+            "daemonset-eviction-for-empty-nodes": True,
+            "daemonset-eviction-for-occupied-nodes": False,
+            "ignore-daemonsets-utilization": True,
+            "balance-similar-node-groups": "true",
+        }
+        self.assertEqual(t11, g11)
+
+        # already-normalized SDK boolean values
+        s12 = {
+            "daemonset-eviction-for-empty-nodes": False,
+            "daemonset-eviction-for-occupied-nodes": True,
+            "ignore-daemonsets-utilization": False,
+        }
+        t12 = ctx._AKSManagedClusterContext__validate_cluster_autoscaler_profile(s12)
+        self.assertEqual(t12, s12)
+
+        # invalid SDK boolean value
+        for key in (
+            "daemonset-eviction-for-empty-nodes",
+            "daemonset-eviction-for-occupied-nodes",
+            "ignore-daemonsets-utilization",
+        ):
+            with self.subTest(key=key), self.assertRaises(InvalidArgumentValueError):
+                ctx._AKSManagedClusterContext__validate_cluster_autoscaler_profile({key: "yes"})
+
     def test_validate_gmsa_options(self):
         # default
         ctx = AKSManagedClusterContext(