update_limits.py #4

#!/usr/bin/env python3
"""
update_limits.py - Ensure all users are in the limits group.

This script makes sure that all Perforce users are in the limits group.

Usage:
    python update_limits.py <instance>

Arguments:
    instance    SDP instance number
"""

from __future__ import annotations

import subprocess
import sys
from pathlib import Path
from typing import List


def run_p4_command(instance: str, args: List[str]) -> subprocess.CompletedProcess:
    """Run a p4 command for the given instance.

    Args:
        instance: SDP instance number
        args: Command arguments

    Returns:
        CompletedProcess result
    """
    p4_cmd = f"/p4/{instance}/bin/p4_{instance}"
    cmd = ["/p4/common/bin/p4master_run", instance, p4_cmd] + args

    return subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        errors="replace",
        check=False,
    )


def login(instance: str) -> None:
    """Login to Perforce.

    Args:
        instance: SDP instance number
    """
    subprocess.run(
        ["/p4/common/bin/p4login", instance],
        capture_output=True,
        check=False,
    )


def get_all_users(instance: str) -> List[str]:
    """Get all users from the server.

    Args:
        instance: SDP instance number

    Returns:
        List of usernames
    """
    result = run_p4_command(instance, ["-ztag", "-F", "%User%", "users"])

    users: List[str] = []
    for line in result.stdout.splitlines():
        user = line.strip()
        if user:
            users.append(user)

    return users


def get_limits_group_spec(instance: str) -> str:
    """Get the limits group specification.

    Args:
        instance: SDP instance number

    Returns:
        Group specification
    """
    result = run_p4_command(instance, ["group", "-o", "limits"])
    return result.stdout


def update_limits_group(instance: str) -> bool:
    """Update the limits group with all users.

    Args:
        instance: SDP instance number

    Returns:
        True if successful
    """
    # Get all users
    users = get_all_users(instance)
    if not users:
        return True  # No users to add

    # Get current limits group spec
    limits_spec = get_limits_group_spec(instance)

    # Build new spec with all users
    lines = [line for line in limits_spec.splitlines() if line.strip()]

    # Add all users
    for user in users:
        lines.append(f"\t{user}")

    new_spec = "\n".join(lines) + "\n"

    # Update the group
    p4_cmd = f"/p4/{instance}/bin/p4_{instance}"
    result = subprocess.run(
        ["/p4/common/bin/p4master_run", instance, p4_cmd, "group", "-i"],
        input=new_spec,
        capture_output=True,
        text=True,
        check=False,
    )

    return result.returncode == 0


def main() -> int:
    """Main entry point."""
    if len(sys.argv) < 2:
        print(__doc__)
        print("Error: Instance number is required", file=sys.stderr)
        return 1

    instance = sys.argv[1]

    # Login
    login(instance)

    # Update limits group
    if update_limits_group(instance):
        return 0
    else:
        return 1


if __name__ == "__main__":
    sys.exit(main())