Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -2362,7 +2362,7 @@ def get_sku_name(self) -> str:
skuName = CONST_MANAGED_CLUSTER_SKU_NAME_BASE
return skuName

def _get_outbound_type(
def _get_outbound_type( # pylint: disable=too-many-branches
self,
enable_validation: bool = False,
read_only: bool = False,
Expand Down Expand Up @@ -2436,15 +2436,31 @@ def _get_outbound_type(

if outbound_type == CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING:
if self.get_vnet_subnet_id() in ["", None]:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userDefinedRouting and it must "
"be pre-configured with a route table with egress rules"
if self.decorator_mode == DecoratorMode.CREATE:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userDefinedRouting and it must "
"be pre-configured with a route table with egress rules"
)
raise InvalidArgumentValueError(
"Updating outbound type to userDefinedRouting is only supported for "
"clusters using a custom (BYO) virtual network. Managed VNet clusters "
"cannot be updated to userDefinedRouting. Please refer to "
"https://learn.microsoft.com/en-us/azure/aks/egress-outboundtype"
"#updating-outboundtype-after-cluster-creation for supported migration paths."
)
if outbound_type == CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY:
if self.get_vnet_subnet_id() in ["", None]:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userAssignedNATGateway and it must "
"be pre-configured with a NAT gateway with outbound ips"
if self.decorator_mode == DecoratorMode.CREATE:
raise RequiredArgumentMissingError(
"--vnet-subnet-id must be specified for userAssignedNATGateway and it must "
"be pre-configured with a NAT gateway with outbound ips"
)
raise InvalidArgumentValueError(
"Updating outbound type to userAssignedNATGateway is only supported for "
"clusters using a custom (BYO) virtual network. Managed VNet clusters "
"cannot be updated to userAssignedNATGateway. Please refer to "
"https://learn.microsoft.com/en-us/azure/aks/egress-outboundtype"
"#updating-outboundtype-after-cluster-creation for supported migration paths."
)
if outbound_type == CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY:
if self.get_vnet_subnet_id() not in ["", None]:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
CONST_MONITORING_ADDON_NAME,
CONST_MONITORING_LOG_ANALYTICS_WORKSPACE_RESOURCE_ID,
CONST_OPEN_SERVICE_MESH_ADDON_NAME,
CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY,
CONST_OUTBOUND_TYPE_LOAD_BALANCER,
Expand Down Expand Up @@ -2122,6 +2123,94 @@ def test_get_outbound_type(self):
expect_outbound_type_13 = CONST_OUTBOUND_TYPE_MANAGED_NAT_GATEWAY
self.assertEqual(outbound_type_13,expect_outbound_type_13)

def test_get_outbound_type_update_udr_byo_vnet(self):
"""UPDATE mode with UDR and BYO VNet (vnet_subnet_id present) should succeed."""
ctx = AKSManagedClusterContext(
self.cmd,
AKSManagedClusterParamDict(
{
"outbound_type": CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
}
),
self.models,
DecoratorMode.UPDATE,
)
ctx.agentpool_context = mock.MagicMock()
ctx.agentpool_context.get_vnet_subnet_id.return_value = (
"/subscriptions/00000000-0000-0000-0000-000000000000/resourceGroups/rg/"
"providers/Microsoft.Network/virtualNetworks/vnet/subnets/subnet"
)
self.assertEqual(ctx.get_outbound_type(), CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING)

def test_get_outbound_type_update_udr_managed_vnet(self):
"""UPDATE mode with UDR and managed VNet (no vnet_subnet_id) should raise InvalidArgumentValueError."""
ctx = AKSManagedClusterContext(
self.cmd,
AKSManagedClusterParamDict(
{
"outbound_type": CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
}
),
self.models,
DecoratorMode.UPDATE,
)
agentpool_ctx = AKSAgentPoolContext(
self.cmd,
AKSAgentPoolParamDict({}),
self.models,
DecoratorMode.UPDATE,
AgentPoolDecoratorMode.MANAGED_CLUSTER,
)
ctx.attach_agentpool_context(agentpool_ctx)
with self.assertRaises(InvalidArgumentValueError):
ctx.get_outbound_type()

def test_get_outbound_type_update_user_assigned_nat_gw_managed_vnet(self):
"""UPDATE mode with userAssignedNATGateway and managed VNet should raise InvalidArgumentValueError."""
ctx = AKSManagedClusterContext(
self.cmd,
AKSManagedClusterParamDict(
{
"outbound_type": CONST_OUTBOUND_TYPE_USER_ASSIGNED_NAT_GATEWAY,
}
),
self.models,
DecoratorMode.UPDATE,
)
agentpool_ctx = AKSAgentPoolContext(
self.cmd,
AKSAgentPoolParamDict({}),
self.models,
DecoratorMode.UPDATE,
AgentPoolDecoratorMode.MANAGED_CLUSTER,
)
ctx.attach_agentpool_context(agentpool_ctx)
with self.assertRaises(InvalidArgumentValueError):
ctx.get_outbound_type()

def test_get_outbound_type_create_udr_no_subnet(self):
"""CREATE mode with UDR and no vnet_subnet_id should still raise RequiredArgumentMissingError."""
ctx = AKSManagedClusterContext(
self.cmd,
AKSManagedClusterParamDict(
{
"outbound_type": CONST_OUTBOUND_TYPE_USER_DEFINED_ROUTING,
}
),
self.models,
DecoratorMode.CREATE,
)
agentpool_ctx = AKSAgentPoolContext(
self.cmd,
AKSAgentPoolParamDict({}),
self.models,
DecoratorMode.CREATE,
AgentPoolDecoratorMode.MANAGED_CLUSTER,
)
ctx.attach_agentpool_context(agentpool_ctx)
with self.assertRaises(RequiredArgumentMissingError):
ctx.get_outbound_type()

def test_get_network_plugin_mode(self):
# default
ctx_1 = AKSManagedClusterContext(
Expand Down
Loading