[AKS] Preserve --max-unavailable in az aks nodepool upgrade

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

@@ -3189,6 +3189,8 @@ def aks_agentpool_upgrade(cmd, client, resource_group_name, cluster_name,
         instance.upgrade_settings.node_soak_duration_in_minutes = node_soak_duration
     if undrainable_node_behavior:
         instance.upgrade_settings.undrainable_node_behavior = undrainable_node_behavior
+    if max_unavailable:
+        instance.upgrade_settings.max_unavailable = max_unavailable
 
     # custom headers
     aks_custom_headers = extract_comma_separated_string(

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

@@ -26,6 +26,7 @@
     _get_command_context,
     _update_addons,
     aks_enable_addons,
+    aks_agentpool_upgrade,
     aks_stop,
     is_monitoring_addon_enabled,
     k8s_install_kubectl,
@@ -944,6 +945,36 @@ def setUp(self):
         self.models = AKSManagedClusterModels(self.cmd, ResourceType.MGMT_CONTAINERSERVICE)
         self.client = MockClient()
 
+    def test_aks_agentpool_upgrade_sets_max_unavailable(self):
+        agent_pool = mock.MagicMock()
+        agent_pool.orchestrator_version = "1.28.0"
+        agent_pool.provisioning_state = "Succeeded"
+        agent_pool.upgrade_settings = None
+        self.client.get = mock.Mock(return_value=agent_pool)
+        self.client.begin_create_or_update = mock.Mock(return_value=agent_pool)
+        self.cmd.get_models = mock.Mock(return_value=mock.Mock)
+
+        result = aks_agentpool_upgrade(
+            self.cmd,
+            self.client,
+            "rg",
+            "cluster",
+            "nodepool",
+            kubernetes_version="1.29.0",
+            max_unavailable="1",
+            yes=True,
+        )
+
+        self.assertEqual(result, agent_pool)
+        self.assertEqual(agent_pool.upgrade_settings.max_unavailable, "1")
+        self.client.begin_create_or_update.assert_called_once_with(
+            "rg",
+            "cluster",
+            "nodepool",
+            agent_pool,
+            headers={},
+        )
+
     def test_aks_stop(self):
         # public cluster: call begin_stop
         mc_1 = self.models.ManagedCluster(location="test_location")