Fixed --registry-server with --image bug. (#78)

* Fixed --registry-server with --image bug.

* Fixed style issues.

Co-authored-by: Haroon Feisal <[email protected]>

src/containerapp/azext_containerapp/_up_utils.py

@@ -2,7 +2,7 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
-# pylint: disable=line-too-long, consider-using-f-string, no-else-return, duplicate-string-formatting-argument, expression-not-assigned, too-many-locals, logging-fstring-interpolation, arguments-differ, abstract-method, logging-format-interpolation
+# pylint: disable=line-too-long, consider-using-f-string, no-else-return, duplicate-string-formatting-argument, expression-not-assigned, too-many-locals, logging-fstring-interpolation, arguments-differ, abstract-method, logging-format-interpolation, broad-except
 
 
 from urllib.parse import urlparse
@@ -278,12 +278,11 @@ def create(self, no_registry=False):
 
     def create_acr_if_needed(self):
         if self.should_create_acr:
-             logger.warning(
-                 f"Creating Azure Container Registry {self.acr.name} in resource group "
-                 f"{self.acr.resource_group.name}"
-             )
-             self.create_acr()
-
+            logger.warning(
+                f"Creating Azure Container Registry {self.acr.name} in resource group "
+                f"{self.acr.resource_group.name}"
+            )
+            self.create_acr()
 
     def create_acr(self):
         registry_rg = self.resource_group
@@ -386,12 +385,12 @@ def _get_dockerfile_content_from_repo(
     repo = repo_url_to_name(repo_url)
     try:
         r = g.get_repo(repo)
-    except:
-        raise ValidationError(f"Could not find repo {repo_url}")
+    except Exception as e:
+        raise ValidationError(f"Could not find repo {repo_url}") from e
     try:
         files = r.get_contents(context_path, ref=branch)
-    except:
-        raise ValidationError(f"Could not find branch {branch}")
+    except Exception as e:
+        raise ValidationError(f"Could not find branch {branch}") from e
     for f in files:
         if f.path == dockerfile or f.path.endswith(f"/{dockerfile}"):
             resp = requests.get(f.download_url)
@@ -606,10 +605,8 @@ def _get_default_registry_name(app):
     return f"ca{registry_name}acr"  # ACR names must start + end in a letter
 
 
-def _set_acr_creds(cmd, app:"ContainerApp", registry_name):
-    logger.info(
-                "No credential was provided to access Azure Container Registry. Trying to look up..."
-            )
+def _set_acr_creds(cmd, app: "ContainerApp", registry_name):
+    logger.info("No credential was provided to access Azure Container Registry. Trying to look up...")
     try:
         app.registry_user, app.registry_pass, registry_rg = _get_acr_cred(
             cmd.cli_ctx, registry_name
@@ -677,7 +674,7 @@ def _set_up_defaults(
         else:
             env_list = [e for e in list_managed_environments(cmd=cmd) if e["name"] == env.name and e["location"] == location]
         if len(env_list) == 1:
-            resource_group.name =  parse_resource_id(env_list[0]["id"])["resource_group"]
+            resource_group.name = parse_resource_id(env_list[0]["id"])["resource_group"]
 
     # get ACR details from --image, if possible
     _get_acr_from_image(cmd, app)
@@ -712,9 +709,9 @@ def _create_github_action(
 
     # need to trigger the workflow manually if it already exists (performing an update)
     try:
-        source_control_info = GitHubActionClient.show(cmd=app.cmd, resource_group_name=app.resource_group.name, name=app.name)
+        GitHubActionClient.show(cmd=app.cmd, resource_group_name=app.resource_group.name, name=app.name)
         trigger_workflow(token, repo, app.name, branch)
-    except Exception as ex:
+    except:  # pylint: disable=bare-except
         pass
 
     create_or_update_github_action(
@@ -760,7 +757,6 @@ def up_output(app):
 
 
 def find_existing_acr(cmd, app: "ContainerApp"):
-    from azure.cli.command_modules.acr.custom import acr_show
     from azure.cli.command_modules.acr._client_factory import cf_acr_registries
     client = cf_acr_registries(cmd.cli_ctx)
 

src/containerapp/azext_containerapp/_utils.py

@@ -168,7 +168,7 @@ def get_github_repo(token, repo):
     return g.get_repo(repo)
 
 
-def get_workflow(github_repo, name):
+def get_workflow(github_repo, name):  # pylint: disable=inconsistent-return-statements
     workflows = github_repo.get_workflows()
     for wf in workflows:
         if wf.path.startswith(f".github/workflows/{name}") and "Trigger auto deployment for containerapp" in wf.name:
@@ -199,7 +199,7 @@ def await_github_action(cmd, token, repo, branch, name, resource_group_name, tim
             raise CLIInternalError("Timed out while waiting for the Github action to be created.")
         animation.flush()
     if gh_action_status == "Failed":
-            raise CLIInternalError("The Github Action creation failed.")  # TODO ask backend team for a status url / message
+        raise CLIInternalError("The Github Action creation failed.")  # TODO ask backend team for a status url / message
 
     workflow = None
     while workflow is None:

src/containerapp/azext_containerapp/custom.py

@@ -2155,6 +2155,13 @@ def containerapp_up_logic(cmd, resource_group_name, name, managed_env, image, en
     registries_def = containerapp_def["properties"]["configuration"]["registries"]
 
     if registry_server:
+        if not registry_pass or not registry_user:
+            if '.azurecr.io' not in registry_server:
+                raise RequiredArgumentMissingError('Registry url is required if using Azure Container Registry, otherwise Registry username and password are required if using Dockerhub')
+            logger.warning('No credential was provided to access Azure Container Registry. Trying to look up...')
+            parsed = urlparse(registry_server)
+            registry_name = (parsed.netloc if parsed.scheme else parsed.path).split('.')[0]
+            registry_user, registry_pass, _ = _get_acr_cred(cmd.cli_ctx, registry_name)
         # Check if updating existing registry
         updating_existing_registry = False
         for r in registries_def: