152 lines
5.5 KiB
Python
152 lines
5.5 KiB
Python
import subprocess
|
|
from typing import Union, List
|
|
|
|
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 = get_settings()
|
|
|
|
tag_expression: str = config["bumpversion"]["tag_name"].replace(
|
|
"{new_version}", "[0-9]*.[0-9]*.[0-9]*"
|
|
)
|
|
|
|
logger.debug("Tag expression: " + str(tag_expression))
|
|
|
|
# Default version is `0.0.0` or what is found in
|
|
version = self.get_file_version(config)
|
|
|
|
# If a version is found in Perforce labels, use that the latest labeled version
|
|
labeled_versions: Union[List[str], None] = None
|
|
try:
|
|
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 SemverException(
|
|
f"Error getting latest labeled Perforce version: {str(e.stderr).rstrip()}"
|
|
)
|
|
|
|
if len(labeled_versions) > 0 and labeled_versions[-1] != "":
|
|
version = labeled_versions[-1]
|
|
|
|
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()}"
|
|
) |