sdputils.py #8

#!/usr/bin/env python3
"""
Utility classes and helpers for interacting with the SDP (Server
Deployment Package) configuration and Perforce command-line tools.

"""

from __future__ import annotations

import os
import platform
import subprocess
from configparser import RawConfigParser
from pathlib import Path
from typing import Iterable, Optional


# Users that should never be deleted by maintenance scripts.
SKIP_USERS = {"p4admin", "perforce", "swarm"}


class SDPUtils:
    """Convenience wrapper around SDP configuration and ``p4`` commands."""

    def __init__(self, sdp_instance: str) -> None:
        self.sdp_instance = str(sdp_instance)
        # Load configuration from maintenance.cfg
        config = RawConfigParser()
        config.read("maintenance.cfg")
        self._config = config
        # Determine server port and user from the environment, falling back to config if needed.
        self.server = os.environ.get("P4PORT") or config.get(self.sdp_instance, "server", fallback="")
        self.p4user = os.environ.get("P4USER") or config.get(self.sdp_instance, "p4user", fallback="")
        # Determine the path to the p4 executable.  On Windows use p4.exe; otherwise use the SDP path.
        if platform.system() == "Windows":
            # ``p4.exe`` will be found on PATH or can be specified via the P4BIN environment variable.
            self.p4binary = os.environ.get("P4BIN", "p4.exe")
        else:
            self.p4binary = f"/p4/{self.sdp_instance}/bin/p4_{self.sdp_instance}"

    @property
    def p4(self) -> str:
        """Alias for ``p4binary`` for backward compatibility."""
        return self.p4binary

    def get(self, varname: str, fallback: Optional[str] = None) -> str:
        """
        Retrieve a configuration value from the ``maintenance.cfg`` file for the
        current instance.

        :param varname: Name of the configuration variable.
        :param fallback: Value to return if the variable is not found.
        :returns: The configuration value.
        """
        return self._config.get(self.sdp_instance, varname, fallback=fallback)

    def run_p4(
        self,
        args: Iterable[str],
        *,
        port: Optional[str] = None,
        user: Optional[str] = None,
        input_data: Optional[str] = None,
        check: bool = False,
        capture_output: bool = False,
    ) -> subprocess.CompletedProcess[str]:
        """
        Run a Perforce command using ``subprocess.run`` with the configured
        port and user by default.

        :param args: Iterable of command arguments (e.g. ["info"]).
        :param port: Optional override for the server port; defaults to the
            instance's configured ``server``.
        :param user: Optional override for the Perforce user; defaults to
            the instance's configured ``p4user``.
        :param input_data: Optional string to pass to the process's stdin.
        :param check: If True, raise CalledProcessError on non‑zero exit.
        :param capture_output: If True, capture stdout/stderr and return
            them in the CompletedProcess.
        :returns: A ``CompletedProcess`` instance.
        """
        cmd = [self.p4binary]
        # Supply port and user if provided; otherwise use defaults.
        p4port = port or self.server
        p4user = user or self.p4user
        if p4port:
            cmd.extend(["-p", p4port])
        if p4user:
            cmd.extend(["-u", p4user])
        cmd.extend(args)
        return subprocess.run(
            cmd,
            input=input_data,
            text=True,
            capture_output=capture_output,
            check=check,
        )

    def login(self, port: Optional[str] = None) -> None:
        """
        Perform a ``p4 login -a`` using the admin password file associated with
        this SDP instance.  By default it uses the instance's configured
        server port, but you can provide a different port.

        The password file is read and its contents passed to the ``p4`` command
        via stdin instead of relying on shell redirection.  This method raises
        ``CalledProcessError`` if the login fails.
        """
        passwd_path = Path(f"/p4/common/config/.p4passwd.p4_{self.sdp_instance}.admin")
        try:
            with passwd_path.open() as pf:
                passwd = pf.read().strip() + "\n"
        except FileNotFoundError as exc:
            raise FileNotFoundError(f"Password file {passwd_path} not found: {exc}")
        self.run_p4(["login", "-a"], port=port, input_data=passwd, check=True)