p4 rough draft

.github/workflows/unittests.yaml

@@ -1,75 +0,0 @@
-name: Pytest
-
-on:
-  push:
-    branches:
-      - master
-      - feature/refactor
-  pull_request: {}
-
-jobs:
-  test:
-    name: Unit Tests
-    runs-on: ${{ matrix.os }}
-
-    strategy:
-      matrix:
-        os: [ubuntu-22.04, windows-latest, macos-latest]
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
-
-    steps:
-    - name: Checkout
-      uses: actions/checkout@v3
-
-    - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v2
-      with:
-        python-version: ${{ matrix.python-version }}
-        cache: pip
-
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install pytest pytest-cov
-        pip install -r requirements.txt
-    
-    - name: Test
-      run: |
-        pytest
-
-  coverage:
-    name: Coverage Summary
-    runs-on: ubuntu-22.04
-    permissions:
-      pull-requests: write
-      contents: read
-
-    if: github.event_name == 'pull_request'
-    strategy:
-      fail-fast: false
-
-    steps:
-    - name: Checkout
-      uses: actions/checkout@v3
-
-    - name: Set up Python
-      uses: actions/setup-python@v2
-      with:
-        python-version: "3.11"
-        cache: pip
-
-    - name: Install dependencies
-      run: |
-        python -m pip install --upgrade pip
-        pip install pytest pytest-cov
-        pip install -r requirements.txt
-
-    - name: Run Coverage
-      run: |
-        pytest --junitxml=pytest.xml --cov-report=term-missing:skip-covered --cov=semver | tee pytest-coverage.txt
-
-    - name: Comment Result Summary
-      uses: MishaKav/pytest-coverage-comment@main
-      with:
-        pytest-coverage-path: ./pytest-coverage.txt
-        junitxml-path: ./pytest.xml

semver/scm/git.py

@@ -1,7 +1,7 @@
 import re
 import subprocess
 from typing import Union, List
-from functools import lru_cache
+from functools import cache
 
 import toml
 
@@ -9,7 +9,6 @@ from semver.scm import SCM
 from semver.logger import logger
 from semver.version_type import VersionType
 from semver.exceptions import SemverException
-from semver.utils import get_settings
 
 
 class Git(SCM):
@@ -29,7 +28,7 @@ class Git(SCM):
 
     def _run_command(
         self, *args: str, throwExceptions: bool = True
-    ) -> subprocess.CompletedProcess:
+    ) -> subprocess.CompletedProcess[str]:
         return subprocess.run(
             args,
             capture_output=True,
@@ -59,7 +58,7 @@ class Git(SCM):
         Get the latest tagged version from git tags
         :return: The latest tagged version
         """
-        config: dict = get_settings()
+        config: dict = toml.load("./.bumpversion.cfg")
 
         tag_expression: str = config["bumpversion"]["tag_name"].replace(
             "{new_version}", "[0-9]*.[0-9]*.[0-9]*"
@@ -82,26 +81,21 @@ class Git(SCM):
                 f"Error getting latest tagged git version: {str(e.stderr).rstrip()}"
             )
 
-        if (
+        if len(tagged_versions) > 0 and tagged_versions[-1] != "":
-            tagged_versions is not None
-            and len(tagged_versions) > 0
-            and tagged_versions[-1] != ""
-        ):
             version = tagged_versions[-1]
 
         logger.debug(f"Tag Version: {version}")
         return version
 
-    @lru_cache(maxsize=None)
+    @cache
     def get_branch(self) -> str:
         """
         Get the main branch
-        :return: The main branch
         """
         proc = self._run_command(self.git_bin, "rev-parse", "--abbrev-ref", "HEAD")
         return proc.stdout.rstrip()
 
-    @lru_cache(maxsize=None)
+    @cache
     def get_merge_branch(self) -> Union[str, None]:
         """
         Get the branches involved in the merge

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(
+            proc = self._run_command(self.p4_bin, "labels", "-e", tag_expression, "-m1")
-                ["p4", "labels", "-e", tag_expression, "-m1"],
-                capture_output=True,
-                text=True,
-                check=True,
-            )
             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()}"
+            )

semver/tests/test_semver.py

@@ -67,8 +67,10 @@ class TestSemVer(unittest.TestCase):
     @mock.patch("toml.load")
     @mock.patch("pathlib.Path.is_file")
     @mock.patch("builtins.open", mock.mock_open())
+    @mock.patch("semver.logger.warning")
     def test_update_file_version(
         self,
+        mock_logger: mock.Mock,
         mock_path_is_file: mock.Mock,
         mock_toml_load: mock.Mock,
     ):
@@ -81,9 +83,11 @@ class TestSemVer(unittest.TestCase):
         }
         mock_path_is_file.return_value = True
         self.semver._update_file_version("1.0.1", "1.0.0")
+        mock_logger.assert_not_called()
 
         mock_path_is_file.return_value = False
         self.semver._update_file_version("1.0.1", "1.0.0")
+        mock_logger.assert_called_once()
 
     @mock.patch("semver.semver.SemVer._version_repo", mock.MagicMock())
     def test_run_ok(self):

semver/utils.py

@@ -1,6 +1,6 @@
 from typing import List
 from pathlib import Path
-from functools import lru_cache
+from functools import cache
 import configparser
 
 import toml
@@ -8,7 +8,7 @@ import toml
 from semver.exceptions import SemverException
 
 
-@lru_cache(maxsize=None)
+@cache
 def get_settings() -> dict:
     """
     Get the settings from the config file

setup.cfg

@@ -24,9 +24,7 @@ classifiers =
 [options]
 include_package_data = True
 packages = find:
-python_requires = >=3.8, <4
+python_requires = >=3.7, <4
-install_requires =
-    toml
 
 [options.packages.find]
 exclude =