feat: add refresh hook for rocket permissions

app/controller/command/commands/team.py

@@ -6,7 +6,7 @@
 from app.controller.command.commands.base import Command
 from app.model.permissions import Permissions
 from db.facade import DBFacade
-from db.utils import get_team_by_name
+from db.utils import get_team_by_name, get_team_members
 from interface.github import GithubAPIException, GithubInterface
 from interface.slack import SlackAPIError
 from interface.gcp import GCPInterface
@@ -649,6 +649,9 @@ def refresh_helper(self, user_id) -> ResponseTuple:
             # add all members (if not already added) to the 'all' team
             self.refresh_all_team()
 
+            # promote members inside special teams
+            self.refresh_all_rocket_permissions()
+
             # enforce Drive permissions
             self.refresh_all_drive_permissions()
         except GithubAPIException as e:
@@ -697,6 +700,41 @@ def refresh_all_team(self):
         else:
             logging.error(f'Could not create {all_name}. Aborting.')
 
+    def refresh_all_rocket_permissions(self):
+        """
+        Refresh Rocket permissions for members in teams like
+        GITHUB_ADMIN_TEAM_NAME and GITHUB_LEADS_TEAM_NAME.
+
+        It only ever promotes users, and does not demote users.
+        """
+        teams = [
+            {
+                'name': self.config.github_team_admin,
+                'permission': Permissions.admin,
+            },
+            {
+                'name': self.config.github_team_leads,
+                'permission': Permissions.team_lead,
+            }
+        ]
+        for t in teams:
+            if len(t['name']) == 0:
+                continue
+
+            team = None
+            try:
+                team = get_team_by_name(self.facade, t['name'])
+            except LookupError:
+                t_id = str(self.gh.org_create_team(t['name']))
+                logging.info(f'team {t["name"]} created')
+                self.facade.store(Team(t_id, t['name'], t['name']))
+
+            if team is not None:
+                team_members = get_team_members(team)
+                for user in team_members:
+                    user.permissions_level = t['permission']
+                    self.facade.store(user)
+
     def refresh_all_drive_permissions(self):
         """
         Refresh Google Drive permissions for all teams. If no GCP client

db/utils.py

@@ -35,6 +35,25 @@ def get_team_by_name(dbf: DBFacade, gh_team_name: str) -> Team:
         return teams[0]
 
 
+def get_team_members(dbf: DBFacade, team: Team) -> List[User]:
+    """
+    Query users that are members of the given team.
+
+    :return: Users that belong to the team
+    """
+    users: List[User] = []
+    for github_id in team.members:
+        users = db.query(User, [('github_user_id', github_id)])
+        if len(users) != 1:
+            logging.warn(f"None/multiple users for GitHub ID {github_id}")
+
+        # For now, naiively iterate over all users, due to
+        # https://github.com/ubclaunchpad/rocket2/issues/493
+        for user in users:
+            users.append(user)
+    return users
+
+
 def get_users_by_ghid(dbf: DBFacade, gh_ids: List[str]) -> List[User]:
     """
     Query users by github user id.

interface/gcp_utils.py

@@ -3,6 +3,7 @@
 from typing import List, Optional
 from interface.gcp import GCPInterface
 from db import DBFacade
+from db.utils import get_team_members
 from app.model import User, Team
 
 
@@ -43,17 +44,11 @@ def sync_team_email_perms(gcp: Optional[GCPInterface],
         return
 
     # Generate who to share with
+    team_members = get_team_members(db, team)
     emails: List[str] = []
-    for github_id in team.members:
-        users = db.query(User, [('github_user_id', github_id)])
-        if len(users) != 1:
-            logging.warn(f"None/multiple users for GitHub ID {github_id}")
-
-        # For now, naiively iterate over all users, due to
-        # https://github.com/ubclaunchpad/rocket2/issues/493
-        for user in users:
-            if len(user.email) > 0:
-                emails.append(user.email)
+    for user in team_members:
+        if len(user.email) > 0:
+            emails.append(user.email)
 
     # Sync permissions
     if len(emails) > 0: