{AKS} Refactor 1st part of the create sub-command in aks-preview module

src/aks-preview/azext_aks_preview/__init__.py

@@ -12,15 +12,19 @@
 from azext_aks_preview._client_factory import CUSTOM_MGMT_AKS_PREVIEW
 
 
+def register_aks_preview_resource_type():
+    register_resource_type(
+        "latest",
+        CUSTOM_MGMT_AKS_PREVIEW,
+        SDKProfile("2021-08-01", {"container_services": "2017-07-01"}),
+    )
+
+
 class ContainerServiceCommandsLoader(AzCommandsLoader):
 
     def __init__(self, cli_ctx=None):
         from azure.cli.core.commands import CliCommandType
-        register_resource_type(
-            "latest",
-            CUSTOM_MGMT_AKS_PREVIEW,
-            SDKProfile("2021-08-01", {"container_services": "2017-07-01"}),
-        )
+        register_aks_preview_resource_type()
 
         acs_custom = CliCommandType(operations_tmpl='azext_aks_preview.custom#{}')
         super(ContainerServiceCommandsLoader, self).__init__(cli_ctx=cli_ctx,

src/aks-preview/azext_aks_preview/decorator.py

@@ -0,0 +1,309 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import os
+from typing import Dict, TypeVar, Union
+
+from azure.cli.command_modules.acs._consts import DecoratorMode
+from azure.cli.command_modules.acs.decorator import (
+    AKSContext,
+    AKSCreateDecorator,
+    AKSModels,
+    AKSUpdateDecorator,
+    safe_list_get,
+)
+from azure.cli.core import AzCommandsLoader
+from azure.cli.core.azclierror import InvalidArgumentValueError
+from azure.cli.core.commands import AzCliCommand
+from azure.cli.core.profiles import ResourceType
+from azure.cli.core.util import get_file_json
+from knack.log import get_logger
+
+logger = get_logger(__name__)
+
+# type variables
+ContainerServiceClient = TypeVar("ContainerServiceClient")
+Identity = TypeVar("Identity")
+ManagedCluster = TypeVar("ManagedCluster")
+ManagedClusterLoadBalancerProfile = TypeVar("ManagedClusterLoadBalancerProfile")
+ResourceReference = TypeVar("ResourceReference")
+KubeletConfig = TypeVar("KubeletConfig")
+LinuxOSConfig = TypeVar("LinuxOSConfig")
+
+
+# pylint: disable=too-many-instance-attributes,too-few-public-methods
+class AKSPreviewModels(AKSModels):
+    def __init__(self, cmd: AzCommandsLoader, resource_type: ResourceType):
+        super().__init__(cmd, resource_type=resource_type)
+        self.__cmd = cmd
+        self.KubeletConfig = self.__cmd.get_models(
+            "KubeletConfig",
+            resource_type=self.resource_type,
+            operation_group="managed_clusters",
+        )
+        self.LinuxOSConfig = self.__cmd.get_models(
+            "LinuxOSConfig",
+            resource_type=self.resource_type,
+            operation_group="managed_clusters",
+        )
+
+
+# pylint: disable=too-many-public-methods
+class AKSPreviewContext(AKSContext):
+    def __init__(
+        self,
+        cmd: AzCliCommand,
+        raw_parameters: Dict,
+        models: AKSPreviewModels,
+        decorator_mode,
+    ):
+        super().__init__(cmd, raw_parameters, models, decorator_mode)
+
+    def get_pod_subnet_id(self) -> Union[str, None]:
+        """Obtain the value of pod_subnet_id.
+
+        :return: bool
+        """
+        # read the original value passed by the command
+        pod_subnet_id = self.raw_param.get("pod_subnet_id")
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if self.mc and self.mc.agent_pool_profiles:
+            agent_pool_profile = safe_list_get(
+                self.mc.agent_pool_profiles, 0, None
+            )
+            if (
+                agent_pool_profile and
+                agent_pool_profile.pod_subnet_id is not None
+            ):
+                pod_subnet_id = agent_pool_profile.pod_subnet_id
+
+        # this parameter does not need dynamic completion
+        # this parameter does not need validation
+        return pod_subnet_id
+
+    def get_enable_fips_image(self) -> bool:
+        """Obtain the value of enable_fips_image.
+
+        :return: bool
+        """
+        # read the original value passed by the command
+        enable_fips_image = self.raw_param.get("enable_fips_image")
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if self.mc and self.mc.agent_pool_profiles:
+            agent_pool_profile = safe_list_get(
+                self.mc.agent_pool_profiles, 0, None
+            )
+            if (
+                agent_pool_profile and
+                agent_pool_profile.enable_fips is not None
+            ):
+                enable_fips_image = agent_pool_profile.enable_fips
+
+        # this parameter does not need dynamic completion
+        # this parameter does not need validation
+        return enable_fips_image
+
+    def get_workload_runtime(self) -> Union[str, None]:
+        """Obtain the value of workload_runtime.
+
+        :return: string or None
+        """
+        # read the original value passed by the command
+        workload_runtime = self.raw_param.get("workload_runtime")
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if self.mc and self.mc.agent_pool_profiles:
+            agent_pool_profile = safe_list_get(
+                self.mc.agent_pool_profiles, 0, None
+            )
+            if (
+                agent_pool_profile and
+                agent_pool_profile.workload_runtime is not None
+            ):
+                workload_runtime = agent_pool_profile.workload_runtime
+
+        # this parameter does not need dynamic completion
+        # this parameter does not need validation
+        return workload_runtime
+
+    def get_gpu_instance_profile(self) -> Union[str, None]:
+        """Obtain the value of gpu_instance_profile.
+
+        :return: string or None
+        """
+        # read the original value passed by the command
+        gpu_instance_profile = self.raw_param.get("gpu_instance_profile")
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if self.mc and self.mc.agent_pool_profiles:
+            agent_pool_profile = safe_list_get(
+                self.mc.agent_pool_profiles, 0, None
+            )
+            if (
+                agent_pool_profile and
+                agent_pool_profile.gpu_instance_profile is not None
+            ):
+                gpu_instance_profile = agent_pool_profile.gpu_instance_profile
+
+        # this parameter does not need dynamic completion
+        # this parameter does not need validation
+        return gpu_instance_profile
+
+    def get_kubelet_config(self) -> Union[dict, KubeletConfig, None]:
+        """Obtain the value of kubelet_config.
+
+        :return: dict, KubeletConfig or None
+        """
+        # read the original value passed by the command
+        kubelet_config = None
+        kubelet_config_file_path = self.raw_param.get("kubelet_config")
+        # validate user input
+        if kubelet_config_file_path:
+            if not os.path.isfile(kubelet_config_file_path):
+                raise InvalidArgumentValueError(
+                    "{} is not valid file, or not accessable.".format(
+                        kubelet_config_file_path
+                    )
+                )
+            kubelet_config = get_file_json(kubelet_config_file_path)
+            if not isinstance(kubelet_config, dict):
+                raise InvalidArgumentValueError(
+                    "Error reading kubelet configuration from {}. "
+                    "Please see https://aka.ms/CustomNodeConfig for correct format.".format(
+                        kubelet_config_file_path
+                    )
+                )
+
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if self.mc and self.mc.agent_pool_profiles:
+            agent_pool_profile = safe_list_get(
+                self.mc.agent_pool_profiles, 0, None
+            )
+            if (
+                agent_pool_profile and
+                agent_pool_profile.kubelet_config is not None
+            ):
+                kubelet_config = agent_pool_profile.kubelet_config
+
+        # this parameter does not need dynamic completion
+        # this parameter does not need validation
+        return kubelet_config
+
+    def get_linux_os_config(self) -> Union[dict, LinuxOSConfig, None]:
+        """Obtain the value of linux_os_config.
+
+        :return: dict, LinuxOSConfig or None
+        """
+        # read the original value passed by the command
+        linux_os_config = None
+        linux_os_config_file_path = self.raw_param.get("linux_os_config")
+        # validate user input
+        if linux_os_config_file_path:
+            if not os.path.isfile(linux_os_config_file_path):
+                raise InvalidArgumentValueError(
+                    "{} is not valid file, or not accessable.".format(
+                        linux_os_config_file_path
+                    )
+                )
+            linux_os_config = get_file_json(linux_os_config_file_path)
+            if not isinstance(linux_os_config, dict):
+                raise InvalidArgumentValueError(
+                    "Error reading Linux OS configuration from {}. "
+                    "Please see https://aka.ms/CustomNodeConfig for correct format.".format(
+                        linux_os_config_file_path
+                    )
+                )
+
+        # try to read the property value corresponding to the parameter from the `mc` object
+        if self.mc and self.mc.agent_pool_profiles:
+            agent_pool_profile = safe_list_get(
+                self.mc.agent_pool_profiles, 0, None
+            )
+            if (
+                agent_pool_profile and
+                agent_pool_profile.linux_os_config is not None
+            ):
+                linux_os_config = agent_pool_profile.linux_os_config
+
+        # this parameter does not need dynamic completion
+        # this parameter does not need validation
+        return linux_os_config
+
+
+class AKSPreviewCreateDecorator(AKSCreateDecorator):
+    # pylint: disable=super-init-not-called
+    def __init__(
+        self,
+        cmd: AzCliCommand,
+        client: ContainerServiceClient,
+        raw_parameters: Dict,
+        resource_type: ResourceType,
+    ):
+        """Internal controller of aks_create in aks-preview.
+
+        Break down the all-in-one aks_create function into several relatively independent functions (some of them have
+        a certain order dependency) that only focus on a specific profile or process a specific piece of logic.
+        In addition, an overall control function is provided. By calling the aforementioned independent functions one
+        by one, a complete ManagedCluster object is gradually decorated and finally requests are sent to create a
+        cluster.
+        """
+        self.cmd = cmd
+        self.client = client
+        self.models = AKSPreviewModels(cmd, resource_type)
+        # store the context in the process of assemble the ManagedCluster object
+        self.context = AKSPreviewContext(
+            cmd,
+            raw_parameters,
+            self.models,
+            decorator_mode=DecoratorMode.CREATE,
+        )
+
+    def set_up_agent_pool_profiles(self, mc: ManagedCluster) -> ManagedCluster:
+        """Set up agent pool profiles for the ManagedCluster object.
+
+        :return: the ManagedCluster object
+        """
+        mc = super().set_up_agent_pool_profiles(mc)
+        agent_pool_profile = safe_list_get(mc.agent_pool_profiles, 0, None)
+
+        # set up extra parameters supported in aks-preview
+        agent_pool_profile.pod_subnet_id = self.context.get_pod_subnet_id()
+        agent_pool_profile.enable_fips = self.context.get_enable_fips_image()
+        agent_pool_profile.workload_runtime = (
+            self.context.get_workload_runtime()
+        )
+        agent_pool_profile.gpu_instance_profile = (
+            self.context.get_gpu_instance_profile()
+        )
+        agent_pool_profile.kubelet_config = self.context.get_kubelet_config()
+        agent_pool_profile.linux_os_config = self.context.get_linux_os_config()
+        return mc
+
+
+class AKSPreviewUpdateDecorator(AKSUpdateDecorator):
+    # pylint: disable=super-init-not-called
+    def __init__(
+        self,
+        cmd: AzCliCommand,
+        client: ContainerServiceClient,
+        raw_parameters: Dict,
+        resource_type: ResourceType,
+    ):
+        """Internal controller of aks_update in aks-preview.
+
+        Break down the all-in-one aks_update function into several relatively independent functions (some of them have
+        a certain order dependency) that only focus on a specific profile or process a specific piece of logic.
+        In addition, an overall control function is provided. By calling the aforementioned independent functions one
+        by one, a complete ManagedCluster object is gradually updated and finally requests are sent to update an
+        existing cluster.
+        """
+        self.cmd = cmd
+        self.client = client
+        self.models = AKSPreviewModels(cmd, resource_type)
+        # store the context in the process of assemble the ManagedCluster object
+        self.context = AKSPreviewContext(
+            cmd,
+            raw_parameters,
+            self.models,
+            decorator_mode=DecoratorMode.UPDATE,
+        )

src/aks-preview/azext_aks_preview/tests/latest/data/invalidconfig.json

@@ -0,0 +1 @@
+[]

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

@@ -0,0 +1,55 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from knack import CLI
+import tempfile
+
+from azure.cli.core import AzCommandsLoader
+from azure.cli.core.cloud import get_active_cloud
+from azure.cli.core.commands import AzCliCommand
+from azure.cli.core._config import ENV_VAR_PREFIX
+
+MOCK_CLI_CONFIG_DIR = tempfile.mkdtemp()
+MOCK_CLI_ENV_VAR_PREFIX = "MOCK_" + ENV_VAR_PREFIX
+
+
+class MockClient(object):
+    def __init__(self):
+        pass
+
+
+class MockCLI(CLI):
+    def __init__(self):
+        super(MockCLI, self).__init__(
+            cli_name="mock_cli",
+            config_dir=MOCK_CLI_CONFIG_DIR,
+            config_env_var_prefix=MOCK_CLI_ENV_VAR_PREFIX,
+        )
+        self.cloud = get_active_cloud(self)
+
+
+class MockCmd(object):
+    def __init__(self, cli_ctx):
+        self.cli_ctx = cli_ctx
+        self.cmd = AzCliCommand(AzCommandsLoader(cli_ctx), "mock-cmd", None)
+
+    def supported_api_version(
+        self,
+        resource_type=None,
+        min_api=None,
+        max_api=None,
+        operation_group=None,
+        parameter_name=None,
+    ):
+        return self.cmd.supported_api_version(
+            resource_type=resource_type,
+            min_api=min_api,
+            max_api=max_api,
+            operation_group=operation_group,
+            parameter_name=parameter_name,
+        )
+
+    def get_models(self, *attr_args, **kwargs):
+        return self.cmd.get_models(*attr_args, **kwargs)