From 850731b47813bc2d91f326f5c2ad2bdd04b804dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bj=C3=B6rn=20Ricks?= Date: Mon, 17 Jul 2023 16:34:12 +0200 Subject: [PATCH] Change: Refactor mattermost_notify * Allow to pass an explicit commit * Split into several modules * Add more unit tests --- mattermost_notify/git.py | 117 +++++++------------------- mattermost_notify/parser.py | 69 ++++++++++++++++ mattermost_notify/status.py | 15 ++++ tests/test_git.py | 161 ++++++++++++++++++++++++------------ tests/test_parser.py | 113 +++++++++++++++++++++++++ 5 files changed, 338 insertions(+), 137 deletions(-) create mode 100644 mattermost_notify/parser.py create mode 100644 mattermost_notify/status.py create mode 100644 tests/test_parser.py diff --git a/mattermost_notify/git.py b/mattermost_notify/git.py index 3b9158e..d2d70bc 100644 --- a/mattermost_notify/git.py +++ b/mattermost_notify/git.py @@ -1,27 +1,19 @@ -# Copyright (C) 2022 Jaspar Stach +# 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: diff --git a/mattermost_notify/parser.py b/mattermost_notify/parser.py new file mode 100644 index 0000000..bb4914b --- /dev/null +++ b/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) diff --git a/mattermost_notify/status.py b/mattermost_notify/status.py new file mode 100644 index 0000000..45f7ad1 --- /dev/null +++ b/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 diff --git a/tests/test_git.py b/tests/test_git.py index 94dfed6..6d916e4 100644 --- a/tests/test_git.py +++ b/tests/test_git.py @@ -1,68 +1,127 @@ -# Copyright (C) 2022 Jaspar Stach +# SPDX-FileCopyrightText: 2022-2023 Greenbone AG +# +# SPDX-License-Identifier: GPL-3.0-or-later + +# ruff: noqa: E501 import unittest from unittest.mock import MagicMock, patch -from mattermost_notify.git import _linker as linker -from mattermost_notify.git import fill_template, parse_args +from mattermost_notify.git import fill_template, linker +from mattermost_notify.status import Status -class GitNotifyTestCase(unittest.TestCase): +class LinkerTestCase(unittest.TestCase): def test_linker(self): - expected = "[foo](www.foo.com)" - link = linker("foo", "www.foo.com") + self.assertEqual(linker("foo", "www.foo.com"), "[foo](www.foo.com)") + + def test_no_url(self): + self.assertEqual(linker("foo"), "foo") + self.assertEqual(linker("foo", None), "foo") + self.assertEqual(linker("foo", ""), "foo") - self.assertEqual(link, expected) - def test_argument_parsing(self): - parsed_args = parse_args(["www.url.de", "channel"]) +class FillTemplateTestCase(unittest.TestCase): + def test_success_no_highlight(self): + actual = fill_template( + highlight=["user1", "user2"], + status=Status.SUCCESS.name, + terminal=MagicMock(), + ) + expected = """#### Status: :white_check_mark: success - self.assertEqual(parsed_args.url, "www.url.de") - self.assertEqual(parsed_args.channel, "channel") +| Workflow | | +| --- | --- | +| Repository (branch) | ([](/tree/None)) | +| Related commit | [](/commit/) | - def test_fail_argument_parsing(self): - with self.assertRaises(SystemExit): - parse_args(["-s"]) +""" + self.assertEqual(expected, actual) - @patch("mattermost_notify.git.get_github_event_json") - def test_no_highlight(self, event_mock: None): - event_mock.return_value = None - parsed_args = parse_args( - ["www.url.de", "channel", "--highlight", "user1", "user2"] + def test_failure_highlight(self): + actual = fill_template( + highlight=["user1", "user2"], + status=Status.FAILURE.name, + terminal=MagicMock(), + ) + expected = """#### Status: :x: failure + +| Workflow | | +| --- | --- | +| Repository (branch) | ([](/tree/None)) | +| Related commit | [](/commit/) | + +@user1 +@user2 +""" + self.assertEqual(expected, actual) + + def test_short_template(self): + actual = fill_template( + short=True, + status=Status.SUCCESS.name, + workflow_name="SomeWorkflow", + workflow_id="w1", + commit="12345", + repository="foo/bar", + branch="main", + terminal=MagicMock(), ) - rf = fill_template(parsed_args, MagicMock()) - rt = ( - "#### Status: :white_check_mark: success\n\n" - "| Workflow | [None](https://githu" - "b.com/None/actions/runs/None) |\n| --- | --- |" - "\n| Repository (branch) | [None](https://githu" - "b.com/None) (None) |\n| Related commit | not a" - "vailable |\n\n" + expected = ( + ":white_check_mark: success: [SomeWorkflow](https://github.com/foo/bar/actions/runs/w1) " + "([12345](https://github.com/foo/bar/commit/12345)) in " + "[foo/bar](https://github.com/foo/bar) ([main](https://github.com/foo/bar/tree/main))" ) - print(rf) - self.assertEqual(rf, rt) + self.assertEqual(expected, actual) - @patch("mattermost_notify.git.get_github_event_json") - def test_highlight(self, event_mock: None): - event_mock.return_value = None - parsed_args = parse_args( - [ - "www.url.de", - "channel", - "--highlight", - "user1", - "user2", - "-S", - "failure", - ] + def test_template(self): + actual = fill_template( + short=False, + status=Status.SUCCESS.name, + workflow_name="SomeWorkflow", + workflow_id="w1", + commit="12345", + repository="foo/bar", + branch="main", + terminal=MagicMock(), ) - rf = fill_template(parsed_args, MagicMock()) - rt = ( - "#### Status: :x: failure\n\n" - "| Workflow | [None](https://githu" - "b.com/None/actions/runs/None) |\n| --- | --- |" - "\n| Repository (branch) | [None](https://githu" - "b.com/None) (None) |\n| Related commit | not a" - "vailable |\n\n@user1\n@user2\n" + expected = """#### Status: :white_check_mark: success + +| Workflow | [SomeWorkflow](https://github.com/foo/bar/actions/runs/w1) | +| --- | --- | +| Repository (branch) | [foo/bar](https://github.com/foo/bar) ([main](https://github.com/foo/bar/tree/main)) | +| Related commit | [12345](https://github.com/foo/bar/commit/12345) | + +""" + self.assertEqual(expected, actual) + + @patch("mattermost_notify.git.get_github_event_json") + def test_template_data_from_github_event(self, mock: MagicMock): + event = { + "workflow_run": { + "conclusion": Status.SUCCESS.name, + "name": "SomeWorkflow", + "head_repository": { + "full_name": "foo/bar", + "html_url": "https://github.com/foo/bar", + }, + "head_branch": "main", + "head_commit": "12345", + "workflow_id": "w1", + } + } + mock.return_value = event + + actual = fill_template( + short=False, + terminal=MagicMock(), ) - self.assertEqual(rf, rt) + expected = """#### Status: :white_check_mark: success + +| Workflow | [SomeWorkflow](https://github.com/foo/bar/actions/runs/w1) | +| --- | --- | +| Repository (branch) | [foo/bar](https://github.com/foo/bar) ([main](https://github.com/foo/bar/tree/main)) | +| Related commit | [12345](https://github.com/foo/bar/commit/12345) | + +""" + self.assertEqual(expected, actual) diff --git a/tests/test_parser.py b/tests/test_parser.py new file mode 100644 index 0000000..6af64fb --- /dev/null +++ b/tests/test_parser.py @@ -0,0 +1,113 @@ +# SPDX-FileCopyrightText: 2022-2023 Greenbone AG +# +# SPDX-License-Identifier: GPL-3.0-or-later + +import unittest + +from mattermost_notify.parser import parse_args + + +class ParseArgsTestCase(unittest.TestCase): + def test_defaults(self): + parsed_args = parse_args(["www.url.de", "channel"]) + + self.assertEqual(parsed_args.url, "www.url.de") + self.assertEqual(parsed_args.channel, "channel") + + self.assertFalse(parsed_args.short) + self.assertEqual(parsed_args.status, "SUCCESS") + self.assertIsNone(parsed_args.repository) + self.assertIsNone(parsed_args.branch) + self.assertIsNone(parsed_args.workflow) + self.assertIsNone(parsed_args.workflow_name) + self.assertIsNone(parsed_args.commit) + self.assertIsNone(parsed_args.free) + self.assertIsNone(parsed_args.highlight) + + def test_fail_argument_parsing(self): + with self.assertRaises(SystemExit): + parse_args(["-s"]) + + def test_parse_status(self): + parsed_args = parse_args( + [ + "www.url.de", + "channel", + "-S", + "failure", + ] + ) + self.assertEqual(parsed_args.status, "failure") + + def test_parse_short(self): + parsed_args = parse_args( + [ + "www.url.de", + "channel", + "--short", + ] + ) + self.assertTrue(parsed_args.short) + + parsed_args = parse_args( + [ + "www.url.de", + "channel", + "-s", + ] + ) + self.assertTrue(parsed_args.short) + + def test_parse_repository(self): + parsed_args = parse_args(["www.url.de", "channel", "-r", "foo/bar"]) + self.assertEqual(parsed_args.repository, "foo/bar") + + parsed_args = parse_args( + ["www.url.de", "channel", "--repository", "foo/bar"] + ) + self.assertEqual(parsed_args.repository, "foo/bar") + + def test_parse_branch(self): + parsed_args = parse_args(["www.url.de", "channel", "-b", "foo"]) + self.assertEqual(parsed_args.branch, "foo") + + parsed_args = parse_args(["www.url.de", "channel", "--branch", "foo"]) + self.assertEqual(parsed_args.branch, "foo") + + def test_parse_workflow(self): + parsed_args = parse_args(["www.url.de", "channel", "-w", "w1"]) + self.assertEqual(parsed_args.workflow, "w1") + + parsed_args = parse_args(["www.url.de", "channel", "--workflow", "w1"]) + self.assertEqual(parsed_args.workflow, "w1") + + def test_parse_workflow_name(self): + parsed_args = parse_args(["www.url.de", "channel", "-n", "foo"]) + self.assertEqual(parsed_args.workflow_name, "foo") + + parsed_args = parse_args( + ["www.url.de", "channel", "--workflow_name", "foo"] + ) + self.assertEqual(parsed_args.workflow_name, "foo") + + def test_parse_commit(self): + parsed_args = parse_args(["www.url.de", "channel", "--commit", "1234"]) + self.assertEqual(parsed_args.commit, "1234") + + def test_parse_free(self): + parsed_args = parse_args( + ["www.url.de", "channel", "--free", "lorem ipsum"] + ) + self.assertEqual(parsed_args.free, "lorem ipsum") + + def test_parse_highlight(self): + parsed_args = parse_args( + [ + "www.url.de", + "channel", + "--highlight", + "user1", + "user2", + ] + ) + self.assertEqual(parsed_args.highlight, ["user1", "user2"])