Merge pull request #1 from github/prs

Add main and pr env vars

.github/linters/.isort.cfg

@@ -1,2 +1,2 @@
-[isort]
+[settings]
 profile = black

.pylintrc

@@ -12,4 +12,5 @@ disable=
 	duplicate-code,
 	too-many-branches,
 	too-many-statements,
-	too-many-locals,
+	too-many-locals,
+	wrong-import-order

README.md

@@ -13,7 +13,7 @@ If you need support using this project or have questions about it, please [open
 1. Create a repository to host this GitHub Action or select an existing repository.
 1. Select a best fit workflow file from the [examples below](#example-workflows).
 1. Copy that example into your repository (from step 1) and into the proper directory for GitHub Actions: `.github/workflows/` directory with the file extension `.yml` (ie. `.github/workflows/cleanowners.yml`)
-1. Edit the values (`ORGANIZATION`, `EXEMPT_REPOS`) from the sample workflow with your information. 
+1. Edit the values (`ORGANIZATION`, `EXEMPT_REPOS`) from the sample workflow with your information.
 1. Also edit the value for `GH_ENTERPRISE_URL` if you are using a GitHub Server and not using github.com. For github.com users, don't put anything in here.
 1. Update the value of `GH_TOKEN`. Do this by creating a [GitHub API token](https://docs.github.com/en/authentication/keeping-your-account-and-data-secure/managing-your-personal-access-tokens#creating-a-personal-access-token-classic) with permissions to read the repository/organization and write issues or pull requests. Then take the value of the API token you just created, and [create a repository secret](https://docs.github.com/en/actions/security-guides/encrypted-secrets) where the name of the secret is `GH_TOKEN` and the value of the secret the API token. It just needs to match between when you create the secret name and when you refer to it in the workflow file.
 1. Commit the workflow file to the default branch (often `master` or `main`)

cleanowners.py

@@ -0,0 +1,166 @@
+"""A GitHub Action to suggest removal of non-organization members from CODEOWNERS files."""
+
+import uuid
+
+import auth
+import env
+import github3
+
+
+def main():  # pragma: no cover
+    """Run the main program"""
+
+    # Get the environment variables
+    (
+        organization,
+        repository_list,
+        token,
+        ghe,
+        exempt_repositories_list,
+        dry_run,
+        title,
+        body,
+        commit_message,
+    ) = env.get_env_vars()
+
+    # Auth to GitHub.com or GHE
+    github_connection = auth.auth_to_github(token, ghe)
+    pull_count = 0
+    eligble_for_pr_count = 0
+
+    # Get the repositories from the organization or list of repositories
+    repos = get_repos_iterator(organization, repository_list, github_connection)
+
+    for repo in repos:
+        # Check if the repository is in the exempt_repositories_list
+        if repo.full_name in exempt_repositories_list:
+            print(f"Skipping {repo.full_name} as it is in the exempt_repositories_list")
+            continue
+
+        # Check to see if repository is archived
+        if repo.archived:
+            print(f"Skipping {repo.full_name} as it is archived")
+            continue
+
+        # Check to see if repository has a CODEOWNERS file
+        file_changed = False
+        codeowners_file_contents = None
+        codeowners_filepath = None
+        try:
+            if repo.file_contents(".github/CODEOWNERS").size > 0:
+                codeowners_file_contents = repo.file_contents(".github/CODEOWNERS")
+                codeowners_filepath = ".github/CODEOWNERS"
+        except github3.exceptions.NotFoundError:
+            pass
+        try:
+            if repo.file_contents("CODEOWNERS").size > 0:
+                codeowners_file_contents = repo.file_contents("CODEOWNERS")
+                codeowners_filepath = "CODEOWNERS"
+        except github3.exceptions.NotFoundError:
+            pass
+        try:
+            if repo.file_contents("docs/CODEOWNERS").size > 0:
+                codeowners_file_contents = repo.file_contents("docs/CODEOWNERS")
+                codeowners_filepath = "docs/CODEOWNERS"
+        except github3.exceptions.NotFoundError:
+            pass
+
+        if not codeowners_file_contents:
+            print(f"Skipping {repo.full_name} as it does not have a CODEOWNERS file")
+            continue
+
+        # Extract the usernames from the CODEOWNERS file
+        usernames = get_usernames_from_codeowners(codeowners_file_contents)
+
+        for username in usernames:
+            # Check to see if the username is a member of the organization
+            if not github_connection.organization(organization).has_member(username):
+                print(
+                    f"\t{username} is not a member of {organization}. Suggest removing them from {repo.full_name}"
+                )
+                if not dry_run:
+                    # Remove that username from the codeowners_file_contents
+                    file_changed = True
+                    codeowners_file_contents = codeowners_file_contents.decoded.replace(
+                        f"@{username}", ""
+                    )
+
+        # Update the CODEOWNERS file if usernames were removed
+        if file_changed:
+            eligble_for_pr_count += 1
+            try:
+                pull = commit_changes(
+                    title,
+                    body,
+                    repo,
+                    codeowners_file_contents,
+                    commit_message,
+                    codeowners_filepath,
+                )
+                pull_count += 1
+                print(f"\tCreated pull request {pull.html_url}")
+            except github3.exceptions.NotFoundError:
+                print("\tFailed to create pull request. Check write permissions.")
+                continue
+
+    # Report the statistics from this run
+    print(f"Found {eligble_for_pr_count} users to remove")
+    print(f"Created {pull_count} pull requests successfully")
+
+
+def get_repos_iterator(organization, repository_list, github_connection):
+    """Get the repositories from the organization or list of repositories"""
+    repos = []
+    if organization and not repository_list:
+        repos = github_connection.organization(organization).repositories()
+    else:
+        # Get the repositories from the repository_list
+        for repo in repository_list:
+            repos.append(
+                github_connection.repository(repo.split("/")[0], repo.split("/")[1])
+            )
+
+    return repos
+
+
+def get_usernames_from_codeowners(codeowners_file_contents):
+    """Extract the usernames from the CODEOWNERS file"""
+    usernames = []
+    for line in codeowners_file_contents.decoded.splitlines():
+        # skip comments
+        if line.startswith("#"):
+            continue
+        # skip empty lines
+        if not line.strip():
+            continue
+        # If the line has an @ symbol, grab the word with the @ in it and add it to the list
+        if "@" in line:
+            usernames.append(line.split("@")[1].split()[0])
+    return usernames
+
+
+def commit_changes(
+    title, body, repo, codeowners_file_contents, commit_message, codeowners_filepath
+):
+    """Commit the changes to the repo and open a pull reques and return the pull request object"""
+    default_branch = repo.default_branch
+    # Get latest commit sha from default branch
+    default_branch_commit = repo.ref("heads/" + default_branch).object.sha
+    front_matter = "refs/heads/"
+    branch_name = "codeowners-" + str(uuid.uuid4())
+    repo.create_ref(front_matter + branch_name, default_branch_commit)
+    repo.create_file(
+        path=codeowners_filepath,
+        message=commit_message,
+        content=codeowners_file_contents.encode(),  # Convert to bytes object
+        branch=branch_name,
+    )
+
+    pull = repo.create_pull(
+        title=title, body=body, head=branch_name, base=repo.default_branch
+    )
+    return pull
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

env.py

@@ -8,7 +8,9 @@
 from dotenv import load_dotenv
 
 
-def get_env_vars() -> tuple[str | None, list[str], str, str, list[str], bool]:
+def get_env_vars() -> (
+    tuple[str | None, list[str], str, str, list[str], bool, str, str, str]
+):
     """
     Get the environment variables for use in the action.
 
@@ -22,6 +24,9 @@ def get_env_vars() -> tuple[str | None, list[str], str, str, list[str], bool]:
         ghe (str): The GitHub Enterprise URL to use for authentication
         exempt_repositories_list (list[str]): A list of repositories to exempt from the action
         dry_run (bool): Whether or not to actually open issues/pull requests
+        title (str): The title to use for the pull request
+        body (str): The body to use for the pull request
+        message (str): Commit message to use
 
     """
     # Load from .env file if it exists
@@ -71,11 +76,45 @@ def get_env_vars() -> tuple[str | None, list[str], str, str, list[str], bool]:
     else:
         dry_run_bool = False
 
+    title = os.getenv("TITLE")
+    # make sure that title is a string with less than 70 characters
+    if title:
+        if len(title) > 70:
+            raise ValueError(
+                "TITLE environment variable is too long. Max 70 characters"
+            )
+    else:
+        title = "Clean up CODEOWNERS file"
+
+    body = os.getenv("BODY")
+    # make sure that body is a string with less than 65536 characters
+    if body:
+        if len(body) > 65536:
+            raise ValueError(
+                "BODY environment variable is too long. Max 65536 characters"
+            )
+    else:
+        body = "Consider these updates to the CODEOWNERS file to remove users no longer in this organization."
+
+    commit_message = os.getenv("COMMIT_MESSAGE")
+    if commit_message:
+        if len(commit_message) > 65536:
+            raise ValueError(
+                "COMMIT_MESSAGE environment variable is too long. Max 65536 characters"
+            )
+    else:
+        commit_message = (
+            "Remove users no longer in this organization from CODEOWNERS file"
+        )
+
     return (
         organization,
         repositories_list,
         token,
         ghe,
         exempt_repositories_list,
         dry_run_bool,
+        title,
+        body,
+        commit_message,
     )

test_cleanowners.py

@@ -0,0 +1,132 @@
+"""Test the functions in the cleanowners module."""
+
+import unittest
+import uuid
+from unittest.mock import MagicMock, patch
+
+from cleanowners import (
+    commit_changes,
+    get_repos_iterator,
+    get_usernames_from_codeowners,
+)
+
+
+class TestCommitChanges(unittest.TestCase):
+    """Test the commit_changes function in cleanowners.py"""
+
+    @patch("uuid.uuid4")
+    def test_commit_changes(self, mock_uuid):
+        """Test the commit_changes function."""
+        mock_uuid.return_value = uuid.UUID(
+            "12345678123456781234567812345678"
+        )  # Mock UUID generation
+        mock_repo = MagicMock()  # Mock repo object
+        mock_repo.default_branch = "main"
+        mock_repo.ref.return_value.object.sha = "abc123"  # Mock SHA for latest commit
+        mock_repo.create_ref.return_value = True
+        mock_repo.create_file.return_value = True
+        mock_repo.create_pull.return_value = "MockPullRequest"
+
+        title = "Test Title"
+        body = "Test Body"
+        dependabot_file = "testing!"
+        branch_name = "codeowners-12345678-1234-5678-1234-567812345678"
+        commit_message = "Test commit message"
+        result = commit_changes(
+            title,
+            body,
+            mock_repo,
+            dependabot_file,
+            commit_message,
+            "CODEOWNERS",
+        )
+
+        # Assert that the methods were called with the correct arguments
+        mock_repo.create_ref.assert_called_once_with(
+            f"refs/heads/{branch_name}", "abc123"
+        )
+        mock_repo.create_file.assert_called_once_with(
+            path="CODEOWNERS",
+            message=commit_message,
+            content=dependabot_file.encode(),
+            branch=branch_name,
+        )
+        mock_repo.create_pull.assert_called_once_with(
+            title=title,
+            body=body,
+            head=branch_name,
+            base="main",
+        )
+
+        # Assert that the function returned the expected result
+        self.assertEqual(result, "MockPullRequest")
+
+
+class TestGetUsernamesFromCodeowners(unittest.TestCase):
+    """Test the get_usernames_from_codeowners function in cleanowners.py"""
+
+    def test_get_usernames_from_codeowners(self):
+        """Test the get_usernames_from_codeowners function."""
+        codeowners_file_contents = MagicMock()
+        codeowners_file_contents.decoded = """
+        # Comment
+        @user1
+        @user2
+        # Another comment
+        @user3
+        """
+        expected_usernames = ["user1", "user2", "user3"]
+
+        result = get_usernames_from_codeowners(codeowners_file_contents)
+
+        self.assertEqual(result, expected_usernames)
+
+
+class TestGetReposIterator(unittest.TestCase):
+    """Test the get_repos_iterator function in evergreen.py"""
+
+    @patch("github3.login")
+    def test_get_repos_iterator_with_organization(self, mock_github):
+        """Test the get_repos_iterator function with an organization"""
+        organization = "my_organization"
+        repository_list = []
+        github_connection = mock_github.return_value
+
+        mock_organization = MagicMock()
+        mock_repositories = MagicMock()
+        mock_organization.repositories.return_value = mock_repositories
+        github_connection.organization.return_value = mock_organization
+
+        result = get_repos_iterator(organization, repository_list, github_connection)
+
+        # Assert that the organization method was called with the correct argument
+        github_connection.organization.assert_called_once_with(organization)
+
+        # Assert that the repositories method was called on the organization object
+        mock_organization.repositories.assert_called_once()
+
+        # Assert that the function returned the expected result
+        self.assertEqual(result, mock_repositories)
+
+    @patch("github3.login")
+    def test_get_repos_iterator_with_repository_list(self, mock_github):
+        """Test the get_repos_iterator function with a repository list"""
+        organization = None
+        repository_list = ["org/repo1", "org/repo2"]
+        github_connection = mock_github.return_value
+
+        mock_repository = MagicMock()
+        mock_repository_list = [mock_repository, mock_repository]
+        github_connection.repository.side_effect = mock_repository_list
+
+        result = get_repos_iterator(organization, repository_list, github_connection)
+
+        # Assert that the repository method was called with the correct arguments for each repository in the list
+        expected_calls = [
+            unittest.mock.call("org", "repo1"),
+            unittest.mock.call("org", "repo2"),
+        ]
+        github_connection.repository.assert_has_calls(expected_calls)
+
+        # Assert that the function returned the expected result
+        self.assertEqual(result, mock_repository_list)