feat: Adds check to ensure org doesnt have any outside collaborator

edx_repo_tools/repo_checks/repo_checks.py

@@ -100,30 +100,18 @@ def get_github_file_contents(api, org, repo, path, ref):
     return api.repos.get_content(org, repo, path, ref).content
 
 
-class Check:
+class OrgCheck:
     """
-    Something that we want to ensure about a given repository.
+    Something that we want to ensure about a given organization.
 
     This is an abstract class; concrete checks should be implemented
     as subclasses and override the four methods below
-    (is_relevant, check, fix, and dry_run).
+    (check, fix, and dry_run).
     """
 
-    def __init__(self, api: GhApi, org: str, repo: str):
+    def __init__(self, api: GhApi, org: str):
         self.api = api
         self.org_name = org
-        self.repo_name = repo
-
-    def is_relevant(self) -> bool:
-        """
-        Checks to see if the given check is relevant to run on the
-        given repo.
-
-        This is independent of whether or not the check passes on this repo
-        and should be run before trying to check the repo.
-        """
-
-        raise NotImplementedError
 
     def check(self) -> tuple[bool, str]:
         """
@@ -157,6 +145,31 @@ def dry_run(self):
         raise NotImplementedError
 
 
+class Check(OrgCheck):
+    """
+    Something that we want to ensure about a given repository.
+
+    This is an abstract class; concrete checks should be implemented
+    as subclasses and override the four methods below
+    (is_relevant, check, fix, and dry_run).
+    """
+
+    def __init__(self, api: GhApi, org: str, repo: str):
+        super().__init__(api, org)
+        self.repo_name = repo
+
+    def is_relevant(self) -> bool:
+        """
+        Checks to see if the given check is relevant to run on the
+        given repo.
+
+        This is independent of whether or not the check passes on this repo
+        and should be run before trying to check the repo.
+        """
+
+        raise NotImplementedError
+
+
 class EnsureRepoSettings(Check):
     """
     There are certain settings that we agree we want to be set a specific way on all repos.  This check
@@ -1117,6 +1130,56 @@ def fix(self, dry_run=False):
         return steps
 
 
+class EnsureNoOutsideCollaborators(OrgCheck):
+    """
+    Organization shouldn't have outside collaborators
+    """
+
+    def __init__(self, api: GhApi, org: str):
+        super().__init__(api, org)
+        self.users_list = []
+
+    def check(self) -> tuple[bool, str]:
+        """
+        Verify whether or not the check is failing.
+
+        This should not change anything and should not have a side-effect
+        other than populating `self` with any data that is needed later for
+        `fix` or `dry_run`.
+
+        The string in the return tuple should be a human readable reason
+        that the check failed.
+        """
+        self.users_list = list(all_paged_items(
+            self.api.orgs.list_outside_collaborators, org=self.org_name
+        ))
+        users = [f"{user.login}" for user in self.users_list]
+        if users:
+            return (
+                False,
+                f"The organization has some outside collaborators:\n\t\t"
+                + "\n\t\t".join(users),
+            )
+        return (True, "The organization doesn't have any outside collaborators.")
+
+    def dry_run(self):
+        return self.fix(dry_run=True)
+
+    def fix(self, dry_run=False):
+        steps = []
+        for user in self.users_list:
+            if not dry_run:
+                self.api.orgs.remove_outside_collaborator(
+                    org=self.org_name,
+                    username=user.login,
+                )
+            steps.append(
+                f"Removed outside collaborator {user.login} from org {self.org_name}."
+            )
+
+        return steps
+
+
 CHECKS = [
     RequiredCLACheck,
     RequireTriageTeamAccess,
@@ -1127,7 +1190,10 @@ def fix(self, dry_run=False):
     EnsureNoDirectRepoAccessToUsers,
 ]
 CHECKS_BY_NAME = {check_cls.__name__: check_cls for check_cls in CHECKS}
-CHECKS_BY_NAME_LOWER = {check_cls.__name__.lower(): check_cls for check_cls in CHECKS}
+
+ORG_CHECKS = [
+    EnsureNoOutsideCollaborators,
+]
 
 
 @click.command()
@@ -1199,12 +1265,34 @@ def main(org, dry_run, _github_token, check_names, repos, start_at):
         active_checks = [CHECKS_BY_NAME[check_name] for check_name in check_names]
     else:
         active_checks = CHECKS
-    click.secho(f"The following checks will be run:", fg="magenta", bold=True)
+
+    display_running_checks(active_checks)
+
+    run_checks_on_org(api, dry_run, org)
+    run_checks_on_repos(active_checks, api, dry_run, org, repos, start_at)
+
+
+def display_running_checks(repo_checks):
+    click.secho(f"The following checks will be run on repos:", fg="magenta", bold=True)
     active_checks_string = "\n".join(
-        "\t" + check_cls.__name__ for check_cls in active_checks
+        "\t" + check_cls.__name__ for check_cls in repo_checks
     )
     click.secho(active_checks_string, fg="magenta")
+    click.secho(f"The following checks will be run on org:", fg="magenta", bold=True)
+    org_checks_string = "\n".join(
+        "\t" + check_cls.__name__ for check_cls in ORG_CHECKS
+    )
+    click.secho(org_checks_string, fg="magenta")
+
+
+def run_checks_on_org(api, dry_run, org):
+        click.secho(f"Org > {org}: ", bold=True)
+        for CheckType in ORG_CHECKS:
+            check = CheckType(api, org)
+            run_check(check, dry_run)
+
 
+def run_checks_on_repos(active_checks, api, dry_run, org, repos, start_at):
     before_start_at = bool(start_at)
     for repo in repos:
         if repo == start_at:
@@ -1213,45 +1301,46 @@ def main(org, dry_run, _github_token, check_names, repos, start_at):
         if before_start_at:
             continue
 
-        click.secho(f"{repo}: ", bold=True)
+        click.secho(f"Repo > {repo}: ", bold=True)
         for CheckType in active_checks:
             check = CheckType(api, org, repo)
 
             if check.is_relevant():
-                result = check.check()
-                if result[0]:
-                    color = "green"
-                else:
-                    color = "red"
-
-                click.secho(f"\t{result[1]}", fg=color)
-
-                if dry_run:
-                    try:
-                        steps = check.dry_run()
-                        steps_color = "yellow"
-                    except HTTP4xxClientError as err:
-                        click.echo(err.fp.read().decode("utf-8"))
-                        raise
-                else:
-                    try:
-                        steps = check.fix()
-                        steps_color = "green"
-                    except HTTP4xxClientError as err:
-                        click.echo(err.fp.read().decode("utf-8"))
-                        raise
-
-                if steps:
-                    click.secho("\tSteps:\n\t\t", fg=steps_color, nl=False)
-                    click.secho(
-                        "\n\t\t".join([step.replace("\n", "\n\t\t") for step in steps])
-                    )
+                run_check(check, dry_run)
             else:
                 click.secho(
                     f"\tSkipping {CheckType.__name__} as it is not relevant on this repo.",
                     fg="cyan",
                 )
 
 
+def run_check(check, dry_run):
+    result = check.check()
+    if result[0]:
+        color = "green"
+    else:
+        color = "red"
+    click.secho(f"\t{result[1]}", fg=color)
+    if dry_run:
+        try:
+            steps = check.dry_run()
+            steps_color = "yellow"
+        except HTTP4xxClientError as err:
+            click.echo(err.fp.read().decode("utf-8"))
+            raise
+    else:
+        try:
+            steps = check.fix()
+            steps_color = "green"
+        except HTTP4xxClientError as err:
+            click.echo(err.fp.read().decode("utf-8"))
+            raise
+    if steps:
+        click.secho("\tSteps:\n\t\t", fg=steps_color, nl=False)
+        click.secho(
+            "\n\t\t".join([step.replace("\n", "\n\t\t") for step in steps])
+        )
+
+
 if __name__ == "__main__":
     main()  # pylint: disable=no-value-for-parameter