feat: Add check runs status before merging PR

This commit introduces a check for the status of check runs
before merging a pull request. The PR will only be merged if
all required checks have passed. If any checks have failed or
are still pending, a comment will be posted on the PR with a
table of the failed or pending checks and the merge will be
aborted. This change ensures that only PRs with successful
checks are merged.

Signed-off-by: Chmouel Boudjnah <chmouel@redhat.com>

Author: chmouel

boussole/boussole.py

@@ -18,6 +18,7 @@
 
 from .messages import (  # isort:skip
     APPROVED_TEMPLATE,
+    CHECKS_NOT_PASSED,
     COMMENTS_FETCH_ERROR,
     HELP_TEXT,
     INSUFFICIENT_PERMISSIONS,
@@ -204,6 +205,48 @@ def _get_pr_status(self, number: int) -> RequestResponse:
         self._pr_status = self.api.get(endpoint)
         return self._pr_status
 
+    def _check_runs_status(self) -> Tuple[bool, List[Dict]]:
+        """
+        Checks if all check runs are successful.
+
+        Returns a tuple of (all_success, failed_checks).
+        """
+        endpoint = f"commits/{self._get_pr_status(self.pr_num).json()['head']['sha']}/check-runs"
+        headers = {"Accept": "application/vnd.github.v3+json"}
+        headers.update(self.api.headers)
+
+        response = self.api.get(endpoint)
+        if response.status_code != 200:
+            return False, []
+
+        check_runs = response.json()["check_runs"]
+
+        failed_checks = [
+            {
+                "name": check["name"],
+                "status": check["status"],
+                "conclusion": check["conclusion"],
+                "html_url": check["html_url"],
+            }
+            for check in check_runs
+            if check["status"] == "completed"
+            and check["conclusion"] not in ["success", "skipped"]
+        ]
+
+        pending_checks = [
+            {
+                "name": check["name"],
+                "status": check["status"],
+                "html_url": check.get("html_url", ""),
+            }
+            for check in check_runs
+            if check["status"] != "completed"
+            and not check["name"].endswith(" / boussole")
+        ]
+
+        all_success = not (failed_checks or pending_checks)
+        return all_success, failed_checks + pending_checks
+
     def check_status(self, num: int, status: str) -> bool:
         pr_status = self._get_pr_status(num)
         if pr_status.status_code != 200:
@@ -351,7 +394,7 @@ def lgtm(self, send_comment: bool = True) -> int:
 
     def merge_pr(self) -> bool:
         """
-        Merges the PR if it has enough LGTM approvals and performs cherry-picks.
+        Merges the PR if it has enough LGTM approvals and all checks are green.
         """
         # Check if the user has sufficient permissions to merge
         permission, is_valid = self._check_membership(self.comment_sender)
@@ -365,6 +408,27 @@ def merge_pr(self) -> bool:
             print(msg, file=sys.stderr)
             sys.exit(1)
 
+        # Check if all check runs are green
+        all_checks_passed, failed_checks = self._check_runs_status()
+        if not all_checks_passed:
+            status_table = "\n| Check Name | Status |\n|------------|--------|\n"
+            for check in failed_checks:
+                status = check.get("conclusion", check["status"])
+                check_name = check["name"]
+                # Add the URL link if available in the check data
+                if "html_url" in check:
+                    check_name = f"[{check['name']}]({check['html_url']})"
+                status_table += f"| {check_name} | `{status}` |\n"
+
+            msg = (
+                "⚠️ Cannot merge PR: Some checks are not passing.\n\n"
+                f"{status_table}\n\n"
+                "Please wait for all checks to pass before merging."
+            )
+            self._post_comment(CHECKS_NOT_PASSED.format(status_table=status_table))
+            print(msg, file=sys.stderr)
+            sys.exit(1)
+
         # Fetch LGTM votes and check if the threshold is met
         valid_votes, lgtm_users = self._fetch_and_validate_lgtm_votes()
 

boussole/messages.py

@@ -214,3 +214,11 @@
 appreciated.
 
 Thank you for your help! 🙌"""
+
+
+CHECKS_NOT_PASSED = """⚠️ Cannot merge PR: Some required checks haven't completed successfully.
+
+{status_table}
+
+🔍 Please ensure all checks pass before merging.
+💡 Tip: Review the failing checks above and address any issues."""

boussole/test_boussole.py

@@ -147,6 +147,41 @@ def test_check_membership_invalid_response(pr_handler, mock_api):
     assert is_valid is False
 
 
+def test_merge_pr_no_all_checks_succeed(capsys, pr_handler, mock_api):
+    all_checks = MyFakeResponse(
+        200,
+        {
+            "check_runs": [
+                {
+                    "name": "check-1",
+                    "status": "completed",
+                    "conclusion": "success",
+                    "html_url": "http://test.url",
+                },
+                {
+                    "name": "check-2",
+                    "status": "completed",
+                    "conclusion": "failure",
+                    "html_url": "http://test.url",
+                },
+            ],
+        },
+    )
+
+    mock_responses = [
+        MyFakeResponse(200, {"permission": "admin"}),
+        MyFakeResponse(200, {"head": {"sha": "abc123"}}),
+        all_checks,
+    ]
+
+    mock_api.get.side_effect = mock_responses
+
+    with pytest.raises(SystemExit) as exc_info:
+        pr_handler.merge_pr()
+        assert "Cannot merge PR" in capsys.err
+    assert exc_info.value.code == 1
+
+
 def test_merge_pr_success(pr_handler, mock_api):
     all_comments = MyFakeResponse(
         200,
@@ -156,8 +191,24 @@ def test_merge_pr_success(pr_handler, mock_api):
         ],
     )
 
+    all_checks = MyFakeResponse(
+        200,
+        {
+            "check_runs": [
+                {
+                    "name": "check-1",
+                    "status": "completed",
+                    "conclusion": "success",
+                    "html_url": "http://test.url",
+                },
+            ],
+        },
+    )
+
     mock_responses = [
         MyFakeResponse(200, {"permission": "admin"}),
+        MyFakeResponse(200, {"head": {"sha": "abc123"}}),
+        all_checks,
         all_comments,
         MyFakeResponse(200, {"permission": "write"}),  # reviewer1
         MyFakeResponse(200, {"permission": "write"}),  # reviewer2