diff --git a/src/azure-cli/azure/cli/command_modules/acs/azurecontainerstorage/_helpers.py b/src/azure-cli/azure/cli/command_modules/acs/azurecontainerstorage/_helpers.py index e642488e800..05c3fd4b368 100644 --- a/src/azure-cli/azure/cli/command_modules/acs/azurecontainerstorage/_helpers.py +++ b/src/azure-cli/azure/cli/command_modules/acs/azurecontainerstorage/_helpers.py @@ -125,6 +125,15 @@ def get_k8s_extension_module(module_name): ) +def _is_config_setting_enabled(config_settings, key, default=False): + value = config_settings.get(key) + if value is None: + return default + if isinstance(value, bool): + return value + return str(value).strip().lower() == "true" + + def check_if_extension_is_installed(cmd, resource_group, cluster_name) -> bool: client_factory = get_k8s_extension_module(CONST_K8S_EXTENSION_CLIENT_FACTORY_MOD_NAME) client = client_factory.cf_k8s_extension_operation(cmd.cli_ctx) @@ -180,25 +189,23 @@ def get_extension_installed_and_cluster_configs_v1( config_settings = extension.configuration_settings if is_extension_installed and config_settings is not None: - is_cli_operation_active = config_settings.get("global.cli.activeControl", "False") == "True" + is_cli_operation_active = _is_config_setting_enabled(config_settings, "global.cli.activeControl") if is_cli_operation_active: - is_azureDisk_enabled = ( - config_settings.get("global.cli.storagePool.azureDisk.enabled", "False") == "True" + is_azureDisk_enabled = _is_config_setting_enabled( + config_settings, "global.cli.storagePool.azureDisk.enabled" ) - is_elasticSan_enabled = ( - config_settings.get("global.cli.storagePool.elasticSan.enabled", "False") == "True" + is_elasticSan_enabled = _is_config_setting_enabled( + config_settings, "global.cli.storagePool.elasticSan.enabled" ) - is_ephemeralDisk_nvme_enabled = ( - config_settings.get("global.cli.storagePool.ephemeralDisk.nvme.enabled", "False") == "True" + is_ephemeralDisk_nvme_enabled = _is_config_setting_enabled( + config_settings, "global.cli.storagePool.ephemeralDisk.nvme.enabled" ) - is_ephemeralDisk_localssd_enabled = ( - config_settings.get("global.cli.storagePool.ephemeralDisk.temp.enabled", "False") == "True" + is_ephemeralDisk_localssd_enabled = _is_config_setting_enabled( + config_settings, "global.cli.storagePool.ephemeralDisk.temp.enabled" ) cpu_value = config_settings.get("global.cli.resources.ioEngine.cpu", "1") - enable_ephemeral_bypass_annotation = ( - config_settings.get( - "global.cli.storagePool.ephemeralDisk.enableEphemeralBypassAnnotation", "False" - ) == "True" + enable_ephemeral_bypass_annotation = _is_config_setting_enabled( + config_settings, "global.cli.storagePool.ephemeralDisk.enableEphemeralBypassAnnotation" ) perf_tier = config_settings.get( "global.cli.storagePool.ephemeralDisk.nvme.perfTier", @@ -273,11 +280,11 @@ def get_extension_installed_and_cluster_configs( config_settings = extension.configuration_settings if is_extension_installed and config_settings is not None: - is_ephemeral_disk_enabled = ( - config_settings.get("csiDriverConfigs.local-csi-driver.enabled", "False") == "True" + is_ephemeral_disk_enabled = _is_config_setting_enabled( + config_settings, "csiDriverConfigs.local-csi-driver.enabled" ) - is_elastic_san_enabled = ( - config_settings.get("csiDriverConfigs.azuresan-csi-driver.enabled", "False") == "True" + is_elastic_san_enabled = _is_config_setting_enabled( + config_settings, "csiDriverConfigs.azuresan-csi-driver.enabled" ) except: # pylint: disable=bare-except diff --git a/src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_helpers.py b/src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_helpers.py index f5211a1b94d..4fe8171813b 100644 --- a/src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_helpers.py +++ b/src/azure-cli/azure/cli/command_modules/acs/tests/latest/test_helpers.py @@ -4,6 +4,7 @@ # -------------------------------------------------------------------------------------------- import unittest +from types import SimpleNamespace from unittest.mock import Mock, patch from azure.cli.command_modules.acs._helpers import ( @@ -23,6 +24,15 @@ safe_list_get, safe_lower, ) +from azure.cli.command_modules.acs.azurecontainerstorage import _helpers as acstor_helpers +from azure.cli.command_modules.acs.azurecontainerstorage._consts import ( + CONST_ACSTOR_K8S_EXTENSION_NAME, + CONST_ACSTOR_V1_K8S_EXTENSION_NAME, + CONST_DISK_TYPE_PV_WITH_ANNOTATION, + CONST_EPHEMERAL_NVME_PERF_TIER_PREMIUM, + CONST_K8S_EXTENSION_CLIENT_FACTORY_MOD_NAME, + CONST_K8S_EXTENSION_CUSTOM_MOD_NAME, +) from azure.cli.command_modules.acs.base_decorator import BaseAKSModels from azure.cli.command_modules.acs.tests.latest.mocks import MockCLI, MockCmd from azure.cli.core.azclierror import ( @@ -207,6 +217,73 @@ def test_check_is_managed_aad_cluster(self): self.assertEqual(check_is_managed_aad_cluster(mc_2), True) +class AzureContainerStorageHelpersTestCase(unittest.TestCase): + def _patch_k8s_extension_modules(self, extension): + client_factory = Mock() + client_factory.cf_k8s_extension_operation.return_value = Mock() + custom_mod = Mock() + custom_mod.show_k8s_extension.return_value = extension + + def get_module(module_name): + if module_name == CONST_K8S_EXTENSION_CLIENT_FACTORY_MOD_NAME: + return client_factory + if module_name == CONST_K8S_EXTENSION_CUSTOM_MOD_NAME: + return custom_mod + raise AssertionError(module_name) + + return patch.object(acstor_helpers, "get_k8s_extension_module", side_effect=get_module) + + def test_get_extension_installed_and_cluster_configs_v1_reads_lowercase_boolean_strings(self): + extension = SimpleNamespace( + extension_type=CONST_ACSTOR_V1_K8S_EXTENSION_NAME, + configuration_settings={ + "global.cli.activeControl": "true", + "global.cli.storagePool.azureDisk.enabled": "true", + "global.cli.storagePool.elasticSan.enabled": "false", + "global.cli.storagePool.ephemeralDisk.nvme.enabled": "true", + "global.cli.storagePool.ephemeralDisk.temp.enabled": "false", + "global.cli.resources.ioEngine.cpu": "2.5", + "global.cli.storagePool.ephemeralDisk.enableEphemeralBypassAnnotation": "true", + "global.cli.storagePool.ephemeralDisk.nvme.perfTier": "premium", + } + ) + + with self._patch_k8s_extension_modules(extension): + result = acstor_helpers.get_extension_installed_and_cluster_configs_v1( + Mock(cli_ctx=Mock()), "rg", "cluster", [] + ) + + self.assertEqual( + result, + ( + True, + True, + False, + False, + True, + 2.5, + CONST_DISK_TYPE_PV_WITH_ANNOTATION, + CONST_EPHEMERAL_NVME_PERF_TIER_PREMIUM, + ) + ) + + def test_get_extension_installed_and_cluster_configs_reads_boolean_settings(self): + extension = SimpleNamespace( + extension_type=CONST_ACSTOR_K8S_EXTENSION_NAME, + configuration_settings={ + "csiDriverConfigs.local-csi-driver.enabled": True, + "csiDriverConfigs.azuresan-csi-driver.enabled": False, + } + ) + + with self._patch_k8s_extension_modules(extension): + result = acstor_helpers.get_extension_installed_and_cluster_configs( + Mock(cli_ctx=Mock()), "rg", "cluster" + ) + + self.assertEqual(result, (True, True, False)) + + class ErrorMappingTestCase(unittest.TestCase): def check_error_equality(self, mapped_error, mock_error): self.assertEqual(type(mapped_error), type(mock_error))