[Dapr] Disable applying CRDs during a downgrade

src/k8s-extension/azext_k8s_extension/partner_extensions/Dapr.py

@@ -10,10 +10,11 @@
 from typing import Tuple
 
 from azure.cli.core.azclierror import InvalidArgumentValueError
+from copy import deepcopy
 from knack.log import get_logger
 from knack.prompting import prompt, prompt_y_n
 
-from ..vendored_sdks.models import Extension, Scope, ScopeCluster
+from ..vendored_sdks.models import Extension, PatchExtension, Scope, ScopeCluster
 from .DefaultExtension import DefaultExtension
 
 logger = get_logger(__name__)
@@ -144,3 +145,39 @@ def Create(self, cmd, client, resource_group_name: str, cluster_name: str, name:
             location=""
         )
         return extension_instance, release_name, create_identity
+
+    def Update(self, cmd, resource_group_name: str, cluster_name: str, auto_upgrade_minor_version: bool,
+               release_train: str, version: str, configuration_settings: dict,
+               configuration_protected_settings: dict, original_extension: Extension, yes: bool = False) \
+            -> PatchExtension:
+        """ExtensionType 'Microsoft.Dapr' specific validations & defaults for Update.
+           Must create and return a valid 'PatchExtension' object.
+        """
+        input_configuration_settings = deepcopy(configuration_settings)
+
+        # configuration_settings can be None, so we need to set it to an empty dict.
+        if configuration_settings is None:
+            configuration_settings = {}
+
+        # If we are downgrading the extension, then we need to disable the apply-CRDs hook.
+        # This is because CRD updates while downgrading can cause issues.
+        original_version = original_extension.version
+        if original_version and version and version < original_version:
+            logger.debug("Downgrade detected from %s to %s. Setting hooks.applyCrds to false.",
+                         original_version, version)
+            configuration_settings['hooks.applyCrds'] = 'false'
+        else:
+            # If we are not downgrading, set the hooks.applyCrds to true.
+            # This is because the value may have been set to false during a previous downgrade.
+            logger.debug("No downgrade detected. Setting hooks.applyCrds to true.")
+            configuration_settings['hooks.applyCrds'] = 'true'
+
+        # If no changes were made, return the original dict (empty or None).
+        if len(configuration_settings) == 0:
+            configuration_settings = input_configuration_settings
+
+        return PatchExtension(auto_upgrade_minor_version=auto_upgrade_minor_version,
+                              release_train=release_train,
+                              version=version,
+                              configuration_settings=configuration_settings,
+                              configuration_protected_settings=configuration_protected_settings)