Merge pull request #119 from druids/INFRA-137

Infra 137

Author: matllubos

developers_chamber/bin/pydev.py

@@ -57,7 +57,9 @@
 # Import external scripts
 for base_path in (Path.home(), Path.cwd()):
     if (base_path / ".pydev" / "scripts" / "__init__.py").exists():
-        SourceFileLoader('*', str(base_path / ".pydev" / "scripts" / "__init__.py")).load_module()
+        SourceFileLoader(
+            "*", str(base_path / ".pydev" / "scripts" / "__init__.py")
+        ).load_module()
 
 from developers_chamber.scripts.init_aliasses import *
 

developers_chamber/click/alias.py

@@ -0,0 +1,125 @@
+import re
+import shlex
+from gettext import gettext as _
+
+import click
+from click.formatting import wrap_text
+
+
+def find_and_replace_command_variable(arg, command, index):
+    match = re.match(r"^--(?P<arg_name>[^=\ ]+)[\ =](?P<arg_value>.+)", arg)
+    if match:
+        arg_name, arg_value = match.groups()
+        if f"${arg_name}" in command:
+            return True, command.replace(f"${arg_name}", arg_value)
+        elif f'${arg_name.replace("-", "_")}' in command:
+            return True, command.replace(f'${arg_name.replace("-", "_")}', arg_value)
+    if f"${index}" in command:
+        return True, command.replace(f"${index}", arg)
+    return False, command
+
+
+def parse_alias(alias):
+    from developers_chamber.scripts import cli
+
+    command_parts = list(shlex.split(alias))
+    used_command_parts = []
+
+    click_command = cli
+    while command_parts:
+        command_part = command_parts[0]
+        current_click_command = click_command.get_command(None, command_part)
+        if current_click_command is None:
+            raise click.ClickException(
+                'Command to the alias "{}" cannot be found'.format(alias)
+            )
+
+        used_command_parts.append(command_parts.pop(0))
+        click_command = current_click_command
+        if not isinstance(current_click_command, click.Group):
+            break
+    return used_command_parts, command_parts, click_command, alias
+
+
+class AliasCommand(click.Command):
+
+    def __init__(self, commands, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._parse_alias(commands)
+
+    def _parse_alias(self, commands):
+        if isinstance(commands, str):
+            commands = [commands]
+            description = ""
+        elif isinstance(commands, list):
+            commands = commands
+            description = ""
+        elif isinstance(commands, dict):
+            description = commands["description"]
+            commands = commands["command"]
+        else:
+            raise click.ClickException("Invalid alias type")
+
+        commands = [commands] if isinstance(commands, str) else commands
+        self.aliases = [(parse_alias(alias)) for alias in commands]
+        self.short_help = description
+
+    def format_help_text(self, ctx, formatter) -> None:
+        formatter.write_paragraph()
+        with formatter.indentation():
+            if self.short_help:
+                formatter.write_text(self.short_help)
+                formatter.write_paragraph()
+            formatter.write_text("Alias to commands: ")
+            with formatter.indentation():
+                for (
+                    used_command_parts,
+                    remaining_command_parts,
+                    click_command,
+                    alias_str,
+                ) in self.aliases:
+                    formatter.write_text("* " + " ".join(used_command_parts))
+                    if remaining_command_parts:
+                        with formatter.indentation(), formatter.indentation():
+                            formatter.write_text(" ".join(remaining_command_parts))
+
+    def format_epilog(self, ctx, formatter) -> None:
+        with formatter.section(_("Commands in alias help")):
+            for (
+                used_command_parts,
+                remaining_command_parts,
+                click_command,
+                alias_str,
+            ) in self.aliases:
+                with formatter.indentation():
+                    formatter.write_paragraph()
+                    formatter.write_text("* " + " ".join(used_command_parts))
+                    with formatter.indentation(), formatter.indentation():
+                        click_command.format_help_text(ctx, formatter)
+                        click_command.format_options(ctx, formatter)
+                        formatter.write_paragraph()
+
+    def invoke(self, ctx):
+        from developers_chamber.scripts import cli
+
+        for i, (
+            used_command_parts,
+            remaining_command_parts,
+            click_command,
+            alias_str,
+        ) in enumerate(self.aliases):
+            if i == 0:
+                alias_args = []
+
+                for index, arg in enumerate(ctx.args, start=1):
+                    replaced_arg, alias_str = find_and_replace_command_variable(
+                        arg, alias_str, index
+                    )
+                    if not replaced_arg:
+                        alias_args.append(arg)
+
+                if "$@" in alias_str:
+                    alias_str = alias_str.replace("$@", " ".join(ctx.args))
+                    alias_args = []
+
+            cli.main(args=shlex.split(alias_str) + alias_args, standalone_mode=False)

developers_chamber/git_utils.py

@@ -2,7 +2,7 @@
 
 import git
 from click import BadParameter, UsageError
-from git import GitCommandError
+from git import GitCommandError, InvalidGitRepositoryError
 
 from .types import ReleaseType, VersionFileType
 from .version_utils import bump_to_next_version, bump_version, get_version
@@ -31,7 +31,9 @@ def create_release_branch(version, release_type, remote_name=None, branch_name=N
     return release_branch_name
 
 
-def create_release(version_file, release_type, remote_name=None, branch_name=None, file_type=None):
+def create_release(
+    version_file, release_type, remote_name=None, branch_name=None, file_type=None
+):
     repo = git.Repo(".")
     g = repo.git
 
@@ -214,3 +216,19 @@ def get_current_issue_key():
         return match.group("issue_key")
     else:
         return None
+
+
+def get_remote_url():
+    try:
+        repo = git.Repo(".")
+        return f"https://{repo.remotes.origin.url.split('@')[1].split(':')[0]}"
+    except InvalidGitRepositoryError:
+        raise UsageError("Git repository not found in the current directory")
+
+
+def get_remote_path():
+    try:
+        repo = git.Repo(".")
+        return repo.remotes.origin.url.split("@")[1].split(":")[1]
+    except InvalidGitRepositoryError:
+        raise UsageError("Git repository not found in the current directory")

developers_chamber/gitlab_utils.py

@@ -3,13 +3,22 @@
 import time
 import requests
 from click import UsageError
+from urllib.parse import quote
 
 
 def create_merge_request(
-    api_url, token, title, description, source_branch, target_branch, project, assignee_id=None, automerge=False
+    url,
+    token,
+    title,
+    description,
+    source_branch,
+    target_branch,
+    project,
+    assignee_id=None,
+    automerge=False,
 ):
     response = requests.post(
-        f"{api_url}/projects/{quote_plus(project)}/merge_requests",
+        f"{url}/api/v4/projects/{quote_plus(project)}/merge_requests",
         headers={
             "PRIVATE-TOKEN": token,
         },
@@ -18,42 +27,38 @@ def create_merge_request(
             "target_branch": target_branch,
             "title": title,
             "description": description,
-            "assignee_id": assignee_id
+            "assignee_id": assignee_id,
         },
     )
 
     if response.status_code != 201:
         raise UsageError(f'GitLab error: {response.content.decode("utf-8")}')
 
     if automerge:
-        time.sleep(5)
-        message = activate_automerge(api_url, token, project, response.json()['iid'])
+        message = activate_automerge(url, token, project, response.json()["iid"])
         return f"{response.json()['web_url']} ({message})"
 
-    return response.json()['web_url']
+    return response.json()["web_url"]
 
 
-def activate_automerge(
-    api_url, token, project, merge_request_id
-):
-    merge_response = requests.put(
-        f"{api_url}/projects/{quote_plus(project)}/merge_requests/{merge_request_id}/merge",
-        json={"merge_when_pipeline_succeeds": True},
-        headers={
-            "PRIVATE-TOKEN": token,
-        },
-    )
-    if merge_response.status_code != 200:
-        return "Automerge activation failed"
-    else:
-        return "Automerge activated"
+def activate_automerge(url, token, project, merge_request_id, retries=5):
+    for _ in range(retries):
+        merge_response = requests.put(
+            f"{url}/api/v4/projects/{quote_plus(project)}/merge_requests/{merge_request_id}/merge",
+            json={"merge_when_pipeline_succeeds": True},
+            headers={
+                "PRIVATE-TOKEN": token,
+            },
+        )
+        if merge_response.status_code == 200:
+            return "Automerge activated"
+        time.sleep(5)
+    return "Automerge activation failed"
 
 
-def run_job(
-    api_url, token, project, ref, variables
-):
+def run_job(url, token, project, ref, variables):
     response = requests.post(
-        f"{api_url}/projects/{quote_plus(project)}/pipeline",
+        f"{url}/api/v4/projects/{quote_plus(project)}/pipeline",
         headers={
             "PRIVATE-TOKEN": token,
         },
@@ -66,10 +71,24 @@ def run_job(
                     "variable_type": "env_var",
                 }
                 for key, value in variables.items()
-            ]
+            ],
         },
     )
     if response.status_code != 201:
         raise UsageError(f'GitLab error: {response.content.decode("utf-8")}')
     else:
         return response.json()["web_url"]
+
+
+def get_project_id(url, project, token):
+    response = requests.get(
+        f"{url}/api/v4/projects/{quote_plus(project)}",
+        headers={
+            "PRIVATE-TOKEN": token,
+            "Content-type": "application/json",
+        },
+    )
+    if response.status_code != 200:
+        raise UsageError(f'GitLab error: {response.content.decode("utf-8")}')
+    else:
+        return response.json()["id"]

developers_chamber/scripts/__init__.py

@@ -1,6 +1,49 @@
 import click
+from click.formatting import HelpFormatter
+from gettext import gettext as _
+from developers_chamber.click.alias import AliasCommand
 
 
-@click.group()
+class FullHelpGroup(click.Group):
+    def format_commands(self, ctx, formatter) -> None:
+        """Extra format methods for multi methods that adds all the commands
+        after the options.
+        """
+        all_commands = []
+        for subcommand in self.list_commands(ctx):
+            cmd = self.get_command(ctx, subcommand)
+            # What is this, the tool lied about a command.  Ignore it
+            if cmd is None:
+                continue
+            if cmd.hidden:
+                continue
+
+            all_commands.append((subcommand, cmd))
+
+        alias_commands = [
+            command for command in all_commands if isinstance(command[1], AliasCommand)
+        ]
+        commands = [
+            command for command in all_commands if command not in alias_commands
+        ]
+
+        limit = formatter.width - 6 - max(len(cmd[0]) for cmd in all_commands)
+
+        for commands, label in [
+            (alias_commands, _("Aliases")),
+            (commands, _("Commands")),
+        ]:
+            if len(commands):
+                rows = []
+                for subcommand, cmd in commands:
+                    help = cmd.get_short_help_str(limit)
+                    rows.append((subcommand, help))
+
+                if rows:
+                    with formatter.section(label):
+                        formatter.write_dl(rows)
+
+
+@click.group(cls=FullHelpGroup)
 def cli():
     pass

developers_chamber/scripts/git.py

@@ -48,9 +48,9 @@ def git():
     "--file",
     "-f",
     help="path to the version file",
-    type=str,
     default=default_version_files[0],
     required=True,
+    type=click.Path(exists=True),
 )
 @click.option(
     "--remote-name",
@@ -96,9 +96,9 @@ def create_release_branch(release_type, file, remote_name, branch_name):
     "--file",
     "-f",
     help="path to the version file",
-    type=str,
     default=default_version_files[0],
     required=True,
+    type=click.Path(exists=True),
 )
 @click.option(
     "--remote-name",
@@ -119,7 +119,7 @@ def create_release_branch(release_type, file, remote_name, branch_name):
     help="version file type",
     type=EnumType(VersionFileType),
     default=default_version_file_type,
-    required=False
+    required=False,
 )
 def create_release(release_type, file, remote_name, branch_name, file_type):
     """
@@ -170,10 +170,10 @@ def checkout_to_release_branch():
     "--file",
     "-f",
     help="path to the version file",
-    type=str,
     default=default_version_files,
     required=True,
     multiple=True,
+    type=click.Path(exists=True),
 )
 def bump_version_from_release_tag(file):
     """
@@ -189,10 +189,10 @@ def bump_version_from_release_tag(file):
     "--file",
     "-f",
     help="path to the version file",
-    type=str,
     default=default_version_files,
     required=True,
     multiple=True,
+    type=click.Path(exists=True),
 )
 @click.option(
     "--remote-name",