[Containerapp] az containerapp sessionpool update: Fix updating registry credentials with --registry-username and --registry-password when existing registry credentials are empty

src/containerapp/HISTORY.rst

@@ -4,6 +4,7 @@ Release History
 ===============
 upcoming
 ++++++
+* 'az containerapp sessionpool update': Fix updating registry credentials with `--registry-username` and `--registry-password` when existing registry credentials are empty
 
 1.1.0b2
 ++++++

src/containerapp/azext_containerapp/_utils.py

@@ -103,15 +103,6 @@ def process_service(cmd, resource_list, service_name, arg_dict, subscription_id,
         raise ResourceNotFoundError("Service with the given name does not exist")
 
 
-def get_linker_client(cmd):
-    resource = cmd.cli_ctx.cloud.endpoints.active_directory_resource_id
-    profile = Profile(cli_ctx=cmd.cli_ctx)
-    credential, subscription_id, _ = profile.get_login_credentials(
-        subscription_id=get_subscription_id(cmd.cli_ctx), resource=resource)
-    linker_client = ServiceLinkerManagementClient(credential)
-    return linker_client
-
-
 def validate_binding_name(binding_name):
     pattern = r'^(?=.{1,60}$)[a-zA-Z0-9._]+$'
     return bool(re.match(pattern, binding_name))

src/containerapp/azext_containerapp/containerapp_sessionpool_decorator.py

@@ -487,14 +487,19 @@ def set_up_container(self, customer_container_template):
         return container_def
 
     def set_up_registry_auth_configuration(self, secrets_def, customer_container_template):
+        if self.has_registry_change():
+            if safe_get(customer_container_template, "registryCredentials") is None:
+                if self.get_argument_registry_server() is None or (self.get_argument_registry_user() is None or self.get_argument_registry_pass() is None):
+                    raise ValidationError("The existing registry credentials are empty. Please provide --registry-server, --registry-username, and --registry-password to update the registry credentials.")
+                safe_set(customer_container_template, "registryCredentials", value={})
         if self.get_argument_registry_server() is not None:
             safe_set(customer_container_template, "registryCredentials", "server", value=self.get_argument_registry_server())
         if self.get_argument_registry_user() is not None:
             safe_set(customer_container_template, "registryCredentials", "username", value=self.get_argument_registry_user())
         if secrets_def is None:
             secrets_def = []
         if self.get_argument_registry_pass() is not None:
-            original_secrets = self.existing_pool_def["properties"]["secrets"]
+            original_secrets = safe_get(self.existing_pool_def, "properties", "secrets", default=[])
             original_secrets_names = []
             for secret in original_secrets:
                 original_secrets_names.append(secret["name"])