chore: Extract functions and add test coverage

open_contrib_pr.py

@@ -9,7 +9,31 @@
 import env
 import github3
 
-if __name__ == "__main__":
+
+def get_repos_json(gh_actor, repos_json_location, token, endpoint):
+    """
+    Get the list of repositories from the JSON file.
+
+    Args:
+        gh_actor (str): The GitHub actor (username).
+        repos_json_location (str): The location of the JSON file containing the repositories.
+        token (str): The GitHub personal access token.
+        endpoint (str): The GitHub endpoint.
+
+    Returns:
+        dict: A dictionary containing the list of repositories.
+    """
+    os.system(f"git clone https://{gh_actor}:{token}@{endpoint}/{repos_json_location}")
+    with open(str(repos_json_location), "r", encoding="utf-8") as repos_file:
+        innersource_repos = json.loads(repos_file.read())
+    return innersource_repos
+
+
+def main():  # pragma: no cover
+    """
+    Automatically open a pull request for repositories that have no CONTRIBUTING.md
+    file from a list of repositories in a JSON file
+    """
     env_vars = env.get_env_vars()
     gh_actor = env_vars.gh_actor
     organization = env_vars.organization
@@ -44,9 +68,7 @@
     os.system(f"git config --global user.email 'no-reply@{endpoint}'")
 
     # Get innersource repos from organization
-    os.system(f"git clone https://{gh_actor}:{token}@{endpoint}/{repos_json_location}")
-    with open(str(repos_json_location), "r", encoding="utf-8") as repos_file:
-        innersource_repos = json.loads(repos_file.read())
+    innersource_repos = get_repos_json(gh_actor, repos_json_location, token, endpoint)
 
     for repo in innersource_repos:
         print(repo["name"])
@@ -56,15 +78,15 @@
                 continue
         except KeyError:
             # clone the repo
-            repo_full_name = repo["full_name"]
-            repo_name = repo["name"]
-            os.system(
-                f"git clone https://{gh_actor}:{token}@{endpoint}/{repo_full_name}"
-            )
+            repo_name = clone_repository(gh_actor, token, endpoint, repo)
+            if not repo_name:
+                continue
+
             # checkout a branch called contributing-doc
-            BRANCH_NAME = "contributing-doc"
+            branch_name = "contributing-doc"
             os.chdir(f"{repo_name}")
-            os.system(f"git checkout -b {BRANCH_NAME}")
+            os.system(f"git checkout -b {branch_name}")
+
             # copy, customize, and git add the template file
             os.system("cp /action/workspace/CONTRIBUTING-template.md CONTRIBUTING.md")
             os.system(f"sed -i 's/Project-Name/{repo_name}/g' CONTRIBUTING.md")
@@ -74,32 +96,69 @@
                 "git commit -m'Request to add a document outlining how to contribute'"
             )
             # git push the branch
-            os.system(f"git push -u origin {BRANCH_NAME}")
+            os.system(f"git push -u origin {branch_name}")
             # open a PR from that branch to the default branch
             default_branch = repo["default_branch"]
             # create the pull request
-            repository_object = github_connection.repository(organization, repo_name)
-            try:
-                repository_object.create_pull(
-                    title=pr_title,
-                    body=pr_body,
-                    head=BRANCH_NAME,
-                    base=default_branch,
-                )
-            except github3.exceptions.UnprocessableEntity:
-                print("Pull request already exists")
-            except github3.exceptions.ForbiddenError:
-                print("Pull request failed")
-            except github3.exceptions.NotFoundError:
-                print("Pull request failed")
-            except github3.exceptions.ConnectionError:
-                print("Pull request failed")
-            except Exception as e:  # pylint: disable=broad-exception-caught
-                print(e)
+            create_pull_request(
+                organization,
+                pr_body,
+                pr_title,
+                github_connection,
+                repo_name,
+                branch_name,
+                default_branch,
+            )
             # Clean up repository dir
             os.chdir("../")
             os.system(f"rm -rf {repo_name}")
 
             # rate limit to 20 repos per hour
             print("Waiting 3 minutes so as not to exceed API limits")
             sleep(180)
+
+
+def create_pull_request(
+    organization,
+    pr_body,
+    pr_title,
+    github_connection,
+    repo_name,
+    branch_name,
+    default_branch,
+):
+    """Create a pull request."""
+    repository_object = github_connection.repository(organization, repo_name)
+    try:
+        repository_object.create_pull(
+            title=pr_title,
+            body=pr_body,
+            head=branch_name,
+            base=default_branch,
+        )
+    except github3.exceptions.UnprocessableEntity:
+        print("Pull request already exists")
+    except github3.exceptions.ForbiddenError:
+        print("Pull request failed")
+    except github3.exceptions.NotFoundError:
+        print("Pull request failed")
+    except github3.exceptions.ConnectionError:
+        print("Pull request failed")
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        print(e)
+
+
+def clone_repository(gh_actor, token, endpoint, repo):
+    """Clone the repository and return the name of the repository."""
+    repo_full_name = repo["full_name"]
+    repo_name = repo["name"]
+    try:
+        os.system(f"git clone https://{gh_actor}:{token}@{endpoint}/{repo_full_name}")
+    except Exception as e:  # pylint: disable=broad-exception-caught
+        print(f"Failed to clone repository: {e}")
+        return None
+    return repo_name
+
+
+if __name__ == "__main__":
+    main()  # pragma: no cover

test_auth.py

@@ -4,6 +4,7 @@
 from unittest.mock import MagicMock, patch
 
 import auth
+import requests
 
 
 class TestAuth(unittest.TestCase):
@@ -91,6 +92,42 @@ def test_get_github_app_installation_token(self, mock_post):
 
         self.assertEqual(result, dummy_token)
 
+    @patch("github3.apps.create_jwt_headers", MagicMock(return_value="gh_token"))
+    @patch("auth.requests.post")
+    def test_get_github_app_installation_token_request_failure(self, mock_post):
+        """
+        Test the get_github_app_installation_token function returns None when the request fails.
+        """
+        # Mock the post request to raise a RequestException
+        mock_post.side_effect = requests.exceptions.RequestException("Request failed")
+
+        # Call the function with test data
+        result = auth.get_github_app_installation_token(
+            ghe="https://api.github.com",
+            gh_app_id=12345,
+            gh_app_private_key_bytes=b"private_key",
+            gh_app_installation_id=678910,
+        )
+
+        # Assert that the result is None
+        self.assertIsNone(result)
+
+    @patch("github3.login")
+    def test_auth_to_github_missing_credentials(self, mock_login):
+        """
+        Test the auth_to_github function raises correct ValueError
+        when credentials are present but incorrect.
+        """
+        mock_login.return_value = None
+        with self.assertRaises(ValueError) as context_manager:
+            auth.auth_to_github("not_a_valid_token", "", "", b"", "", False)
+
+        the_exception = context_manager.exception
+        self.assertEqual(
+            str(the_exception),
+            "Unable to authenticate to GitHub",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

test_open_contrib_pr.py

@@ -0,0 +1,139 @@
+""" Tests for the open_contrib_pr.py functions. """
+
+import unittest
+from unittest.mock import MagicMock, mock_open, patch
+
+import github3
+from open_contrib_pr import clone_repository, create_pull_request, get_repos_json
+
+
+class TestOpenContribPR(unittest.TestCase):
+    """Test case for the open_contrib_pr module."""
+
+    @patch(
+        "builtins.open",
+        new_callable=mock_open,
+        read_data='{"repos": ["repo1", "repo2"]}',
+    )
+    @patch("os.system")
+    def test_get_repos_json(self, mock_system, mock_file):
+        """
+        Test the get_repos_json function.
+        """
+        gh_actor = "test_actor"
+        repos_json_location = "test_location"
+        token = "test_token"
+        endpoint = "test_endpoint"
+
+        expected_repos = {"repos": ["repo1", "repo2"]}
+
+        result = get_repos_json(gh_actor, repos_json_location, token, endpoint)
+
+        mock_system.assert_called_once_with(
+            f"git clone https://{gh_actor}:{token}@{endpoint}/{repos_json_location}"
+        )
+        mock_file.assert_called_once_with(
+            str(repos_json_location), "r", encoding="utf-8"
+        )
+        self.assertEqual(result, expected_repos)
+
+
+class TestCloneRepository(unittest.TestCase):
+    """Test case for the clone_repository function."""
+
+    @patch("os.system")
+    def test_clone_repository_success(self, mock_system):
+        """
+        Test the clone_repository function when the clone is successful.
+        """
+        mock_system.return_value = 0  # Simulate successful clone
+
+        result = clone_repository(
+            gh_actor="test_actor",
+            token="test_token",
+            endpoint="test_endpoint",
+            repo={"full_name": "test_actor/test_repo", "name": "test_repo"},
+        )
+
+        mock_system.assert_called_once_with(
+            "git clone https://test_actor:test_token@test_endpoint/test_actor/test_repo"
+        )
+        self.assertEqual(result, "test_repo")
+
+    @patch("os.system")
+    def test_clone_repository_failure(self, mock_system):
+        """
+        Test the clone_repository function when the clone fails.
+        """
+        mock_system.side_effect = Exception("Clone failed")  # Simulate clone failure
+
+        result = clone_repository(
+            gh_actor="test_actor",
+            token="test_token",
+            endpoint="test_endpoint",
+            repo={"full_name": "test_actor/test_repo", "name": "test_repo"},
+        )
+
+        mock_system.assert_called_once_with(
+            "git clone https://test_actor:test_token@test_endpoint/test_actor/test_repo"
+        )
+        self.assertIsNone(result)
+
+
+class TestCreatePullRequest(unittest.TestCase):
+    """Test case for the create_pull_request function."""
+
+    def test_create_pull_request_success(self):
+        """
+        Test the create_pull_request function when the pull request is created successfully.
+        """
+        github_connection = MagicMock()
+        github_connection.repository.return_value = MagicMock()
+
+        create_pull_request(
+            organization="test_org",
+            pr_body="Test PR body",
+            pr_title="Test PR title",
+            github_connection=github_connection,
+            repo_name="test_repo",
+            branch_name="test_branch",
+            default_branch="main",
+        )
+
+        github_connection.repository.return_value.create_pull.assert_called_once_with(
+            title="Test PR title", body="Test PR body", head="test_branch", base="main"
+        )
+
+    def test_create_pull_exceptions(self):
+        """
+        Test the create_pull_request function when an exception occurs.
+        """
+        github_connection = MagicMock()
+        github_connection.repository.return_value = MagicMock()
+        for exception, message in [
+            (
+                github3.exceptions.UnprocessableEntity(MagicMock()),
+                "Pull request already exists",
+            ),
+            (github3.exceptions.ForbiddenError(MagicMock()), "Pull request failed"),
+            (github3.exceptions.NotFoundError(MagicMock()), "Pull request failed"),
+            (github3.exceptions.ConnectionError(MagicMock()), "Pull request failed"),
+        ]:
+            github_connection.repository.return_value.create_pull.side_effect = (
+                exception
+            )
+            with patch("builtins.print") as mock_print:
+                create_pull_request(
+                    organization="test_org",
+                    pr_body="Test PR body",
+                    pr_title="Test PR title",
+                    github_connection=github_connection,
+                    repo_name="test_repo",
+                    branch_name="test_branch",
+                    default_branch="main",
+                )
+                mock_print.assert_called_once_with(message)
+
+
+if __name__ == "__main__":
+    unittest.main()