{AKS} az aks create/update: Outbound type should not be updated if it is not set (#30539)

Author: MartinForReal

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

@@ -2161,7 +2161,6 @@ def _get_outbound_type(
         self,
         enable_validation: bool = False,
         read_only: bool = False,
-        load_balancer_profile: ManagedClusterLoadBalancerProfile = None,
     ) -> Union[str, None]:
         """Internal function to dynamically obtain the value of outbound_type according to the context.
 
@@ -2202,6 +2201,7 @@ def _get_outbound_type(
 
         # dynamic completion
         if (
+            self.decorator_mode == DecoratorMode.CREATE and
             not read_from_mc and
             outbound_type != CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY and
             outbound_type != CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY and
@@ -2232,33 +2232,29 @@ def _get_outbound_type(
                             "be pre-configured with a route table with egress rules"
                         )
 
-                if outbound_type == CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING:
-                    if load_balancer_profile:
-                        if (
-                            load_balancer_profile.managed_outbound_i_ps or
-                            load_balancer_profile.outbound_i_ps or
-                            load_balancer_profile.outbound_ip_prefixes
-                        ):
-                            raise MutuallyExclusiveArgumentError(
-                                "userDefinedRouting doesn't support customizing \
-                                a standard load balancer with IP addresses"
-                            )
+                if outbound_type != CONST_OUTBOUND_TYPE_LOAD_BALANCER:
                     if (
                         self.get_load_balancer_managed_outbound_ip_count() or
                         self.get_load_balancer_managed_outbound_ipv6_count() or
                         self.get_load_balancer_outbound_ips() or
                         self.get_load_balancer_outbound_ip_prefixes()
                     ):
                         raise MutuallyExclusiveArgumentError(
-                            "userDefinedRouting doesn't support customizing \
-                            a standard load balancer with IP addresses"
+                            outbound_type + " doesn't support customizing "
+                            "a standard load balancer with IP addresses"
+                        )
+                if outbound_type != CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY:
+                    if (
+                        self.get_nat_gateway_managed_outbound_ip_count()
+                    ):
+                        raise MutuallyExclusiveArgumentError(
+                            outbound_type + " doesn't support customizing "
+                            "a standard nat gateway with IP addresses"
                         )
-
         return outbound_type
 
     def get_outbound_type(
         self,
-        load_balancer_profile: ManagedClusterLoadBalancerProfile = None
     ) -> Union[str, None]:
         """Dynamically obtain the value of outbound_type according to the context.
 
@@ -2280,7 +2276,7 @@ def get_outbound_type(
         :return: string or None
         """
         return self._get_outbound_type(
-            enable_validation=True, load_balancer_profile=load_balancer_profile
+            enable_validation=True
         )
 
     def _get_network_plugin_mode(self, enable_validation: bool = False) -> Union[str, None]:
@@ -5694,9 +5690,7 @@ def set_up_network_profile(self, mc: ManagedCluster) -> ManagedCluster:
         # verify outbound type
         # Note: Validation internally depends on load_balancer_sku, which is a temporary value that is
         # dynamically completed.
-        outbound_type = self.context.get_outbound_type(
-            load_balancer_profile=load_balancer_profile
-        )
+        outbound_type = self.context.get_outbound_type()
 
         # verify load balancer sku
         load_balancer_sku = safe_lower(self.context.get_load_balancer_sku())

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

@@ -1757,7 +1757,7 @@ def test_get_outbound_type(self):
             DecoratorMode.UPDATE,
         )
         self.assertEqual(ctx_1._get_outbound_type(read_only=True), None)
-        self.assertEqual(ctx_1.get_outbound_type(), "loadBalancer")
+        self.assertEqual(ctx_1.get_outbound_type(), None)
         network_profile_1 = self.models.ContainerServiceNetworkProfile(outbound_type="test_outbound_type")
         mc = self.models.ManagedCluster(location="test_location", network_profile=network_profile_1)
         ctx_1.attach_mc(mc)
@@ -1835,6 +1835,7 @@ def test_get_outbound_type(self):
                 {
                     "outbound_type": CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
                     "vnet_subnet_id": "test_vnet_subnet_id",
+                    "nat_gateway_managed_outbound_ip_count": 10
                 }
             ),
             self.models,
@@ -1848,17 +1849,10 @@ def test_get_outbound_type(self):
             AgentPoolDecoratorMode.MANAGED_CLUSTER,
         )
         ctx_5.attach_agentpool_context(agentpool_ctx_5)
-        load_balancer_profile = self.models.load_balancer_models.ManagedClusterLoadBalancerProfile(
-            outbound_ip_prefixes=self.models.load_balancer_models.ManagedClusterLoadBalancerProfileOutboundIPPrefixes(
-                public_ip_prefixes=[self.models.load_balancer_models.ResourceReference(id="test_public_ip_prefix")]
-            )
-        )
         # fail on mutually exclusive outbound_type and managed_outbound_ip_count/outbound_ips/outbound_ip_prefixes of
         # load balancer
         with self.assertRaises(MutuallyExclusiveArgumentError):
-            ctx_5.get_outbound_type(
-                load_balancer_profile=load_balancer_profile,
-            )
+            ctx_5.get_outbound_type()
 
         # invalid parameter
         ctx_6 = AKSManagedClusterContext(
@@ -1885,6 +1879,46 @@ def test_get_outbound_type(self):
         # load balancer
         with self.assertRaises(MutuallyExclusiveArgumentError):
             ctx_6.get_outbound_type()
+        ctx_7 = AKSManagedClusterContext(
+            self.cmd,
+            AKSManagedClusterParamDict(
+                {
+                    "outbound_type": CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
+                    "vnet_subnet_id": "test_vnet_subnet_id",
+                    "nat_gateway_managed_outbound_ip_count": 10,
+                }
+            ),
+            self.models,
+            DecoratorMode.CREATE,
+        )
+        agentpool_ctx_7 = AKSAgentPoolContext(
+            self.cmd,
+            AKSAgentPoolParamDict({"vnet_subnet_id": "test_vnet_subnet_id"}),
+            self.models,
+            DecoratorMode.CREATE,
+            AgentPoolDecoratorMode.MANAGED_CLUSTER,
+        )
+        ctx_7.attach_agentpool_context(agentpool_ctx_7)
+        # fail on mutually exclusive outbound_type and nat_gateway_managed_outbound_ip_count on
+        # nat gateway
+        with self.assertRaises(MutuallyExclusiveArgumentError):
+            ctx_7.get_outbound_type()
+        
+        network_profile_1 = self.models.ContainerServiceNetworkProfile(outbound_type="test_outbound_type")
+        mc = self.models.ManagedCluster(location="test_location", network_profile=network_profile_1)
+        # existing value should not be validated
+        ctx_8 = AKSManagedClusterContext(
+            self.cmd,
+            AKSManagedClusterParamDict(
+                {
+                }
+            ),
+            self.models,
+            DecoratorMode.UPDATE,
+        )
+        ctx_8.attach_mc(mc)
+        existingOutboundType = ctx_8.get_outbound_type()
+        self.assertEqual(existingOutboundType, "test_outbound_type")
 
     def test_get_network_plugin_mode(self):
         # default