Merge pull request #325 from callowayproject/317-no-commit-on-mercurial-repository

Refactor Mercurial SCM support and improve test coverage

Author: coordt

bumpversion/scm/hg.py

@@ -1,12 +1,18 @@
 """Mercurial source control management."""
 
+import json
+import os
+import re
+import shlex
 import subprocess
+from pathlib import Path
+from tempfile import NamedTemporaryFile
 from typing import ClassVar, List, MutableMapping, Optional
 
 from bumpversion.exceptions import DirtyWorkingDirectoryError, SignedTagsError
 from bumpversion.scm.models import LatestTagInfo, SCMConfig
 from bumpversion.ui import get_indented_logger
-from bumpversion.utils import Pathlike, run_command
+from bumpversion.utils import Pathlike, format_and_raise_error, is_subpath, run_command
 
 logger = get_indented_logger(__name__)
 
@@ -16,7 +22,6 @@ class Mercurial:
 
     _TEST_AVAILABLE_COMMAND: ClassVar[List[str]] = ["hg", "root"]
     _COMMIT_COMMAND: ClassVar[List[str]] = ["hg", "commit", "--logfile"]
-    _ALL_TAGS_COMMAND: ClassVar[List[str]] = ["hg", "log", '--rev="tag()"', '--template="{tags}\n"']
 
     def __init__(self, config: SCMConfig):
         self.config = config
@@ -50,14 +55,83 @@ def latest_tag_info(self) -> LatestTagInfo:
         self._latest_tag_info = LatestTagInfo(**info)
         return self._latest_tag_info
 
+    def get_all_tags(self) -> List[str]:
+        """Return all tags in a mercurial repository."""
+        try:
+            result = run_command(["hg", "tags", "-T", "json"])
+            tags = json.loads(result.stdout) if result.stdout else []
+            return [tag["tag"] for tag in tags]
+        except (
+            FileNotFoundError,
+            PermissionError,
+            NotADirectoryError,
+            subprocess.CalledProcessError,
+        ) as e:
+            format_and_raise_error(e)
+            return []
+
     def add_path(self, path: Pathlike) -> None:
         """Add a path to the Source Control Management repository."""
-        pass
+        repository_root = self.latest_tag_info().repository_root
+        if not (repository_root and is_subpath(repository_root, path)):
+            return
+
+        cwd = Path.cwd()
+        temp_path = os.path.relpath(path, cwd)
+        try:
+            run_command(["hg", "add", str(temp_path)])
+        except subprocess.CalledProcessError as e:  # pragma: no-cover
+            format_and_raise_error(e)
 
     def commit_and_tag(self, files: List[Pathlike], context: MutableMapping, dry_run: bool = False) -> None:
         """Commit and tag files to the repository using the configuration."""
+        if dry_run:
+            return
+
+        if self.config.commit:
+            for path in files:
+                self.add_path(path)
+
+            self.commit(context)
+
+        if self.config.tag:
+            tag_name = self.config.tag_name.format(**context)
+            tag_message = self.config.tag_message.format(**context)
+            tag(tag_name, sign=self.config.sign_tags, message=tag_message)
+
+            # for m_tag_name in self.config.moveable_tags:
+            #     moveable_tag(m_tag_name)
+
+    def commit(self, context: MutableMapping) -> None:
+        """Commit the changes."""
+        extra_args = shlex.split(self.config.commit_args) if self.config.commit_args else []
+
+        current_version = context.get("current_version", "")
+        new_version = context.get("new_version", "")
+        commit_message = self.config.message.format(**context)
 
-    def tag(self, name: str, sign: bool = False, message: Optional[str] = None) -> None:
+        if not current_version:  # pragma: no-coverage
+            logger.warning("No current version given, using an empty string.")
+        if not new_version:  # pragma: no-coverage
+            logger.warning("No new version given, using an empty string.")
+
+        with NamedTemporaryFile("wb", delete=False) as f:
+            f.write(commit_message.encode("utf-8"))
+
+        env = os.environ.copy()
+        env["BUMPVERSION_CURRENT_VERSION"] = current_version
+        env["BUMPVERSION_NEW_VERSION"] = new_version
+
+        try:
+            cmd = [*self._COMMIT_COMMAND, f.name, *extra_args]
+            run_command(cmd, env=env)
+        except (subprocess.CalledProcessError, TypeError) as exc:  # pragma: no-coverage
+            format_and_raise_error(exc)
+        finally:
+            os.unlink(f.name)
+
+    @staticmethod
+    def tag(name: str, sign: bool = False, message: Optional[str] = None) -> None:
         """
         Create a tag of the new_version in VCS.
 
@@ -72,14 +146,10 @@ def tag(self, name: str, sign: bool = False, message: Optional[str] = None) -> N
         Raises:
             SignedTagsError: If ``sign`` is ``True``
         """
-        command = ["hg", "tag", name]
-        if sign:
-            raise SignedTagsError("Mercurial does not support signed tags.")
-        if message:
-            command += ["--message", message]
-        run_command(command)
-
-    def assert_nondirty(self) -> None:
+        tag(name, sign=sign, message=message)
+
+    @staticmethod
+    def assert_nondirty() -> None:
         """Assert that the working directory is clean."""
         assert_nondirty()
 
@@ -95,21 +165,53 @@ def commit_info(config: SCMConfig) -> dict:
         A dictionary containing information about the latest commit.
     """
     tag_pattern = config.tag_name.replace("{new_version}", ".*")
-    info = dict.fromkeys(["dirty", "commit_sha", "distance_to_latest_tag", "current_version", "current_tag"])
+    info = dict.fromkeys(
+        [
+            "dirty",
+            "commit_sha",
+            "distance_to_latest_tag",
+            "current_version",
+            "current_tag",
+            "branch_name",
+            "short_branch_name",
+            "repository_root",
+        ]
+    )
+
     info["distance_to_latest_tag"] = 0
-    result = run_command(["hg", "log", "-r", f"tag('re:{tag_pattern}')", "--template", "{latesttag}\n"])
-    result.check_returncode()
+    result = run_command(["hg", "log", "-r", f"tag('re:{tag_pattern}')", "-T", "json"])
+    repo_path = run_command(["hg", "root"]).stdout.strip()
+
+    output_info = parse_commit_log(result.stdout, config)
+    info |= output_info
 
-    if result.stdout:
-        tag_string = result.stdout.splitlines(keepends=False)[-1]
-        info["current_version"] = config.get_version_from_tag(tag_string)
-    else:
+    if not output_info:
         logger.debug("No tags found")
 
+    info["repository_root"] = Path(repo_path)
     info["dirty"] = len(run_command(["hg", "status", "-mard"]).stdout) != 0
     return info
 
 
+def parse_commit_log(log_string: str, config: SCMConfig) -> dict:
+    """Parse the commit log string."""
+    output_info = json.loads(log_string) if log_string else {}
+    if not output_info:
+        return {}
+    first_rev = output_info[0]
+    branch_name = first_rev["branch"]
+    short_branch_name = re.sub(r"([^a-zA-Z0-9]*)", "", branch_name).lower()[:20]
+
+    return {
+        "current_version": config.get_version_from_tag(first_rev["tags"][0]),
+        "current_tag": first_rev["tags"][0],
+        "commit_sha": first_rev["node"],
+        "distance_to_latest_tag": 0,
+        "branch_name": branch_name,
+        "short_branch_name": short_branch_name,
+    }
+
+
 def tag(name: str, sign: bool = False, message: Optional[str] = None) -> None:
     """
     Create a tag of the new_version in VCS.
@@ -135,7 +237,6 @@ def tag(name: str, sign: bool = False, message: Optional[str] = None) -> None:
 
 def assert_nondirty() -> None:
     """Assert that the working directory is clean."""
-    print(run_command(["hg", "status", "-mard"]).stdout.splitlines())
     if lines := [
         line.strip()
         for line in run_command(["hg", "status", "-mard"]).stdout.splitlines()

bumpversion/utils.py

@@ -116,7 +116,7 @@ def set_nested_value(d: dict, value: Any, path: str) -> None:
             current_element = current_element[key]
 
 
-def format_and_raise_error(exc: Union[TypeError, subprocess.CalledProcessError]) -> None:
+def format_and_raise_error(exc: Union[BaseException, subprocess.CalledProcessError]) -> None:
     """Format the error message from an exception and re-raise it as a BumpVersionError."""
     if isinstance(exc, subprocess.CalledProcessError):
         output = "\n".join([x for x in [exc.stdout, exc.stderr] if x])

pyproject.toml

@@ -106,11 +106,15 @@ omit = [
     "*.tox*",
 ]
 show_missing = true
-exclude_lines = [
+exclude_also = [
     "raise NotImplementedError",
     "pragma: no-coverage",
     "pragma: no-cov",
+    "def __str__",
+    "def __repr__",
 ]
+skip_covered = true
+skip_empty = true
 
 [tool.coverage.html]
 directory = "test-reports/htmlcov"

tests/test_cli/test_bump.py

@@ -183,8 +183,6 @@ def test_dirty_work_dir_raises_error(repo: str, scm_command: str, request, runne
         )
 
     # Assert
-    print(f"{result.output=}")
-    assert result.exit_code != 0
     assert "working directory is not clean" in result.output