Merge pull request based on the pr_lifetime

auto_merger/api.py

@@ -55,6 +55,7 @@ def merger(print_results, merger_labels, approvals, pr_lifetime, send_email) ->
         return ret_value
     if print_results:
         auto_merger.print_pull_request_to_merge()
+    auto_merger.merge_pull_requests()
     if not auto_merger.send_results(send_email):
         return 1
     return ret_value

auto_merger/cli/merger.py

@@ -38,7 +38,7 @@
 @click.option("--approvals",
               default=2, type=int,
               help="Specify number of approvals to automatically merge PR. Default 2")
-@click.option("--pr-lifetime", default=1, type=int, help="Specify a smallest time for which PR should opened")
+@click.option("--pr-lifetime", default=1, type=int, help="Specify a day for which PR should opened. Default is 1 day. To invalidate set 0")
 @pass_global_config
 def merger(ctx, print_results, merger_labels, approvals, pr_lifetime, send_email):
     logger.debug(ctx.debug)

auto_merger/merger.py

@@ -28,12 +28,15 @@
 import os
 import shutil
 
+from datetime import datetime, timedelta
 from typing import List
 from pathlib import Path
 
+import pytest
+
 from auto_merger import utils
 from auto_merger.constants import UPSTREAM_REPOS
-from auto_merger.utils import setup_logger
+from auto_merger.utils import setup_logger, cwd
 from auto_merger.email import EmailSender
 
 
@@ -45,12 +48,15 @@ class AutoMerger:
     def __init__(self, github_labels, approvals: int = 2, pr_lifetime: int = 1):
         self.logger = setup_logger("AutoMerger")
         self.approval_labels = list(github_labels)
+        self.pr_lifetime = pr_lifetime
         self.approvals = approvals
         self.logger.debug(f"GitHub Labels: {self.approval_labels}")
         self.logger.debug(f"Approvals Labels: {self.approvals}")
+        self.logger.debug(f"PR lifetime Labels: {self.pr_lifetime}")
         self.pr_to_merge = {}
         self.approval_body = []
         self.repo_data: List = []
+        self.temp_dir = ""
 
     def is_correct_repo(self) -> bool:
         cmd = ["gh repo view --json name"]
@@ -66,7 +72,7 @@ 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,isDraft"]
+        cmd = ["gh pr list -s open --json number,title,labels,reviews,isDraft,createdAt"]
         repo_data_output = AutoMerger.get_gh_json_output(cmd=cmd)
         for pr in repo_data_output:
             if AutoMerger.is_draft(pr):
@@ -94,7 +100,8 @@ def add_approved_pr(self, pr: {}):
             "number": pr["number"],
             "pr_dict": {
                 "title": pr["title"],
-                "labels": pr["labels"]
+                "labels": pr["labels"],
+                "createdAt": pr["createdAt"],
             }
         })
         self.logger.debug(f"PR {pr['number']} added to approved")
@@ -116,25 +123,42 @@ def check_pr_approvals(self, reviews_to_check) -> int:
         for review in reviews_to_check:
             if review["state"] == "APPROVED":
                 approval_cnt += 1
-        if approval_cnt < 2:
-            self.logger.debug(f"Approval count: {approval_cnt}")
+        if approval_cnt < self.approvals:
+            self.logger.debug(f"Not enough approvals: {approval_cnt}. Should be at least {self.approvals}")
         return approval_cnt
 
-    def is_changes_requested(self, pr):
+    @staticmethod
+    def is_changes_requested(pr):
         if "labels" not in pr:
             return False
         for labels in pr["labels"]:
             if "pr/changes-requested" == labels["name"]:
                 return True
         return False
 
+    @staticmethod
+    def get_realtime():
+        from datetime import datetime
+        return datetime.now()
+
     @staticmethod
     def is_draft(pull_request):
         if "isDraft" in pull_request:
             if pull_request["isDraft"] in ["True", "true"]:
                 return True
         return False
 
+    def check_pr_lifetime(self, pr: dict) -> bool:
+        if self.pr_lifetime == 0:
+            return True
+        if "createdAt" not in pr:
+            return False
+        pr_life = pr["createdAt"]
+        date_created = datetime.strptime(pr_life, "%Y-%m-%dT%H:%M:%SZ") + timedelta(days=1)
+        if date_created < AutoMerger.get_realtime():
+            return True
+        return False
+
     def check_pr_to_merge(self) -> bool:
         if len(self.repo_data) == 0:
             return False
@@ -147,32 +171,49 @@ def check_pr_to_merge(self) -> bool:
             if "reviews" not in pr:
                 continue
             approval_count = self.check_pr_approvals(pr["reviews"])
+            if not self.check_pr_lifetime(pr=pr):
+                continue
             self.pr_to_merge[self.container_name] = {
                 "number": pr["number"],
                 "approvals": approval_count,
                 "pr_dict": {
-                    "title": pr["title"]
+                    "title": pr["title"],
                 }
             }
 
     def clone_repo(self):
-        temp_dir = utils.temporary_dir()
+        self.temp_dir = utils.temporary_dir()
         utils.run_command(
-            f"gh repo clone https://github.com/sclorg/{self.container_name} {temp_dir}/{self.container_name}"
+            f"gh repo clone https://github.com/sclorg/{self.container_name} {self.temp_dir}/{self.container_name}"
         )
-        self.container_dir = Path(temp_dir) / f"{self.container_name}"
+        self.container_dir = Path(self.temp_dir) / f"{self.container_name}"
         if self.container_dir.exists():
             os.chdir(self.container_dir)
 
     def merge_pull_requests(self):
-        for pr in self.pr_to_merge:
-            self.logger.debug(f"PR to merge {pr} in repo {self.container_name}.")
+        for container in UPSTREAM_REPOS:
+            self.container_name = container
+            self.container_dir = Path(self.temp_dir) / f"{self.container_name}"
+            with cwd(self.container_dir) as _:
+                self.merge_pr()
+            self.clean_dirs()
+
 
     def clean_dirs(self):
         os.chdir(self.current_dir)
         if self.container_dir.exists():
             shutil.rmtree(self.container_dir)
 
+    def merge_pr(self):
+        for pr in self.pr_to_merge[self.container_name]:
+            self.logger.info(f"Let's try to merge {pr['number']}....")
+            try:
+                output = utils.run_command(f"gh pr merge {pr['number']}", return_output=True)
+                self.logger.debug(f"The output from merging command '{output}'")
+            except subprocess.CalledProcessError as cpe:
+                self.logger.error(f"Merging pr {pr} failed with reason {cpe.output}")
+                continue
+
     def check_all_containers(self) -> int:
         if not self.is_authenticated():
             return 1
@@ -193,10 +234,8 @@ def check_all_containers(self) -> int:
                 self.clean_dirs()
                 self.logger.error(f"Something went wrong {self.container_name}.")
                 continue
-            self.clean_dirs()
         return 0
 
-
     def print_pull_request_to_merge(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]]:
@@ -224,14 +263,9 @@ def print_pull_request_to_merge(self):
         print('\n'.join(self.approval_body))
 
     def send_results(self, recipients):
-        self.logger.debug(f"Recepients are: {recipients}")
+        self.logger.debug(f"Recipients are: {recipients}")
         if not recipients:
             return 1
         sender_class = EmailSender(recipient_email=list(recipients))
         subject_msg = "Pull request statuses for organization https://gibhub.com/sclorg"
         sender_class.send_email(subject_msg, self.approval_body)
-
-
-def run():
-    auto_merger = AutoMerger()
-    auto_merger.check_all_containers()

auto_merger/pr_checker.py

@@ -289,8 +289,3 @@ def send_results(self, recipients):
         sender_class = EmailSender(recipient_email=list(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():
-    auto_merger = PRStatusChecker()
-    auto_merger.check_all_containers()

auto_merger/utils.py

@@ -24,6 +24,10 @@
 import logging
 import tempfile
 import sys
+import os
+
+from pathlib import Path
+from contextlib import contextmanager
 
 logger = logging.getLogger(__name__)
 
@@ -99,3 +103,19 @@ def setup_logger(logger_id, level=logging.DEBUG):
     stderr.setFormatter(formatter)
     logger.addHandler(stderr)
     return logger
+
+
+@contextmanager
+def cwd(path):
+    """
+    Switch to Path directory and once action is done
+    returns back
+    :param path:
+    :return:
+    """
+    prev_cwd = Path.cwd()
+    os.chdir(path)
+    try:
+        yield
+    finally:
+        os.chdir(prev_cwd)

tests/test_merger.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+
+import pytest
+
+from datetime import datetime
+from flexmock import flexmock
+
+from auto_merger.merger import AutoMerger
+
+
+@pytest.mark.parametrize(
+    "pr_created_date,pr_lifetime,return_code",
+    (
+            ("2024-12-20T09:30:11Z", 1, False),
+            ("2024-12-19T07:30:11Z", 1, True),
+            ("2024-12-21T09:30:11Z", 1, False),
+            ("2024-12-21T09:30:11Z", 0, True),
+    )
+)
+def test_get_gh_pr_list(pr_created_date, pr_lifetime, return_code):
+    set_time = datetime.strptime("2024-12-20T10:35:20Z", '%Y-%m-%dT%H:%M:%SZ')
+    flexmock(AutoMerger).should_receive("get_realtime").and_return(set_time)
+    auto_merger = AutoMerger(github_labels=["READY-to-MERGE"], approvals=1, pr_lifetime=pr_lifetime)
+    pr = {
+        "createdAt": pr_created_date,
+    }
+    assert auto_merger.check_pr_lifetime(pr) == return_code