github_team_membership_check.py

#!/usr/bin/env python3
"""
Script to check GitHub organization team membership for users in Zephyr Project.

Usage:
  # Check a specific user's team membership
  python github_team_membership_check.py --org zephyrproject-rtos --user username

  # Check all users in MAINTAINERS.yml
  python github_team_membership_check.py --org zephyrproject-rtos --maintainers-file MAINTAINERS.yml
  
  # Check a specific user from MAINTAINERS.yml
  python github_team_membership_check.py --org zephyrproject-rtos --maintainers-file MAINTAINERS.yml --user username
"""

import argparse
import requests
import sys
import os
import yaml
import time

def get_org_teams(org, token=None):
    """
    Get all teams in a GitHub organization.
    
    Args:
        org (str): GitHub organization name
        token (str, optional): GitHub personal access token for authentication
    
    Returns:
        list: List of team objects
    """
    # Construct the API URL
    url = f"https://api.github.com/orgs/{org}/teams?per_page=100"
    
    # Set up headers
    headers = {
        "Accept": "application/vnd.github.v3+json"
    }
    
    # Add token if provided
    if token:
        headers["Authorization"] = f"token {token}"
    
    teams = []
    
    try:
        # GitHub uses pagination for large result sets
        while url:
            # Make the request
            response = requests.get(url, headers=headers)
            
            # Check if request was successful
            if response.status_code == 200:
                # Add teams to our list
                page_teams = response.json()
                teams.extend(page_teams)
                
                # Check if there are more pages
                if "next" in response.links:
                    url = response.links["next"]["url"]
                else:
                    url = None
            else:
                if response.status_code == 401:
                    print("Error: Authentication failed. Check your GitHub token.", file=sys.stderr)
                elif response.status_code == 403:
                    print("Error: API rate limit exceeded or insufficient permissions. Try using a GitHub token with 'read:org' scope.", file=sys.stderr)
                    # Get rate limit info
                    rate_limit_url = "https://api.github.com/rate_limit"
                    rate_response = requests.get(rate_limit_url, headers=headers)
                    if rate_response.status_code == 200:
                        rate_info = rate_response.json()
                        reset_time = rate_info['resources']['core']['reset']
                        import datetime
                        reset_datetime = datetime.datetime.fromtimestamp(reset_time)
                        print(f"Rate limit will reset at: {reset_datetime}", file=sys.stderr)
                else:
                    print(f"Error: Unexpected status code {response.status_code}", file=sys.stderr)
                    print(f"Response: {response.text}", file=sys.stderr)
                
                sys.exit(1)
        
        return teams
    
    except requests.exceptions.RequestException as e:
        print(f"Error making request: {e}", file=sys.stderr)
        sys.exit(1)

def check_team_membership(org, team_slug, username, token=None):
    """
    Check if a user is a member of a GitHub team.
    
    Args:
        org (str): GitHub organization name
        team_slug (str): Team slug
        username (str): GitHub username to check
        token (str, optional): GitHub personal access token for authentication
    
    Returns:
        str or None: Membership state ('active', 'pending') or None if not a member
    """
    # Construct the API URL
    url = f"https://api.github.com/orgs/{org}/teams/{team_slug}/memberships/{username}"
    
    # Set up headers
    headers = {
        "Accept": "application/vnd.github.v3+json"
    }
    
    # Add token if provided
    if token:
        headers["Authorization"] = f"token {token}"
    
    try:
        # Make the request
        response = requests.get(url, headers=headers)
        
        # Status code 200 means the user is a member
        if response.status_code == 200:
            membership = response.json()
            return membership.get('state')
        # Status code 404 means the user is not a member
        elif response.status_code == 404:
            return None
        # Handle other status codes
        else:
            if response.status_code == 401:
                print("Error: Authentication failed. Check your GitHub token.", file=sys.stderr)
            elif response.status_code == 403:
                print("Error: API rate limit exceeded or insufficient permissions. Try using a GitHub token with 'read:org' scope.", file=sys.stderr)
            else:
                print(f"Error: Unexpected status code {response.status_code}", file=sys.stderr)
                print(f"Response: {response.text}", file=sys.stderr)
            
            sys.exit(1)
    
    except requests.exceptions.RequestException as e:
        print(f"Error making request: {e}", file=sys.stderr)
        sys.exit(1)

def get_user_team_memberships(org, username, teams, token=None):
    """
    Get all teams a user is a member of in a GitHub organization.
    
    Args:
        org (str): GitHub organization name
        username (str): GitHub username to check
        teams (list): List of team objects
        token (str, optional): GitHub personal access token for authentication
    
    Returns:
        list: List of team names the user is a member of
    """
    user_teams = []
    
    for team in teams:
        team_slug = team['slug']
        membership = check_team_membership(org, team_slug, username, token)
        
        if membership == 'active':
            user_teams.append(team['name'])
        
        # Add a small delay to avoid rate limiting
        time.sleep(0.1)
    
    return user_teams

def get_all_team_members(org, token=None):
    """
    Get all members of all teams in a GitHub organization.
    
    Args:
        org (str): GitHub organization name
        token (str, optional): GitHub personal access token for authentication
    
    Returns:
        dict: Dictionary mapping usernames to lists of team names
    """
    # Get all teams in the organization
    teams = get_org_teams(org, token)
    print(f"Found {len(teams)} teams in the {org} organization")
    
    # Dictionary to store team memberships
    all_members = {}
    
    # For each team, get its members
    for team in teams:
        team_name = team['name']
        team_slug = team['slug']
        
        # Construct the API URL for team members
        url = f"https://api.github.com/orgs/{org}/teams/{team_slug}/members?per_page=100"
        
        # Set up headers
        headers = {
            "Accept": "application/vnd.github.v3+json"
        }
        
        # Add token if provided
        if token:
            headers["Authorization"] = f"token {token}"
        
        try:
            # GitHub uses pagination for large result sets
            while url:
                # Make the request
                response = requests.get(url, headers=headers)
                
                # Check if request was successful
                if response.status_code == 200:
                    # Process members
                    members = response.json()
                    for member in members:
                        username = member['login']
                        if username not in all_members:
                            all_members[username] = []
                        all_members[username].append(team_name)
                    
                    # Check if there are more pages
                    if "next" in response.links:
                        url = response.links["next"]["url"]
                    else:
                        url = None
                else:
                    if response.status_code == 401:
                        print(f"Error: Authentication failed when fetching members for team {team_name}.", file=sys.stderr)
                    elif response.status_code == 403:
                        print(f"Error: API rate limit exceeded or insufficient permissions for team {team_name}.", file=sys.stderr)
                    else:
                        print(f"Error: Unexpected status code {response.status_code} for team {team_name}", file=sys.stderr)
                    
                    # Skip this team and continue with others
                    break
            
            # Add a small delay to avoid rate limiting
            time.sleep(0.1)
            
        except requests.exceptions.RequestException as e:
            print(f"Error making request for team {team_name}: {e}", file=sys.stderr)
            # Skip this team and continue with others
            continue
    
    return all_members

def get_maintainers_users(file_path):
    """
    Load the MAINTAINERS.yml file and extract all unique usernames.
    
    Args:
        file_path (str): Path to the MAINTAINERS.yml file
    
    Returns:
        dict: Dictionary with sets of maintainers and collaborators
    """
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            maintainers_data = yaml.safe_load(f)
        
        # Sets to store unique usernames
        all_maintainers = set()
        all_collaborators = set()
        
        # Track which areas each person is involved with
        maintainer_areas = {}
        collaborator_areas = {}
        
        # Process each area in the MAINTAINERS.yml file
        for area_id, area_data in maintainers_data.items():
            # Skip non-dictionary entries (like description)
            if not isinstance(area_data, dict):
                continue
            
            area_name = area_data.get('name', area_id)
            
            # Extract maintainers for this area
            maintainers = area_data.get('maintainers', [])
            for maintainer in maintainers:
                all_maintainers.add(maintainer)
                if maintainer not in maintainer_areas:
                    maintainer_areas[maintainer] = []
                maintainer_areas[maintainer].append(area_name)
            
            # Extract collaborators for this area
            collaborators = area_data.get('collaborators', [])
            for collaborator in collaborators:
                all_collaborators.add(collaborator)
                if collaborator not in collaborator_areas:
                    collaborator_areas[collaborator] = []
                collaborator_areas[collaborator].append(area_name)
        
        return {
            'maintainers': all_maintainers,
            'collaborators': all_collaborators,
            'all_users': all_maintainers.union(all_collaborators),
            'maintainer_areas': maintainer_areas,
            'collaborator_areas': collaborator_areas
        }
    
    except Exception as e:
        print(f"Error loading MAINTAINERS.yml: {e}", file=sys.stderr)
        sys.exit(1)

def main():
    parser = argparse.ArgumentParser(
        description="Check GitHub organization team membership for users in Zephyr Project."
    )
    
    parser.add_argument("--org", "-o", required=True, help="GitHub organization name (e.g., zephyrproject-rtos)")
    parser.add_argument("--user", "-u", help="GitHub username to check")
    parser.add_argument("--maintainers-file", "-m", help="Path to MAINTAINERS.yml file")
    parser.add_argument("--token", "-t", help="GitHub personal access token for authentication")
    parser.add_argument("--output", "-f", help="Output results to a text file")
    parser.add_argument("--list-teams", "-l", action="store_true", help="List all teams in the organization")
    
    args = parser.parse_args()
    
    # Get token from environment variable if not provided
    token = args.token or os.environ.get("GITHUB_TOKEN")
    
    if not token:
        print("Warning: No GitHub token provided. API rate limits may apply and organization data may be inaccessible.", file=sys.stderr)
        print("For best results, provide a token with 'read:org' scope using --token or GITHUB_TOKEN environment variable.", file=sys.stderr)
    
    # List all teams in the organization if requested
    if args.list_teams:
        teams = get_org_teams(args.org, token)
        if teams:
            print(f"Teams in the {args.org} organization:")
            for team in sorted(teams, key=lambda t: t['name']):
                print(f"- {team['name']} (slug: {team['slug']})")
        else:
            print(f"No teams found in the {args.org} organization or you don't have permission to view them.")
        return
    
    # Check a specific user (without MAINTAINERS.yml)
    if args.user and not args.maintainers_file:
        teams = get_org_teams(args.org, token)
        user_teams = get_user_team_memberships(args.org, args.user, teams, token)
        
        if user_teams:
            print(f"{args.user} is a member of the following teams in {args.org}:")
            for team in sorted(user_teams):
                print(f"- {team}")
        else:
            print(f"{args.user} is not a member of any team in {args.org}")
        return
    
    # If we have a maintainers file, process it
    if args.maintainers_file:
        print(f"Loading maintainers from {args.maintainers_file}...")
        maintainers_data = get_maintainers_users(args.maintainers_file)
        
        all_maintainers = maintainers_data['maintainers']
        all_collaborators = maintainers_data['collaborators']
        all_users = maintainers_data['all_users']
        maintainer_areas = maintainers_data['maintainer_areas']
        collaborator_areas = maintainers_data['collaborator_areas']
        
        print(f"Found {len(all_maintainers)} maintainers and {len(all_collaborators)} collaborators "
              f"({len(all_users)} unique users) in MAINTAINERS.yml")
        
        # Check a specific user from MAINTAINERS.yml
        if args.user:
            if args.user not in all_users:
                print(f"{args.user} is not found in MAINTAINERS.yml")
                return
            
            # Determine role in MAINTAINERS.yml
            roles = []
            if args.user in all_maintainers:
                roles.append("Maintainer")
            if args.user in all_collaborators:
                roles.append("Collaborator")
            
            maintainer_yml_role = " & ".join(roles)
            
            # Get areas
            areas = []
            if args.user in maintainer_areas:
                areas.extend(maintainer_areas[args.user])
            if args.user in collaborator_areas:
                areas.extend(collaborator_areas[args.user])
            
            # Remove duplicates
            areas = sorted(set(areas))
            
            # Check GitHub team membership
            teams = get_org_teams(args.org, token)
            user_teams = get_user_team_memberships(args.org, args.user, teams, token)
            
            print(f"\nUser: {args.user}")
            print(f"Role in MAINTAINERS.yml: {maintainer_yml_role}")
            
            if user_teams:
                print(f"GitHub team membership: Yes, member of {len(user_teams)} teams")
                print("\nTeams:")
                for team in sorted(user_teams):
                    print(f"- {team}")
            else:
                print("GitHub team membership: No")
            
            if areas:
                print("\nAreas in MAINTAINERS.yml:")
                for area in areas:
                    print(f"- {area}")
            return
        
        # Check all users from MAINTAINERS.yml
        print(f"Checking team membership for all {len(all_users)} users in MAINTAINERS.yml...")
        print("This may take some time due to GitHub API rate limits...")
        
        # Get all team members in one go (more efficient than checking each user individually)
        all_team_members = get_all_team_members(args.org, token)
        
        # Find users in MAINTAINERS.yml who are not team members
        not_team_members = set()
        are_team_members = set()
        
        for user in all_users:
            if user in all_team_members:
                are_team_members.add(user)
            else:
                not_team_members.add(user)
        
        # Separate maintainers and collaborators for reporting
        maintainers_not_members = all_maintainers.intersection(not_team_members)
        collaborators_not_members = all_collaborators.intersection(not_team_members)
        
        maintainers_are_members = all_maintainers.intersection(are_team_members)
        collaborators_are_members = all_collaborators.intersection(are_team_members)
        
        # Prepare output
        output_lines = []
        
        # Users who are NOT team members
        output_lines.append(f"{len(not_team_members)} users in MAINTAINERS.yml are NOT members of any team in {args.org}:")
        
        if maintainers_not_members:
            output_lines.append(f"\nMaintainers who are NOT team members ({len(maintainers_not_members)} users):")
            for username in sorted(maintainers_not_members):
                output_lines.append(f"- {username}")
        
        if collaborators_not_members:
            output_lines.append(f"\nCollaborators who are NOT team members ({len(collaborators_not_members)} users):")
            for username in sorted(collaborators_not_members):
                output_lines.append(f"- {username}")
        
        # Users who ARE team members
        output_lines.append(f"\n{len(are_team_members)} users in MAINTAINERS.yml ARE members of teams in {args.org}:")
        
        if maintainers_are_members:
            output_lines.append(f"\nMaintainers who ARE team members ({len(maintainers_are_members)} users):")
            for username in sorted(maintainers_are_members):
                teams_str = ", ".join(all_team_members.get(username, []))
                output_lines.append(f"- {username} (Teams: {teams_str})")
        
        if collaborators_are_members:
            output_lines.append(f"\nCollaborators who ARE team members ({len(collaborators_are_members)} users):")
            for username in sorted(collaborators_are_members):
                teams_str = ", ".join(all_team_members.get(username, []))
                output_lines.append(f"- {username} (Teams: {teams_str})")
        
        # Add verification counts
        output_lines.append("\nVerification Counts:")
        output_lines.append(f"- Total users in MAINTAINERS.yml: {len(all_users)}")
        output_lines.append(f"  - Maintainers: {len(all_maintainers)}")
        output_lines.append(f"  - Collaborators: {len(all_collaborators)}")
        output_lines.append(f"- Users who are team members: {len(are_team_members)}")
        output_lines.append(f"  - Maintainers: {len(maintainers_are_members)}")
        output_lines.append(f"  - Collaborators: {len(collaborators_are_members)}")
        output_lines.append(f"- Users who are not team members: {len(not_team_members)}")
        output_lines.append(f"  - Maintainers: {len(maintainers_not_members)}")
        output_lines.append(f"  - Collaborators: {len(collaborators_not_members)}")
        output_lines.append(f"- Sum of both lists: {len(are_team_members) + len(not_team_members)}")
        
        if len(all_users) == len(are_team_members) + len(not_team_members):
            output_lines.append("[SUCCESS] All users accounted for!")
        else:
            output_lines.append("[ERROR] User count mismatch!")
        
        # Print output
        for line in output_lines:
            print(line)
        
        # Write to file if requested
        if args.output:
            try:
                with open(args.output, 'w', encoding='utf-8') as f:
                    f.write(f"Users in MAINTAINERS.yml and their GitHub team membership status in {args.org}:\n\n")
                    for line in output_lines:
                        f.write(f"{line}\n")
                
                print(f"\nResults written to {args.output}")
            except Exception as e:
                print(f"Error writing to file: {e}", file=sys.stderr)
    
    # If we get here without any action, show usage
    else:
        parser.print_help()

if __name__ == "__main__":
    main()