TutorTask555: Add rate-limited gh auto-invite code (from Krishna's latest checkpoint) and some unit tests using mocking

Pre-commit checks:
All checks passed ✅

Author: indrayudd

DATA605/TutorialsTask76_automate_collaborator_invitations_from_gsheet.py

@@ -1,27 +1,49 @@
+#!/usr/bin/env python
 """
+Invite GitHub collaborators listed in a Google Sheet while obeying the
+50-invite / 24-hour cap.
+
+> automate_collaborator_invitations.py \
+    --drive_url "https://docs.google.com/spreadsheets/d/1Ez5uRvOgvDMkFc9c6mI21kscTKnpiCSh4UkUh_ifLIw
+    /edit?gid=0#gid=0" \
+    --gh_token "$GH_PAT" \
+    --org_name causify-ai \
+    --repo_name tutorials
+
+
 Import as:
 
 import DATA605.TutorialsTask76_automate_collaborator_invitations_from_gsheet as dtacifrgs
 """
-
+import argparse
+import datetime
 import logging
 import subprocess
+import sys
 from typing import List
 
-# Install pygithub.
-subprocess.run(
-    ["sudo", "/venv/bin/pip", "install", "--quiet", " pygithub"], check=True
-)
-import helpers.hgoogle_file_api as hgofiapi
-from github import Github
+# Install required packages and configure.
+packages = [
+    "pygithub",
+    "google-api-python-client",
+    "oauth2client",
+    "gspread",
+    "ratelimit",
+]
+for pkg in packages:
+    subprocess.run(
+        [sys.executable, "-m", "pip", "install", "--quiet", "--upgrade", pkg],
+        check=True,
+    )
 
 _LOG = logging.getLogger(__name__)
 
-# Globals.
-DRIVE_URL = "https://docs.google.com/spreadsheets/d/1Ez5uRvOgvDMkFc9c6mI21kscTKnpiCSh4UkUh_ifLIw/edit?gid=0#gid=0"
-GH_ACCESS_TOKEN = ""
-REPO_NAME = "tutorials"
-ORG_NAME = "causify-ai"
+import github
+import helpers.hgoogle_drive_api as hgodrapi
+import ratelimit
+
+_INVITES_PER_WINDOW = 50
+_WINDOW_SECONDS = int(datetime.timedelta(hours=24).total_seconds())
 
 
 def extract_usernames_from_gsheet(gsheet_url: str) -> List[str]:
@@ -31,15 +53,31 @@ def extract_usernames_from_gsheet(gsheet_url: str) -> List[str]:
     :param gsheet_url: URL of the Google Sheet
     :return: github usernames
     """
-    credentials = hgofiapi.get_credentials(
+    credentials = hgodrapi.get_credentials(
         service_key_path="/app/DATA605/google_secret.json"
     )
-    df = hgofiapi.read_google_file(url, credentials=credentials)
-    usernames = df["github_username"].tolist()
+    df = hgodrapi.read_google_file(gsheet_url, credentials=credentials)
+    usernames = [
+        user for user in df["GitHub user"].tolist() if user and user.strip()
+    ]
     _LOG.info("Usernames = \n  %s", usernames)
     return usernames
 
 
+@ratelimit.sleep_and_retry
+@ratelimit.limits(calls=_INVITES_PER_WINDOW, period=_WINDOW_SECONDS)
+def _invite(repo, username: str, *, permission: str = "write") -> None:
+    """
+    Invite one user, limiting it to 50 invites/24h.
+
+    :param repo: path to repo
+    :param username: username to add
+    :param permission: type of permission
+    """
+    repo.add_to_collaborators(username, permission=permission)
+    _LOG.info("Invitation sent to %s", username)
+
+
 def send_invitations(
     usernames: List[str],
     gh_access_token: str,
@@ -54,29 +92,53 @@ def send_invitations(
     :param repo_url: URL of the target repository
     """
     # Initialize GitHub API.
-    g = Github(gh_access_token)
+    gh = github.Github(gh_access_token)
     # Get the repository.
-    repo = g.get_repo(f"{org_name}/{repo_name}")
+    repo = gh.get_repo(f"{org_name}/{repo_name}")
     # Send invitations.
     for username in usernames:
         try:
-            # Send invitation by adding as collaborator.
-            repo.add_to_collaborators(
-                username,
-                # TODO Krishna: Update the permission accordingly.
-                permission="pull",
+            _invite(repo, username)
+        except github.GithubException as exc:
+            _LOG.error(
+                "Failed to invite %s: %s", username, exc.data.get("message")
             )
-            _LOG.info(f"Invitation sent to {username}")
-        except Exception as e:
-            _LOG.error(f"Failed to invite {username}: {str(e)}")
 
 
-def main():
-    # Extract usernames from Google Sheet.
-    usernames = extract_usernames_from_gsheet(DRIVE_URL)
-    # Send invitations.
-    send_invitations(usernames, GH_ACCESS_TOKEN, REPO_NAME, ORG_NAME)
+def _parse() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Invite GitHub collaborators from a Google Sheet, respecting the 50‑per‑day limit.",  # noqa: E501
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    parser.add_argument(
+        "--drive_url",
+        required=True,
+        help="Google‑Sheet URL with a 'GitHub user' column",
+    )
+    parser.add_argument(
+        "--gh_token",
+        required=True,
+        help="GitHub personal‑access token (repo scope)",
+    )
+    parser.add_argument(
+        "--repo_name", required=True, help="Target repository name (without org)"
+    )
+    parser.add_argument(
+        "--org_name", required=True, help="GitHub organisation name"
+    )
+    parser.add_argument(
+        "--log_level", type=int, default=logging.INFO, help="Logging verbosity"
+    )
+    return parser.parse_args()
+
+
+def _main(args: argparse.Namespace) -> None:
+    logging.basicConfig(
+        level=args.log_level, format="%(asctime)s %(levelname)s %(message)s"
+    )
+    usernames = extract_usernames_from_gsheet(args.drive_url)
+    send_invitations(usernames, args.gh_token, args.repo_name, args.org_name)
 
 
 if __name__ == "__main__":
-    main()
+    _main(_parse())

DATA605/test/Test_TutorialsTask76_automate_collaborator_invitations_from_gsheet.py

@@ -0,0 +1,66 @@
+import unittest.mock as mock
+from typing import List
+
+import helpers.hunit_test as hunitest
+import pandas as pd
+
+MODULE_PATH = "TutorialsTask76_automate_collaborator_invitations_from_gsheet"
+
+
+# #############################################################################
+# Test_extract_usernames_from_gsheet
+# #############################################################################
+
+
+class Test_extract_usernames_from_gsheet(hunitest.TestCase):
+    """
+    Test that github usernames are correctly pulled.
+    """
+
+    def test1(self) -> None:
+        # Mock and import.
+        with (
+            mock.patch(
+                f"{MODULE_PATH}.hgodrapi.get_credentials", return_value="creds"
+            ),
+            mock.patch(
+                f"{MODULE_PATH}.hgodrapi.read_google_file",
+                return_value=pd.DataFrame(
+                    {"GitHub user": ["alice", "bob", None, ""]}
+                ),
+            ),
+        ):
+            mod = __import__(
+                MODULE_PATH, fromlist=["extract_usernames_from_gsheet"]
+            )
+            extract = mod.extract_usernames_from_gsheet
+            actual = extract("dummy_url")
+        expected: List[str] = ["alice", "bob"]
+        self.assertEqual(actual, expected)
+
+
+# #############################################################################
+# Test_send_invitations
+# #############################################################################
+
+
+class Test_send_invitations(hunitest.TestCase):
+    """
+    Test that an invitation is sent once per user.
+    """
+
+    def test2(self) -> None:
+        usernames = ["alice", "bob"]
+        # Mock Github SDK, its repo, and our internal _invite helper.
+        with mock.patch(f"{MODULE_PATH}.github.Github") as m_github:
+            with mock.patch(f"{MODULE_PATH}._invite") as m_invite:
+                dummy_repo = mock.Mock()
+                m_github.return_value.get_repo.return_value = dummy_repo
+                # Import after patches so they take effect.
+                mod = __import__(MODULE_PATH, fromlist=["send_invitations"])
+                send_invitations = mod.send_invitations
+                send_invitations(usernames, "fake_token", "myrepo", "myorg")
+        # Assert one invite call per username.
+        calls = [mock.call(dummy_repo, u) for u in usernames]
+        m_invite.assert_has_calls(calls, any_order=False)
+        self.assertEqual(m_invite.call_count, len(usernames))