diff --git a/semver/scm/perforce.py b/semver/scm/perforce.py index a82a348..37ee8e3 100644 --- a/semver/scm/perforce.py +++ b/semver/scm/perforce.py @@ -5,18 +5,32 @@ import toml from semver.scm import SCM from semver.logger import logger +from semver.utils import get_settings +from semver.exceptions import SemverException class Perforce(SCM): def __init__(self) -> None: + self.p4_bin = "p4" + super().__init__() + def _run_command( + self, *args: str, throwExceptions: bool = True + ) -> subprocess.CompletedProcess[str]: + return subprocess.run( + args, + capture_output=True, + text=True, + check=throwExceptions, + ) + def get_tag_version(self) -> str: """ Get the latest tagged version from Perforce labels :return: The latest tagged version """ - config: dict = toml.load("./.bumpversion.cfg") + config: dict = get_settings() tag_expression: str = config["bumpversion"]["tag_name"].replace( "{new_version}", "[0-9]*.[0-9]*.[0-9]*" @@ -30,15 +44,10 @@ class Perforce(SCM): # If a version is found in Perforce labels, use that the latest labeled version labeled_versions: Union[List[str], None] = None try: - proc = subprocess.run( - ["p4", "labels", "-e", tag_expression, "-m1"], - capture_output=True, - text=True, - check=True, - ) + proc = self._run_command(self.p4_bin, "labels", "-e", tag_expression, "-m1") labeled_versions = proc.stdout.rstrip().split("\n") except subprocess.CalledProcessError as e: - raise RuntimeError( + raise SemverException( f"Error getting latest labeled Perforce version: {str(e.stderr).rstrip()}" ) @@ -47,3 +56,97 @@ class Perforce(SCM): logger.debug("Label Version: " + str(version)) return version + + def get_branch(self) -> str: + """ + Returns the name of the current Perforce stream + :return: The current Perforce stream + """ + try: + proc = self._run_command(self.p4_bin, "client", "-o") + client = toml.loads(proc.stdout) + return client["Stream"] + except subprocess.CalledProcessError as e: + raise SemverException( + f"Error getting current Perforce stream: {str(e.stderr).rstrip()}" + ) + + def get_merge_branch(self) -> Union[str, None]: + """ + Returns the name of the stream being intergrated into mainline + :return: The Perforce stream being intergrated into mainline + """ + try: + proc = self._run_command(self.p4_bin, "client", "-o") + client = toml.loads(proc.stdout) + return client["StreamAtChange"] + except subprocess.CalledProcessError as e: + raise SemverException( + f"Error getting current Perforce stream: {str(e.stderr).rstrip()}" + ) + + def commit_and_push(self, branch: str) -> None: + """ + Creates a changelist and submits it to Perforce + :param branch: The current Perforce stream + """ + try: + proc = self._run_command(self.p4_bin, "change", "-o") + change = toml.loads(proc.stdout) + change["Description"] = f"Version Bump" + change["Files"] = [f"//{branch}/..."] + proc = subprocess.Popen([self.p4_bin, "change", "-i"], stdin=subprocess.PIPE) + proc.communicate(input=toml.dumps(change).encode()) + proc.wait() + proc = self._run_command(self.p4_bin, "submit", "-c", str(change["Change"])) + except subprocess.CalledProcessError as e: + raise SemverException( + f"Error creating a CL and submitting to Perforce: {str(e.stderr).rstrip()}" + ) + + def tag_version(self, version: str) -> None: + """ + Creates a Perforce label + :param version: The version to label + """ + try: + proc = self._run_command(self.p4_bin, "label", "-o", version) + label = toml.loads(proc.stdout) + label["Description"] = f"Version {version}" + label["Options"] = "locked" + label["Revision"] = f"//{self.get_branch()}/..." + proc = subprocess.Popen([self.p4_bin, "label", "-i"], stdin=subprocess.PIPE) + proc.communicate(input=toml.dumps(label).encode()) + proc.wait() + except subprocess.CalledProcessError as e: + raise SemverException( + f"Error creating a Perforce label: {str(e.stderr).rstrip()}" + ) + def get_version_hash(self, version: str) -> str: + """ + Returns the hash of the version label + :param version: The version label + :return: The hash of the version label + """ + try: + proc = self._run_command(self.p4_bin, "labels", "-e", version, "-m1") + label = toml.loads(proc.stdout) + return label["Revision"] + except subprocess.CalledProcessError as e: + raise SemverException( + f"Error getting hash of version label: {str(e.stderr).rstrip()}" + ) + + def get_hash(self) -> str: + """ + Returns the hash of the current commit + :return: The hash of the current commit + """ + try: + proc = self._run_command(self.p4_bin, "changes", "-m1") + change = toml.loads(proc.stdout) + return change["Change"] + except subprocess.CalledProcessError as e: + raise SemverException( + f"Error getting hash of current commit: {str(e.stderr).rstrip()}" + ) \ No newline at end of file