Change: Refactor mattermost_notify

* Allow to pass an explicit commit
* Split into several modules
* Add more unit tests

mattermost_notify/git.py

@@ -1,27 +1,19 @@
-# Copyright (C) 2022 Jaspar Stach <jasp.stac@gmx.de>
+# SPDX-FileCopyrightText: 2022-2023 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
 
 # pylint: disable=invalid-name
 
 import json
 import os
-from argparse import ArgumentParser, Namespace
-from enum import Enum
 from pathlib import Path
 from typing import Any, Optional
 
 import httpx
 from pontos.terminal.terminal import ConsoleTerminal
 
-
-class Status(Enum):
-    SUCCESS = ":white_check_mark: success"
-    FAILURE = ":x: failure"
-    UNKNOWN = ":grey_question: unknown"
-    CANCELLED = ":no_entry_sign: canceled"
-
-    def __str__(self):
-        return self.name
-
+from mattermost_notify.parser import parse_args
+from mattermost_notify.status import Status
 
 LONG_TEMPLATE = (
     "#### Status: {status}\n\n"
@@ -32,7 +24,7 @@ def __str__(self):
     "{highlight}"
 )
 
-SHORT_TEMPLATE = "{status}: {workflow} ({commit})  in {repository} ({branch})"
+SHORT_TEMPLATE = "{status}: {workflow} ({commit}) in {repository} ({branch})"
 
 DEFAULT_GIT = "https://github.com"
 
@@ -73,9 +65,7 @@ def fill_template(
     workflow_name: Optional[str] = None,
     terminal: ConsoleTerminal,
 ) -> str:
-    template = LONG_TEMPLATE
-    if short:
-        template = SHORT_TEMPLATE
+    template = SHORT_TEMPLATE if short else LONG_TEMPLATE
 
     # try to get information from the GiTHUB_EVENT json
     event = get_github_event_json(terminal)
@@ -84,9 +74,12 @@ def fill_template(
     status = status if status else workflow_info.get("conclusion")
     workflow_status = Status[status.upper()] if status else Status.UNKNOWN
 
-    workflow_name: str = (
+    used_workflow_name: str = (
         workflow_name if workflow_name else workflow_info.get("name", "")
     )
+    used_workflow_id = (
+        workflow_id if workflow_id else workflow_info.get("workflow_id", "")
+    )
 
     head_repo: dict[str, Any] = workflow_info.get("head_repository", {})
     repository = repository if repository else head_repo.get("full_name", "")
@@ -96,11 +89,13 @@ def fill_template(
         else head_repo.get("html_url", "")
     )
 
-    branch: str = branch if branch else workflow_info.get("head_branch", "")
+    used_branch: str = (
+        branch if branch else workflow_info.get("head_branch", "")
+    )
     branch_url = f"{repository_url}/tree/{branch}"
 
     workflow_url = (
-        f"{repository}/actions/runs/{workflow_id}"
+        f"{repository_url}/actions/runs/{used_workflow_id}"
         if repository
         else workflow_info.get("html_url", "")
     )
@@ -111,89 +106,39 @@ def fill_template(
     else:
         head_commit = workflow_info.get("head_commit", {})
         commit_url = f'{repository_url}/commit/{head_commit.get("id", "")}'
-        commit_message: str = head_commit["message"].split("\n", 1)[0]
+        commit_message: str = head_commit.get("message", "").split("\n", 1)[0]
 
     highlight_str = ""
-    if highlight and status is not Status.SUCCESS:
+    if highlight and workflow_status is not Status.SUCCESS:
         highlight_str = "".join([f"@{h}\n" for h in highlight])
 
     return template.format(
         status=workflow_status.value,
-        workflow=linker(workflow_name, workflow_url),
+        workflow=linker(used_workflow_name, workflow_url),
         repository=linker(repository, repository_url),
-        branch=linker(branch, branch_url),
+        branch=linker(used_branch, branch_url),
         commit=linker(commit_message, commit_url),
         highlight=highlight_str,
     )
 
 
-def parse_args(args=None) -> Namespace:
-    parser = ArgumentParser(prog="mnotify-git")
-
-    parser.add_argument(
-        "url",
-        help="Mattermost (WEBHOOK) URL",
-        type=str,
-    )
-
-    parser.add_argument(
-        "channel",
-        type=str,
-        help="Mattermost Channel",
-    )
-
-    parser.add_argument(
-        "-s",
-        "--short",
-        action="store_true",
-        help="Send a short single line message",
-    )
-
-    parser.add_argument(
-        "-S",
-        "--status",
-        type=str,
-        choices=["success", "failure"],
-        default=Status.SUCCESS.name,
-        help="Status of Job",
-    )
-
-    parser.add_argument(
-        "-r", "--repository", type=str, help="git repository name (orga/repo)"
-    )
-
-    parser.add_argument("-b", "--branch", type=str, help="git branch")
-
-    parser.add_argument(
-        "-w", "--workflow", type=str, help="hash/ID of the workflow"
-    )
-
-    parser.add_argument(
-        "-n", "--workflow_name", type=str, help="name of the workflow"
-    )
-
-    parser.add_argument(
-        "--free",
-        type=str,
-        help="Print a free-text message to the given channel",
-    )
-
-    parser.add_argument(
-        "--highlight",
-        nargs="+",
-        help="List of persons to highlight in the channel",
-    )
-
-    return parser.parse_args(args=args)
-
-
 def main() -> None:
-    parsed_args: Namespace = parse_args()
+    parsed_args = parse_args()
 
     term = ConsoleTerminal()
 
     if not parsed_args.free:
-        body = fill_template(args=parsed_args, term=term)
+        body = fill_template(
+            highlight=parsed_args.highlight,
+            short=parsed_args.short,
+            branch=parsed_args.branch,
+            commit=parsed_args.commit,
+            repository=parsed_args.repository,
+            status=parsed_args.status,
+            workflow_id=parsed_args.workflow,
+            workflow_name=parsed_args.workflow_name,
+            terminal=term,
+        )
 
         data = {"channel": parsed_args.channel, "text": body}
     else:

mattermost_notify/parser.py

@@ -0,0 +1,69 @@
+# SPDX-FileCopyrightText: 2023 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from argparse import ArgumentParser, Namespace
+
+from mattermost_notify.status import Status
+
+
+def parse_args(args=None) -> Namespace:
+    parser = ArgumentParser()
+
+    parser.add_argument(
+        "url",
+        help="Mattermost (WEBHOOK) URL",
+        type=str,
+    )
+
+    parser.add_argument(
+        "channel",
+        type=str,
+        help="Mattermost Channel",
+    )
+
+    parser.add_argument(
+        "-s",
+        "--short",
+        action="store_true",
+        help="Send a short single line message",
+    )
+
+    parser.add_argument(
+        "-S",
+        "--status",
+        type=str,
+        choices=["success", "failure"],
+        default=Status.SUCCESS.name,
+        help="Status of Job",
+    )
+
+    parser.add_argument(
+        "-r", "--repository", type=str, help="git repository name (orga/repo)"
+    )
+
+    parser.add_argument("-b", "--branch", type=str, help="git branch")
+
+    parser.add_argument(
+        "-w", "--workflow", type=str, help="hash/ID of the workflow"
+    )
+
+    parser.add_argument(
+        "-n", "--workflow_name", type=str, help="name of the workflow"
+    )
+
+    parser.add_argument("--commit", help="Commit to use")
+
+    parser.add_argument(
+        "--free",
+        type=str,
+        help="Print a free-text message to the given channel",
+    )
+
+    parser.add_argument(
+        "--highlight",
+        nargs="+",
+        help="List of persons to highlight in the channel",
+    )
+
+    return parser.parse_args(args=args)

mattermost_notify/status.py

@@ -0,0 +1,15 @@
+# SPDX-FileCopyrightText: 2022-2023 Greenbone AG
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from enum import Enum
+
+
+class Status(Enum):
+    SUCCESS = ":white_check_mark: success"
+    FAILURE = ":x: failure"
+    UNKNOWN = ":grey_question: unknown"
+    CANCELLED = ":no_entry_sign: canceled"
+
+    def __str__(self):
+        return self.name