Validate auth token on init with dataverse

Author: nheeb

src/hermes/commands/init/base.py

@@ -196,6 +196,7 @@ def __init__(self, parser: argparse.ArgumentParser):
         self.tokens: dict = {}
         self.setup_method: str = ""
         self.deposit_platform: DepositPlatform = DepositPlatform()
+        self.git_branch: str = ""
         self.git_remote: str = ""
         self.git_remote_url = ""
         self.git_hoster: GitHoster = GitHoster.Empty
@@ -344,7 +345,8 @@ def test_initialization(self) -> None:
             self.no_git_setup()
             sys.exit()
 
-        # Look at git remotes
+        # Look at git branch & remotes
+        self.git_branch = git_info.get_current_branch()
         remotes = git_info.get_remotes()
         if remotes:
             self.git_remote = remotes[0]
@@ -544,7 +546,7 @@ def create_zenodo_token(self) -> None:
                     else:
                         sc.echo(f"The token could not be validated by {self.deposit_platform.name}. "
                                 "Make sure to enter the complete token.\n"
-                                "(If this error persists, you should try switching to the manual setup mode.)",
+                                "(If this error persists, you should restart and switch to the manual setup mode.)",
                                 formatting=sc.Formats.WARNING)
 
     def create_dataverse_token(self):
@@ -556,8 +558,20 @@ def create_dataverse_token(self):
         if self.setup_method == "m":
             sc.press_enter_to_continue()
         else:
-            # TODO try to validate the token
-            self.deposit_platform.token = sc.answer("Enter the token here: ")
+            while True:
+                token = sc.answer("Enter the token here: ")
+                token_valid_url = f"{self.deposit_platform.url}/api/users/token"
+                token_valid_response = requests.get(token_valid_url, headers={"X-Dataverse-key": token})
+                if token_valid_response.ok:
+                    sc.echo(f"The token was validated by {self.deposit_platform.name}.",
+                            formatting=sc.Formats.OKGREEN)
+                    self.deposit_platform.token = token
+                    break
+                else:
+                    sc.echo(f"The token could not be validated by {self.deposit_platform.name}. "
+                            "Make sure to enter the complete token.\n"
+                            "(If this error persists, you should restart and switch to the manual setup mode.)",
+                            formatting=sc.Formats.WARNING)
 
     def create_rodare_token(self):
         token_url = urljoin(self.deposit_platform.url, "account/settings/applications/tokens/new/")
@@ -807,7 +821,8 @@ def choose_push_branch(self) -> None:
         push_choice = sc.choose(
             "When should the automated HERMES process start?",
             [
-                "When I push a branch",
+                "When I push on custom branch",
+                f"When I push on current branch ({self.git_branch})",
                 "When I push a specific tag (not implemented)",
             ]
         )
@@ -818,6 +833,12 @@ def choose_push_branch(self) -> None:
                     formatting=sc.Formats.OKGREEN)
             sc.echo()
         elif push_choice == 1:
+            self.ci_parameters["push_branch"] = self.git_branch
+            bold_branch = sc.Formats.BOLD.wrap_around(self.git_branch)
+            sc.echo(f"The HERMES pipeline will be activated when you push on {bold_branch}",
+                    formatting=sc.Formats.OKGREEN)
+            sc.echo()
+        elif push_choice == 2:
             sc.echo("Setting up triggering by tags is currently not implemented.", formatting=sc.Formats.WARNING)
             sc.echo(f"You can visit {TUTORIAL_URL} to set it up manually later-on.", formatting=sc.Formats.WARNING)
 

src/hermes/commands/init/util/git_info.py

@@ -30,7 +30,7 @@ def get_valid_cwd(cwd="") -> str:
     return str(path)
 
 
-def run_git_command(command: str, cwd="") -> str:
+def run_git_command(command: str, cwd="", throw_exception=False) -> str:
     """
     Runs any git command using subprocess. Raises Exception when returncode != 0.
     :param command: The command as string with or without the 'git' main command.
@@ -44,7 +44,7 @@ def run_git_command(command: str, cwd="") -> str:
     if command_list[0] != "git":
         command_list.insert(0, "git")
     # Run subprocess
-    result = subprocess.run(command_list, cwd=cwd, capture_output=True, text=True)
+    result = subprocess.run(command_list, cwd=cwd, capture_output=True, text=True, check=throw_exception)
     # Return output or error
     if result.returncode != 0:
         raise Exception(result.stderr)
@@ -87,11 +87,11 @@ def get_current_branch() -> str:
     """
     Returns the name of the current branch.
     """
-    branch_info = run_git_command("branch")
-    for line in branch_info.splitlines():
-        if line.startswith("*"):
-            return line.split()[1].strip()
-    raise Exception("Current branch not found.")
+    try:
+        branch_name = run_git_command("rev-parse --abbrev-ref HEAD", throw_exception=True)
+        return branch_name.strip()
+    except subprocess.CalledProcessError:
+        return ""
 
 
 def is_git_installed() -> bool: