Automerger support with Click

Signed-off-by: Petr "Stone" Hracek <[email protected]>

auto-merger

@@ -25,9 +25,44 @@
 #
 # Authors: Petr Hracek <[email protected]>
 
+import logging
+import sys
+
+import click
+
 from auto_merger.merger import AutoMerger
+from auto_merger.utils import setup_logger
+
+logger = logging.getLogger(__name__)
+
+
+@click.command()
+@click.option("-d", "--debug", is_flag=True, help="Enable debug logs")
+@click.option("--print-results", is_flag=True, help="Prints readable summary")
+@click.option("--github-labels", required=True, multiple=True,
+              help="Specify Git Hub labels to meet criteria")
+@click.option("--blocking-labels", multiple=True,
+              help="Specify Git Hub labels that blocks PR to merge")
+@click.option("--send-email", multiple=True, help="Specify email addresses to which the mail will be sent.")
+@click.option("--approvals",
+              default=2, type=int,
+              help="Specify number of approvals to automatically merge PR. Default 2")
+def auto_merger(debug, print_results, github_labels, blocking_labels, approvals, send_email):
+    am = AutoMerger(github_labels, blocking_labels,  approvals)
+    if debug:
+        setup_logger("auto-merger", level=logging.DEBUG)
+    else:
+        setup_logger("auto-merger", level=logging.INFO)
+    ret_value = am.check_all_containers()
+    if ret_value != 0:
+        sys.exit(2)
+    if print_results:
+        am.print_blocked_pull_request()
+        am.print_approval_pull_request()
+    if not am.send_results(send_email):
+        sys.exit(1)
+    sys.exit(ret_value)
 
 
 if __name__ == "__main__":
-    auto_merger = AutoMerger()
-    auto_merger.check_all_containers()
+    auto_merger()

auto_merger/constants.py

@@ -23,6 +23,7 @@
 # SOFTWARE.
 
 UPSTREAM_REPOS = [
+    "httpd-container",
     "s2i-base-container",
     "s2i-perl-container",
     "s2i-nodejs-container",
@@ -36,5 +37,4 @@
     "redis-container",
     "valkey-container",
     "varnish-container",
-    "httpd-container",
 ]

auto_merger/email.py

@@ -0,0 +1,52 @@
+# MIT License
+#
+# Copyright (c) 2024 Red Hat, Inc.
+
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import smtplib
+
+
+from email.mime.application import MIMEApplication
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from typing import List
+
+
+class EmailSender:
+
+    def __init__(self, recipient_email: List[str]):
+        self.recipient_email = recipient_email
+        self.mime_msg = MIMEMultipart()
+
+    def send_email(self, subject_msg, body: List[str]):
+        send_from = "[email protected]"
+        send_to = self.recipient_email
+        print(body)
+        msg = "<br>".join(body)
+        print(msg)
+        self.mime_msg["From"] = send_from
+        self.mime_msg["To"] = ", ".join(send_to)
+        self.mime_msg["Subject"] = subject_msg
+        self.mime_msg.attach(MIMEText(msg, "html"))
+        smtp = smtplib.SMTP("127.0.0.1")
+        smtp.sendmail(send_from, send_to, self.mime_msg.as_string())
+        smtp.close()
+        print("Sending email finished")
+

auto_merger/merger.py

@@ -34,17 +34,27 @@
 from auto_merger import utils
 from auto_merger.constants import UPSTREAM_REPOS
 from auto_merger.utils import setup_logger
+from auto_merger.email import EmailSender
 
 
 class AutoMerger:
     repo_data: List = []
-    pr_to_merge: List[int] = []
     container_name: str = ""
     container_dir: Path
     current_dir = os.getcwd()
 
-    def __init__(self):
+    def __init__(self, github_labels, blocking_labels, approvals=2):
         self.logger = setup_logger("AutoMerger")
+        self.github_labels = list(github_labels)
+        self.blocking_labels = list(blocking_labels)
+        self.approvals = approvals
+        self.logger.debug(f"GitHub Labels: {self.github_labels}")
+        self.logger.debug(f"GitHub Blocking Labels: {self.blocking_labels}")
+        self.logger.debug(f"Approvals Labels: {self.approvals}")
+        self.blocked_pr = {}
+        self.pr_to_merge = {}
+        self.blocked_body = []
+        self.approval_body = []
 
     def is_correct_repo(self) -> bool:
         cmd = ["gh repo view --json name"]
@@ -60,61 +70,117 @@ def get_gh_json_output(cmd):
         return json.loads(gh_repo_list)
 
     def get_gh_pr_list(self):
-        cmd = ["gh pr list -s open --json number,title,labels,reviews"]
-        self.repo_data = AutoMerger.get_gh_json_output(cmd=cmd)
+        cmd = ["gh pr list -s open --json number,title,labels,reviews,isDraft"]
+        repo_data = AutoMerger.get_gh_json_output(cmd=cmd)
+        for pr in repo_data:
+            if self.is_draft(pr):
+                continue
+            if self.is_changes_requested(pr):
+                continue
+            self.repo_data.append(pr)
 
-    def check_pr_labels(self, labels_to_check) -> bool:
-        self.logger.debug(f"Labels to check: {labels_to_check}")
-        if not labels_to_check:
+    def is_authenticated(self):
+        token = os.getenv("GH_TOKEN")
+        if token == "":
+            self.logger.error(f"Environment variable GH_TOKEN is not specified.")
+            return False
+        cmd = [f"gh status"]
+        self.logger.debug(f"Authentication command: {cmd}")
+        try:
+            return_output = utils.run_command(cmd=cmd, return_output=True)
+        except subprocess.CalledProcessError as cpe:
+            self.logger.error(f"Authentication to GitHub failed. {cpe}")
             return False
-        pr_failed_tags = ["pr/missing_review", "pr/failing-ci"]
-        pr_present = ["READY-to-MERGE"]
-        failed_pr = True
-        for label in labels_to_check:
-            if label["name"] in pr_failed_tags:
-                failed_pr = False
-            if label["name"] not in pr_present:
-                failed_pr = False
-        return failed_pr
-
-    def check_pr_approvals(self, reviews_to_check) -> bool:
+        return True
+
+    def add_blocked_pr(self, pr: {}):
+        present = False
+        for stored_pr in self.blocked_pr[self.container_name]:
+            if int(stored_pr["number"]) == int(pr["number"]):
+                present = True
+        if present:
+            return
+        self.blocked_pr[self.container_name].append({
+            "number": pr["number"],
+            "pr_dict": {
+                "title": pr["title"],
+                "labels": pr["labels"]
+            }
+        })
+        self.logger.debug(f"PR {pr['number']} added to blocked")
+
+    def add_approved_pr(self, pr: {}):
+        self.pr_to_merge[self.container_name].append({
+            "number": pr["number"],
+            "pr_dict": {
+                "title": pr["title"],
+                "labels": pr["labels"]
+            }
+        })
+        self.logger.debug(f"PR {pr['number']} added to approved")
+
+    def check_blocked_labels(self):
+        for pr in self.repo_data:
+            self.logger.debug(f"Check blocked: {pr}")
+            if "labels" not in pr:
+                continue
+            for label in pr["labels"]:
+                if label["name"] not in self.blocking_labels:
+                    continue
+                self.logger.debug(f"Add '{pr['number']}' to blocked PRs.")
+                self.add_blocked_pr(pr)
+
+    def check_labels_to_merge(self, pr):
+        if "labels" not in pr:
+            return True
+        for label in pr["labels"]:
+            if label["name"] in self.blocking_labels:
+                return False
+        self.logger.debug(f"Add '{pr['number']}' to approved PRs.")
+        return True
+
+    def check_pr_approvals(self, reviews_to_check) -> int:
         self.logger.debug(f"Approvals to check: {reviews_to_check}")
         if not reviews_to_check:
             return False
-        approval = "APPROVED"
         approval_cnt = 0
         for review in reviews_to_check:
-            if review["state"] == approval:
+            if review["state"] == "APPROVED":
                 approval_cnt += 1
         if approval_cnt < 2:
             self.logger.debug(f"Approval count: {approval_cnt}")
+        return approval_cnt
+
+    def is_changes_requested(self, pr):
+        if "labels" not in pr:
             return False
-        return True
+        for labels in pr["labels"]:
+            if "pr/changes-requested" == labels["name"]:
+                return True
+        return False
+
+    def is_draft(self, pr):
+        if pr['isDraft']:
+            return True
+        return False
 
     def check_pr_to_merge(self) -> bool:
         if len(self.repo_data) == 0:
             return False
-        pr_to_merge = []
         for pr in self.repo_data:
-            self.logger.debug(f"PR status: {pr}")
-            if "labels" not in pr:
+            if self.is_draft(pr):
                 continue
-            if not self.check_pr_labels(pr["labels"]):
-                self.logger.info(
-                    f"PR {pr['number']} does not have valid flag to merging in repo {self.container_name}."
-                )
-                continue
-            if not self.check_pr_approvals(pr["reviews"]):
-                self.logger.info(
-                    f"PR {pr['number']} does not have enought APPROVALS to merging in repo {self.container_name}."
-                )
+            self.logger.debug(f"PR status: {pr}")
+            if not self.check_labels_to_merge(pr):
                 continue
-            pr_to_merge.append(pr["number"])
-        self.logger.debug(f"PR to merge {pr_to_merge}")
-        if not pr_to_merge:
-            return False
-        self.pr_to_merge = pr_to_merge
-        return True
+            approval_count = self.check_pr_approvals(pr["reviews"])
+            self.pr_to_merge[self.container_name] = {
+                "number": pr["number"],
+                "approvals": approval_count,
+                "pr_dict": {
+                    "title": pr["title"]
+                }
+            }
 
     def clone_repo(self):
         temp_dir = utils.temporary_dir()
@@ -134,27 +200,86 @@ def clean_dirs(self):
         if self.container_dir.exists():
             shutil.rmtree(self.container_dir)
 
-    def check_all_containers(self):
+    def check_all_containers(self) -> int:
+        if not self.is_authenticated():
+            return 1
         for container in UPSTREAM_REPOS:
-            self.pr_to_merge = []
             self.container_name = container
+            self.repo_data = []
             self.clone_repo()
             if not self.is_correct_repo():
                 self.logger.error(f"This is not correct repo {self.container_name}.")
                 self.clean_dirs()
                 continue
+            if self.container_name not in self.blocked_pr:
+                self.blocked_pr[self.container_name] = []
+            if self.container_name not in self.pr_to_merge:
+                self.pr_to_merge[self.container_name] = []
             try:
                 self.get_gh_pr_list()
-                if self.check_pr_to_merge():
+                self.check_blocked_labels()
+                if len(self.blocked_pr[self.container_name]) != 0:
                     self.logger.info(
-                        f"This pull request can be merged {self.pr_to_merge}"
+                        f"This pull request can not be merged {self.pr_to_merge}"
                     )
-                    # auto_merger.merge_pull_requests()
+                self.check_pr_to_merge()
             except subprocess.CalledProcessError:
                 self.clean_dirs()
                 self.logger.error(f"Something went wrong {self.container_name}.")
                 continue
             self.clean_dirs()
+        return 0
+
+    def get_blocked_labels(self, pr_dict) -> List [str]:
+        labels = []
+        for lbl in pr_dict["labels"]:
+            labels.append(lbl["name"])
+        return labels
+
+    def print_blocked_pull_request(self):
+        # Do not print anything in case we do not have PR.
+        if not [x for x in self.blocked_pr if self.blocked_pr[x]]:
+            return 0
+        self.blocked_body.append(f"Pull requests that are blocked by labels [{', '.join(self.blocking_labels)}]")
+        for container, pull_requests in self.blocked_pr.items():
+            if not pull_requests:
+                continue
+            self.blocked_body.append(f"{container}:")
+            for pr in pull_requests:
+                blocked_labels = self.get_blocked_labels(pr["pr_dict"])
+                self.blocked_body.append(
+                    f"https://github.com/sclorg/{container}/pull/{pr['number']} - "
+                    f"[{pr['pr_dict']['title']}] -> {' '.join(blocked_labels)}"
+                )
+            self.blocked_body.extend([""])
+        self.blocked_body.extend(["", ""])
+        print('\n'.join(self.blocked_body))
+
+    def print_approval_pull_request(self):
+        # Do not print anything in case we do not have PR.
+        if not [x for x in self.pr_to_merge if self.pr_to_merge[x]]:
+            return 0
+        self.approval_body.append(f"Pull requests that can be merged or missing {self.approvals} approvals")
+        for container, pr in self.pr_to_merge.items():
+            if not pr:
+                continue
+            if int(pr["approvals"]) >= self.approvals:
+                result_pr = f" -> CAN BE MERGED"
+            else:
+                result_pr = f" -> Missing {self.approvals-int(pr['approvals'])} APPROVAL"
+            self.approval_body.append(
+                f"https://github.com/sclorg/{container}/pull/{pr['number']} - "
+                f"[{pr['pr_dict']['title']}]{result_pr}"
+            )
+        self.approval_body.extend(["", ""])
+        print('\n'.join(self.approval_body))
+
+    def send_results(self, recipients):
+        if not recipients:
+            return 1
+        sender_class = EmailSender(recipient_email=recipients)
+        subject_msg = "Pull request statuses for organization https://gibhub.com/sclorg"
+        sender_class.send_email(subject_msg, self.blocked_body + self.approval_body)
 
 
 def run():

requirements.txt

@@ -1,3 +1,4 @@
 pytest
 PyYAML
 flexmock
+click