Fix auto-merge premature evaluation by waiting for all check suites t…

services/github/branches/get_required_status_checks.py

@@ -0,0 +1,45 @@
+# Standard imports
+from typing import cast
+
+# Third party imports
+import requests
+
+# Local imports
+from config import GITHUB_API_URL, TIMEOUT
+from services.github.types.branch_protection import BranchProtection
+from services.github.utils.create_headers import create_headers
+from utils.error.handle_exceptions import handle_exceptions
+
+
+@handle_exceptions(default_return_value=None, raise_on_error=False)
+def get_required_status_checks(owner: str, repo: str, branch: str, token: str):
+    """https://docs.github.com/en/rest/branches/branch-protection#get-branch-protection"""
+    url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/branches/{branch}/protection"
+    headers = create_headers(token=token)
+    response = requests.get(url=url, headers=headers, timeout=TIMEOUT)
+
+    # NOTE: 403 happens when GitHub App lacks "Administration: Read" permission
+    if response.status_code == 403:
+        print(f"No permission to read branch protection for {owner}/{repo}/{branch}")
+        return None
+
+    if response.status_code == 404:
+        print(f"No branch protection configured for {owner}/{repo}/{branch}")
+        return None
+
+    response.raise_for_status()
+    protection = cast(BranchProtection, response.json())
+    required_status_checks = protection.get("required_status_checks")
+
+    if not required_status_checks:
+        print(
+            f"Branch protection exists but no required status checks configured for {owner}/{repo}/{branch}"
+        )
+        return None
+
+    contexts = set(required_status_checks.get("contexts", []))
+    checks = {
+        check.get("context") for check in required_status_checks.get("checks", [])
+    }
+
+    return list(contexts | checks)

services/github/branches/test_get_required_status_checks.py

@@ -0,0 +1,211 @@
+# Standard imports
+from unittest.mock import MagicMock, patch
+
+# Third party imports
+import pytest
+import requests
+
+# Local imports
+from services.github.branches.get_required_status_checks import (
+    get_required_status_checks,
+)
+
+
+@pytest.fixture
+def mock_branch_protection_response():
+    return {
+        "required_status_checks": {
+            "strict": True,
+            "contexts": ["ci/circleci: test", "Codecov"],
+            "checks": [
+                {"context": "CircleCI Checks", "app_id": 12345},
+                {"context": "Aikido Security", "app_id": 67890},
+            ],
+        }
+    }
+
+
+def test_get_required_status_checks_success(
+    test_owner, test_repo, test_token, mock_branch_protection_response
+):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = mock_branch_protection_response
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        mock_get.assert_called_once_with(
+            url=f"https://api.github.com/repos/{test_owner}/{test_repo}/branches/main/protection",
+            headers={"Authorization": "Bearer test_token"},
+            timeout=120,
+        )
+        assert result is not None
+        assert set(result) == {
+            "ci/circleci: test",
+            "Codecov",
+            "CircleCI Checks",
+            "Aikido Security",
+        }
+
+
+def test_get_required_status_checks_403_no_permission(
+    test_owner, test_repo, test_token, capsys
+):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 403
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result is None
+        captured = capsys.readouterr()
+        assert "No permission to read branch protection" in captured.out
+
+
+def test_get_required_status_checks_404_no_protection(
+    test_owner, test_repo, test_token, capsys
+):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 404
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result is None
+        captured = capsys.readouterr()
+        assert "No branch protection configured" in captured.out
+
+
+def test_get_required_status_checks_no_required_checks(
+    test_owner, test_repo, test_token, capsys
+):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"required_status_checks": None}
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result is None
+        captured = capsys.readouterr()
+        assert "no required status checks configured" in captured.out
+
+
+def test_get_required_status_checks_only_contexts(test_owner, test_repo, test_token):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "required_status_checks": {
+                "strict": True,
+                "contexts": ["ci/circleci: test"],
+                "checks": [],
+            }
+        }
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result == ["ci/circleci: test"]
+
+
+def test_get_required_status_checks_only_checks(test_owner, test_repo, test_token):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "required_status_checks": {
+                "strict": False,
+                "contexts": [],
+                "checks": [{"context": "CircleCI Checks", "app_id": 12345}],
+            }
+        }
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result == ["CircleCI Checks"]
+
+
+def test_get_required_status_checks_http_error_500(test_owner, test_repo, test_token):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 500
+        http_error = requests.exceptions.HTTPError("500 Server Error")
+        http_error.response = mock_response
+        mock_response.raise_for_status.side_effect = http_error
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result is None
+
+
+def test_get_required_status_checks_network_error(test_owner, test_repo, test_token):
+    with patch(
+        "services.github.branches.get_required_status_checks.requests.get"
+    ) as mock_get, patch(
+        "services.github.branches.get_required_status_checks.create_headers"
+    ) as mock_headers:
+        mock_get.side_effect = requests.exceptions.ConnectionError("Network error")
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_required_status_checks(
+            owner=test_owner, repo=test_repo, branch="main", token=test_token
+        )
+
+        assert result is None

services/github/check_suites/get_check_suites.py

@@ -0,0 +1,20 @@
+# Third party imports
+import requests
+
+# Local imports
+from config import GITHUB_API_URL, TIMEOUT
+from services.github.types.check_suite import CheckSuite
+from services.github.utils.create_headers import create_headers
+from utils.error.handle_exceptions import handle_exceptions
+
+
+@handle_exceptions(default_return_value=None, raise_on_error=False)
+def get_check_suites(owner: str, repo: str, ref: str, token: str):
+    """https://docs.github.com/en/rest/checks/suites#list-check-suites-for-a-git-reference"""
+    url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/commits/{ref}/check-suites"
+    headers = create_headers(token=token)
+    response = requests.get(url=url, headers=headers, timeout=TIMEOUT)
+    response.raise_for_status()
+    data = response.json()
+    check_suites: list[CheckSuite] = data.get("check_suites", [])
+    return check_suites

services/github/check_suites/test_get_check_suites.py

@@ -0,0 +1,111 @@
+# Standard imports
+from unittest.mock import MagicMock, patch
+
+# Third party imports
+import pytest
+import requests
+
+# Local imports
+from services.github.check_suites.get_check_suites import get_check_suites
+
+
+@pytest.fixture
+def mock_check_suites_response():
+    return {
+        "total_count": 2,
+        "check_suites": [
+            {
+                "id": 53232653600,
+                "status": "completed",
+                "conclusion": "failure",
+                "app": {"name": "CircleCI Checks", "slug": "circleci-checks"},
+            },
+            {
+                "id": 53232653801,
+                "status": "completed",
+                "conclusion": "success",
+                "app": {"name": "Aikido PR Checks", "slug": "aikido"},
+            },
+        ],
+    }
+
+
+def test_get_check_suites_success(
+    test_owner, test_repo, test_token, mock_check_suites_response
+):
+    with patch(
+        "services.github.check_suites.get_check_suites.requests.get"
+    ) as mock_get, patch(
+        "services.github.check_suites.get_check_suites.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.json.return_value = mock_check_suites_response
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_check_suites(
+            owner=test_owner, repo=test_repo, ref="abc123", token=test_token
+        )
+
+        mock_get.assert_called_once_with(
+            url=f"https://api.github.com/repos/{test_owner}/{test_repo}/commits/abc123/check-suites",
+            headers={"Authorization": "Bearer test_token"},
+            timeout=120,
+        )
+        mock_response.raise_for_status.assert_called_once()
+        assert result == mock_check_suites_response["check_suites"]
+
+
+def test_get_check_suites_headers_creation(test_owner, test_repo):
+    with patch(
+        "services.github.check_suites.get_check_suites.requests.get"
+    ) as mock_get, patch(
+        "services.github.check_suites.get_check_suites.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"check_suites": []}
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer custom_token"}
+
+        get_check_suites(
+            owner=test_owner, repo=test_repo, ref="abc123", token="custom_token_123"
+        )
+
+        mock_headers.assert_called_once_with(token="custom_token_123")
+
+
+def test_get_check_suites_http_error_404(test_owner, test_repo, test_token):
+    with patch(
+        "services.github.check_suites.get_check_suites.requests.get"
+    ) as mock_get, patch(
+        "services.github.check_suites.get_check_suites.create_headers"
+    ) as mock_headers:
+        mock_response = MagicMock()
+        mock_response.status_code = 404
+        http_error = requests.exceptions.HTTPError("404 Client Error")
+        http_error.response = mock_response
+        mock_response.raise_for_status.side_effect = http_error
+        mock_get.return_value = mock_response
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_check_suites(
+            owner=test_owner, repo=test_repo, ref="nonexistent", token=test_token
+        )
+
+        assert result is None
+
+
+def test_get_check_suites_network_error(test_owner, test_repo, test_token):
+    with patch(
+        "services.github.check_suites.get_check_suites.requests.get"
+    ) as mock_get, patch(
+        "services.github.check_suites.get_check_suites.create_headers"
+    ) as mock_headers:
+        mock_get.side_effect = requests.exceptions.ConnectionError("Network error")
+        mock_headers.return_value = {"Authorization": "Bearer test_token"}
+
+        result = get_check_suites(
+            owner=test_owner, repo=test_repo, ref="abc123", token=test_token
+        )
+
+        assert result is None