Merge pull request #3 from sclorg/support_command_line_args

Automerger support with Click

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