fear: support github enterprise api

Author: ricardojdsilva87

auth.py

@@ -25,32 +25,31 @@ def auth_to_github(
         github3.GitHub: the GitHub connection object
     """
     if gh_app_id and gh_app_private_key_bytes and gh_app_installation_id:
-        gh = github3.github.GitHub()
-        gh.login_as_app_installation(
-            gh_app_private_key_bytes, gh_app_id, gh_app_installation_id
-        )
+        if ghe:
+            gh = github3.github.GitHubEnterprise(url=ghe)
+        else:
+            gh = github3.github.GitHub()
+        gh.login_as_app_installation(gh_app_private_key_bytes, gh_app_id, gh_app_installation_id)
         github_connection = gh
     elif ghe and token:
-        github_connection = github3.github.GitHubEnterprise(ghe, token=token)
+        github_connection = github3.github.GitHubEnterprise(url=ghe, token=token)
     elif token:
         github_connection = github3.login(token=token)
     else:
-        raise ValueError(
-            "GH_TOKEN or the set of [GH_APP_ID, GH_APP_INSTALLATION_ID, GH_APP_PRIVATE_KEY] environment variables are not set"
-        )
+        raise ValueError("GH_TOKEN or the set of [GH_APP_ID, GH_APP_INSTALLATION_ID, GH_APP_PRIVATE_KEY] environment variables are not set")
 
     if not github_connection:
         raise ValueError("Unable to authenticate to GitHub")
     return github_connection  # type: ignore
 
 
-def get_github_app_installation_token(
-    gh_app_id: str, gh_app_private_key_bytes: bytes, gh_app_installation_id: str
-) -> str | None:
+def get_github_app_installation_token(ghe: str, gh_app_id: str, gh_app_private_key_bytes: bytes, gh_app_installation_id: str) -> str | None:
     """
     Get a GitHub App Installation token.
+    API: https://docs.github.com/en/apps/creating-github-apps/authenticating-with-a-github-app/authenticating-as-a-github-app-installation
 
     Args:
+        ghe (str): the GitHub Enterprise endpoint
         gh_app_id (str): the GitHub App ID
         gh_app_private_key_bytes (bytes): the GitHub App Private Key
         gh_app_installation_id (str): the GitHub App Installation ID
@@ -59,7 +58,8 @@ def get_github_app_installation_token(
         str: the GitHub App token
     """
     jwt_headers = github3.apps.create_jwt_headers(gh_app_private_key_bytes, gh_app_id)
-    url = f"https://api.github.com/app/installations/{gh_app_installation_id}/access_tokens"
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/app/installations/{gh_app_installation_id}/access_tokens"
 
     try:
         response = requests.post(url, headers=jwt_headers, json=None, timeout=5)

evergreen.py

@@ -45,28 +45,20 @@ def main():  # pragma: no cover
     ) = env.get_env_vars()
 
     # Auth to GitHub.com or GHE
-    github_connection = auth.auth_to_github(
-        token, gh_app_id, gh_app_installation_id, gh_app_private_key, ghe
-    )
+    github_connection = auth.auth_to_github(token, gh_app_id, gh_app_installation_id, gh_app_private_key, ghe)
 
     if not token and gh_app_id and gh_app_installation_id and gh_app_private_key:
-        token = auth.get_github_app_installation_token(
-            gh_app_id, gh_app_private_key, gh_app_installation_id
-        )
+        token = auth.get_github_app_installation_token(ghe, gh_app_id, gh_app_private_key, gh_app_installation_id)
 
     # If Project ID is set, lookup the global project ID
     if project_id:
         # Check Organization is set as it is required for linking to a project
         if not organization:
-            raise ValueError(
-                "ORGANIZATION environment variable was not set. Please set it"
-            )
-        project_id = get_global_project_id(token, organization, project_id)
+            raise ValueError("ORGANIZATION environment variable was not set. Please set it")
+        project_id = get_global_project_id(ghe, token, organization, project_id)
 
     # Get the repositories from the organization, team name, or list of repositories
-    repos = get_repos_iterator(
-        organization, team_name, repository_list, github_connection
-    )
+    repos = get_repos_iterator(organization, team_name, repository_list, github_connection)
 
     # Iterate through the repositories and open an issue/PR if dependabot is not enabled
     count_eligible = 0
@@ -78,13 +70,13 @@ def main():  # pragma: no cover
 
         # Check all the things to see if repo is eligble for a pr/issue
         if repo.full_name in exempt_repositories_list:
-            print("Skipping " + repo.full_name + " (exempted)")
+            print(f"Skipping {repo.full_name} (exempted)")
             continue
         if repo.archived:
-            print("Skipping " + repo.full_name + " (archived)")
+            print(f"Skipping {repo.full_name} (archived)")
             continue
         if repo.visibility.lower() not in filter_visibility:
-            print("Skipping " + repo.full_name + " (visibility-filtered)")
+            print(f"Skipping {repo.full_name} (visibility-filtered)")
             continue
         existing_config = None
         filename_list = [".github/dependabot.yaml", ".github/dependabot.yml"]
@@ -96,20 +88,14 @@ def main():  # pragma: no cover
                 break
 
         if existing_config and not update_existing:
-            print(
-                "Skipping "
-                + repo.full_name
-                + " (dependabot file already exists and update_existing is False)"
-            )
+            print(f"Skipping {repo.full_name} (dependabot file already exists and update_existing is False)")
             continue
 
-        if created_after_date and is_repo_created_date_before(
-            repo.created_at, created_after_date
-        ):
-            print("Skipping " + repo.full_name + " (created after filter)")
+        if created_after_date and is_repo_created_date_before(repo.created_at, created_after_date):
+            print(f"Skipping {repo.full_name} (created after filter)")
             continue
 
-        print("Checking " + repo.full_name + " for compatible package managers")
+        print(f"Checking {repo.full_name} for compatible package managers")
         # Try to detect package managers and build a dependabot file
         dependabot_file = build_dependabot_file(
             repo,
@@ -133,42 +119,32 @@ def main():  # pragma: no cover
                 if not skip:
                     print("\tEligible for configuring dependabot.")
                     count_eligible += 1
-                    print("\tConfiguration:\n" + dependabot_file)
+                    print(f"\tConfiguration:\n {dependabot_file}")
             if follow_up_type == "pull":
                 # Try to detect if the repo already has an open pull request for dependabot
                 skip = check_pending_pulls_for_duplicates(title, repo)
                 if not skip:
                     print("\tEligible for configuring dependabot.")
                     count_eligible += 1
-                    print("\tConfiguration:\n" + dependabot_file)
+                    print(f"\tConfiguration:\n {dependabot_file}")
             continue
 
         # Get dependabot security updates enabled if possible
         if enable_security_updates:
-            if not is_dependabot_security_updates_enabled(repo.owner, repo.name, token):
-                enable_dependabot_security_updates(repo.owner, repo.name, token)
+            if not is_dependabot_security_updates_enabled(ghe, repo.owner, repo.name, token):
+                enable_dependabot_security_updates(ghe, repo.owner, repo.name, token)
 
         if follow_up_type == "issue":
             skip = check_pending_issues_for_duplicates(title, repo)
             if not skip:
                 count_eligible += 1
-                body_issue = (
-                    body
-                    + "\n\n```yaml\n"
-                    + "# "
-                    + dependabot_filename_to_use
-                    + "\n"
-                    + dependabot_file
-                    + "\n```"
-                )
+                body_issue = f"{body}\n\n```yaml\n# {dependabot_filename_to_use} \n{dependabot_file}\n```"
                 issue = repo.create_issue(title, body_issue)
-                print("\tCreated issue " + issue.html_url)
+                print(f"\tCreated issue {issue.html_url}")
                 if project_id:
-                    issue_id = get_global_issue_id(
-                        token, organization, repo.name, issue.number
-                    )
-                    link_item_to_project(token, project_id, issue_id)
-                    print("\tLinked issue to project " + project_id)
+                    issue_id = get_global_issue_id(ghe, token, organization, repo.name, issue.number)
+                    link_item_to_project(ghe, token, project_id, issue_id)
+                    print(f"\tLinked issue to project {project_id}")
         else:
             # Try to detect if the repo already has an open pull request for dependabot
             skip = check_pending_pulls_for_duplicates(title, repo)
@@ -186,34 +162,32 @@ def main():  # pragma: no cover
                         dependabot_filename_to_use,
                         existing_config,
                     )
-                    print("\tCreated pull request " + pull.html_url)
+                    print(f"\tCreated pull request {pull.html_url}")
                     if project_id:
-                        pr_id = get_global_pr_id(
-                            token, organization, repo.name, pull.number
-                        )
-                        response = link_item_to_project(token, project_id, pr_id)
+                        pr_id = get_global_pr_id(ghe, token, organization, repo.name, pull.number)
+                        response = link_item_to_project(ghe, token, project_id, pr_id)
                         if response:
-                            print("\tLinked pull request to project " + project_id)
+                            print(f"\tLinked pull request to project {project_id}")
                 except github3.exceptions.NotFoundError:
                     print("\tFailed to create pull request. Check write permissions.")
                     continue
 
-    print("Done. " + str(count_eligible) + " repositories were eligible.")
+    print(f"Done. {str(count_eligible)} repositories were eligible.")
 
 
 def is_repo_created_date_before(repo_created_at: str, created_after_date: str):
     """Check if the repository was created before the created_after_date"""
     repo_created_at_date = datetime.fromisoformat(repo_created_at).replace(tzinfo=None)
-    return created_after_date and repo_created_at_date < datetime.strptime(
-        created_after_date, "%Y-%m-%d"
-    )
+    return created_after_date and repo_created_at_date < datetime.strptime(created_after_date, "%Y-%m-%d")
 
 
-def is_dependabot_security_updates_enabled(owner, repo, access_token):
-    """Check if Dependabot security updates are enabled at the
-    /repos/:owner/:repo/automated-security-fixes endpoint using the requests library
+def is_dependabot_security_updates_enabled(ghe, owner, repo, access_token):
+    """
+    Check if Dependabot security updates are enabled at the /repos/:owner/:repo/automated-security-fixes endpoint using the requests library
+    API: https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#check-if-automated-security-fixes-are-enabled-for-a-repository
     """
-    url = f"https://api.github.com/repos/{owner}/{repo}/automated-security-fixes"
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/repos/{owner}/{repo}/automated-security-fixes"
     headers = {
         "Authorization": f"Bearer {access_token}",
         "Accept": "application/vnd.github.london-preview+json",
@@ -247,9 +221,13 @@ def check_existing_config(repo, filename):
     return None
 
 
-def enable_dependabot_security_updates(owner, repo, access_token):
-    """Enable Dependabot security updates at the /repos/:owner/:repo/automated-security-fixes endpoint using the requests library"""
-    url = f"https://api.github.com/repos/{owner}/{repo}/automated-security-fixes"
+def enable_dependabot_security_updates(ghe, owner, repo, access_token):
+    """
+    Enable Dependabot security updates at the /repos/:owner/:repo/automated-security-fixes endpoint using the requests library
+    API: https://docs.github.com/en/rest/repos/repos?apiVersion=2022-11-28#enable-automated-security-fixes
+    """
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/repos/{owner}/{repo}/automated-security-fixes"
     headers = {
         "Authorization": f"Bearer {access_token}",
         "Accept": "application/vnd.github.london-preview+json",
@@ -277,9 +255,7 @@ def get_repos_iterator(organization, team_name, repository_list, github_connecti
     else:
         # Get the repositories from the repository_list
         for repo in repository_list:
-            repos.append(
-                github_connection.repository(repo.split("/")[0], repo.split("/")[1])
-            )
+            repos.append(github_connection.repository(repo.split("/")[0], repo.split("/")[1]))
 
     return repos
 
@@ -290,7 +266,7 @@ def check_pending_pulls_for_duplicates(title, repo) -> bool:
     skip = False
     for pull_request in pull_requests:
         if pull_request.title.startswith(title):
-            print("\tPull request already exists: " + pull_request.html_url)
+            print(f"\tPull request already exists: {pull_request.html_url}")
             skip = True
             break
     return skip
@@ -302,7 +278,7 @@ def check_pending_issues_for_duplicates(title, repo) -> bool:
     skip = False
     for issue in issues:
         if issue.title.startswith(title):
-            print("\tIssue already exists: " + issue.html_url)
+            print(f"\tIssue already exists: {issue.html_url}")
             skip = True
             break
     return skip
@@ -338,19 +314,19 @@ def commit_changes(
             branch=branch_name,
         )
 
-    pull = repo.create_pull(
-        title=title, body=body, head=branch_name, base=repo.default_branch
-    )
+    pull = repo.create_pull(title=title, body=body, head=branch_name, base=repo.default_branch)
     return pull
 
 
-def get_global_project_id(token, organization, number):
-    """Fetches the project ID from GitHub's GraphQL API."""
-    url = "https://api.github.com/graphql"
+def get_global_project_id(ghe, token, organization, number):
+    """
+    Fetches the project ID from GitHub's GraphQL API.
+    API: https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
+    """
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/graphql"
     headers = {"Authorization": f"Bearer {token}"}
-    data = {
-        "query": f'query{{organization(login: "{organization}") {{projectV2(number: {number}){{id}}}}}}'
-    }
+    data = {"query": f'query{{organization(login: "{organization}") {{projectV2(number: {number}){{id}}}}}}'}
 
     try:
         response = requests.post(url, headers=headers, json=data, timeout=20)
@@ -366,9 +342,13 @@ def get_global_project_id(token, organization, number):
         return None
 
 
-def get_global_issue_id(token, organization, repository, issue_number):
-    """Fetches the issue ID from GitHub's GraphQL API"""
-    url = "https://api.github.com/graphql"
+def get_global_issue_id(ghe, token, organization, repository, issue_number):
+    """
+    Fetches the issue ID from GitHub's GraphQL API
+    API: https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
+    """
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/graphql"
     headers = {"Authorization": f"Bearer {token}"}
     data = {
         "query": f"""
@@ -396,9 +376,13 @@ def get_global_issue_id(token, organization, repository, issue_number):
         return None
 
 
-def get_global_pr_id(token, organization, repository, pr_number):
-    """Fetches the pull request ID from GitHub's GraphQL API"""
-    url = "https://api.github.com/graphql"
+def get_global_pr_id(ghe, token, organization, repository, pr_number):
+    """
+    Fetches the pull request ID from GitHub's GraphQL API
+    API: https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
+    """
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/graphql"
     headers = {"Authorization": f"Bearer {token}"}
     data = {
         "query": f"""
@@ -426,13 +410,15 @@ def get_global_pr_id(token, organization, repository, pr_number):
         return None
 
 
-def link_item_to_project(token, project_id, item_id):
-    """Links an item (issue or pull request) to a project in GitHub."""
-    url = "https://api.github.com/graphql"
+def link_item_to_project(ghe, token, project_id, item_id):
+    """
+    Links an item (issue or pull request) to a project in GitHub.
+    API: https://docs.github.com/en/graphql/guides/forming-calls-with-graphql
+    """
+    api_endpoint = f"{ghe}/api/v3" if ghe else "https://api.github.com"
+    url = f"{api_endpoint}/graphql"
     headers = {"Authorization": f"Bearer {token}"}
-    data = {
-        "query": f'mutation {{addProjectV2ItemById(input: {{projectId: "{project_id}", contentId: "{item_id}"}}) {{item {{id}}}}}}'
-    }
+    data = {"query": f'mutation {{addProjectV2ItemById(input: {{projectId: "{project_id}", contentId: "{item_id}"}}) {{item {{id}}}}}}'}
 
     try:
         response = requests.post(url, headers=headers, json=data, timeout=20)

test_auth.py

@@ -57,9 +57,7 @@ def test_get_github_app_installation_token(self, mock_post):
         mock_response.json.return_value = {"token": dummy_token}
         mock_post.return_value = mock_response
 
-        result = auth.get_github_app_installation_token(
-            b"gh_private_token", "gh_app_id", "gh_installation_id"
-        )
+        result = auth.get_github_app_installation_token(b"ghe", "gh_private_token", "gh_app_id", "gh_installation_id")
 
         self.assertEqual(result, dummy_token)