[AKS] Fix ACStor configuration setting detection

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

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))