ENH: implement deploy-time tagging and nitpick outputs more

scripts/ioc_deploy.py

@@ -389,6 +389,7 @@ def get_deploy_info(args: CliArgs) -> DeployInfo:
             name=name,
             github_org=args.github_org,
             release=release,
+            auto_confirm=args.auto_confirm,
             verbose=args.verbose,
         )
 
@@ -553,7 +554,7 @@ def casing_from_text(uncased: str, casing_source: str) -> str:
     return casing_source[index : index + len(uncased)]
 
 
-def finalize_tag(name: str, github_org: str, release: str, verbose: bool) -> str:
+def finalize_tag(name: str, github_org: str, release: str, auto_confirm: bool, verbose: bool) -> str:
     """
     Check if release is present in the org.
 
@@ -565,6 +566,8 @@ def finalize_tag(name: str, github_org: str, release: str, verbose: bool) -> str
     - 1.0.0
     """
     logger.debug(f"Getting all tags in {github_org}/{name}")
+    if not release:
+        raise ValueError("Recieved empty string as release name")
     try:
         tags = get_repo_tags(
             name=name,
@@ -580,7 +583,66 @@ def finalize_tag(name: str, github_org: str, release: str, verbose: bool) -> str
         if rel in tags:
             logger.info(f"Release {rel} exists in {github_org}/{name}")
             return rel
-    raise ValueError(f"Unable to find {release} in {github_org}/{name}")
+
+    logger.warning(f"Unable to find {release} in {github_org}/{name}")
+    if release[0] == "R":
+        suggested_tag = release
+    elif release[0].isdigit():
+        suggested_tag = f"R{release}"
+    else:
+        suggested_tag = f"R{release[1:]}"
+
+    if not auto_confirm:
+        user_text = input(f"Create a tag named {suggested_tag}? yes/true or no/false\n")
+        if not is_yes(user_text, error_on_empty=True):
+            raise ValueError(f"Unable to find {release} in {github_org}/{name}")
+
+    logger.info(f"Creating a tag named {suggested_tag}")
+
+    with TemporaryDirectory() as tmpdir:
+        logger.info(f"Cloning {github_org}/{name}")
+        try:
+            _clone(
+                name=name, github_org=github_org, working_dir=tmpdir, verbose=verbose
+            )
+        except subprocess.CalledProcessError as exc:
+            raise ValueError(
+                f"Error cloning {github_org}/{name}, please make sure you have the correct access rights and the repository exists."
+            ) from exc
+        cloned_dir = str(Path(tmpdir) / name)
+        tag_msg = ""
+        if not auto_confirm:
+            # Best effort to get context for the commit and show the default message
+            # I'm not too bothered if this fails somehow, I'd rather keep going
+            try:
+                last_commit = get_last_commit_info(working_dir=cloned_dir)
+            except subprocess.CalledProcessError:
+                ...
+            else:
+                # No commit message = not an annotated commit = displays the linked commit's message
+                print()
+                print("The default message comes from the most recent commit, which is:")
+                print()
+                print(last_commit.strip())
+                print()
+            print("(Optional) if you'd like, you may type a different tag message.")
+            print("For multiline messages in git, the first line is a summary of the message.")
+            print("End with ctrl+D on blank line.")
+            print()
+            while True:
+                try:
+                    tag_msg += input() + "\n"
+                except EOFError:
+                    break
+        # Raise errors from these without modifying the error message
+        logger.info(f"Creating tag {suggested_tag}")
+        _tag(release=suggested_tag, message=tag_msg.strip(), working_dir=cloned_dir, verbose=verbose)
+        logger.info("Pushing tag to GitHub")
+        _push_tag(release=suggested_tag, working_dir=cloned_dir, verbose=verbose)
+
+    logger.info(f"{suggested_tag} created and pushed")
+    logger.info("Remember to create a GitHub release later!")
+    return suggested_tag
 
 
 def release_permutations(release: str) -> List[str]:
@@ -757,6 +819,74 @@ def _clone(
     return subprocess.run(cmd, **kwds)
 
 
+def _tag(
+    release: str,
+    message: str = "",
+    working_dir: str = "",
+    verbose: bool = False,
+) -> subprocess.CompletedProcess:
+    """
+    Create a tag on a github repo or raise a subprocess.CalledProcessError
+
+    Allow a release message but do not require it.
+    Avoid opening the editor dialog because this breaks the script.
+    """
+    cmd = ["git", "tag", release]
+    if message:
+        cmd.extend(["-a", "-m", message])
+    kwds = {"check": True}
+    if working_dir:
+        kwds["cwd"] = working_dir
+    if not verbose:
+        kwds["stdout"] = subprocess.PIPE
+        kwds["stderr"] = subprocess.PIPE
+    logger.debug(f"Calling '{' '.join(cmd)}' with kwargs {kwds}")
+    return subprocess.run(cmd, **kwds)
+
+
+def _push_tag(
+    release: str,
+    working_dir: str = "",
+    verbose: bool = False,
+) -> subprocess.CompletedProcess:
+    """
+    Push a tag to github or raise a subprocess.CalledProcessError
+
+    Relies on their being an existing git clone with origin set
+    appropriately and a local tag matching "release".
+    """
+    cmd = ["git", "push", "origin", release]
+    kwds = {"check": True}
+    if working_dir:
+        kwds["cwd"] = working_dir
+    if not verbose:
+        kwds["stdout"] = subprocess.PIPE
+        kwds["stderr"] = subprocess.PIPE
+    logger.debug(f"Calling '{' '.join(cmd)}' with kwargs {kwds}")
+    return subprocess.run(cmd, **kwds)
+
+
+def get_last_commit_info(
+    working_dir: str = "",
+    verbose: bool = False,
+) -> str:
+    """
+    Return the most recent commit message or raise a subprocess.CalledProcessError
+    """
+    cmd = ["git", "log", "-1"]
+    kwds = {
+        "check": True,
+        "stdout": subprocess.PIPE,
+        "universal_newlines": True,
+    }
+    if working_dir:
+        kwds["cwd"] = working_dir
+    if not verbose:
+        kwds["stderr"] = subprocess.PIPE
+    logger.debug(f"Calling '{' '.join(cmd)}' with kwargs {kwds}")
+    return subprocess.run(cmd, **kwds).stdout
+
+
 def get_github_available(verbose: bool = False) -> bool:
     """
     Return whether or not github is available.
@@ -797,6 +927,7 @@ def _ping(
     last_exc = None
     for _ in range(tries):
         try:
+            logger.debug(f"Calling '{' '.join(cmd)}' with kwargs {kwds}")
             proc = subprocess.run(cmd, **kwds)
         except subprocess.CalledProcessError as exc:
             last_exc = exc
@@ -859,6 +990,7 @@ def _ls_remote(
     if not verbose:
         kwds["stderr"] = subprocess.PIPE
     output = []
+    logger.debug(f"Calling '{' '.join(cmd)}' with kwargs {kwds}")
     with subprocess.Popen(cmd, **kwds) as proc:
         for line in proc.stdout:
             if verbose:
@@ -955,13 +1087,17 @@ def _main() -> int:
         logger.error(exc)
         logger.debug("Traceback", exc_info=True)
         rval = ReturnCode.EXCEPTION
+    except KeyboardInterrupt:
+        logger.error("Interruped with ctrl+C")
+        logger.debug("Traceback", exc_info=True)
+        rval = ReturnCode.EXCEPTION
 
     if rval == ReturnCode.SUCCESS:
         logger.info("ioc-deploy completed successfully")
     elif rval == ReturnCode.EXCEPTION:
         logger.error("ioc-deploy errored out")
     elif rval == ReturnCode.NO_CONFIRM:
-        logger.info("ioc-deploy cancelled")
+        logger.warning("ioc-deploy cancelled")
     return rval