Refactor Mercurial SCM support and improve test coverage

Revamped the Mercurial SCM implementation with new features including full tag retrieval, commit handling, and clean working directory assertion. Enhanced test suite with new Mercurial-specific tests for functionality and edge cases.

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:
+            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