Repository: callowayproject/bump-my-version (python)

Issue:
No commit on mercurial repository
Since version 0.30.0, bump-my-version does not commit changes when asked to on mercurial repositories.

How to reproduce:

```sh
mkdir project
cd project
hg init
bump-my-version sample-config --no-prompt --destination .bumpversion.toml
sed -i -e 's/commit = false/commit = true/g' .bumpversion.toml
hg add .bumpversion.toml
hg commit -m"bump-my-version configuration"
bump-my-version bump minor
```

At that point, local changes for the new version exists but there is no commit done with it. (`hg diff` and `hg log` show this).

There is also a strange `[]` output on the console that was not present with version 0.29.0.

When using version 0.29.0, the commit is correctly created with the changes.

Pull request:
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.

Fixes #317 

Patch:
diff --git a/bumpversion/scm/hg.py b/bumpversion/scm/hg.py
index d091b9f5..a3ecdd6d 100644
--- a/bumpversion/scm/hg.py
+++ b/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 @@ class Mercurial:
         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 @@ class Mercurial:
         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()
diff --git a/bumpversion/utils.py b/bumpversion/utils.py
index a5d8e9ca..dd4e744e 100644
--- a/bumpversion/utils.py
+++ b/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])
diff --git a/pyproject.toml b/pyproject.toml
index 61f41a21..5834afc5 100644
--- a/pyproject.toml
+++ b/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"


Test patch:
diff --git a/tests/test_cli/test_bump.py b/tests/test_cli/test_bump.py
index 7d04b4d9..3b8690f5 100644
--- a/tests/test_cli/test_bump.py
+++ b/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
 
 
diff --git a/tests/test_scm/test_hg.py b/tests/test_scm/test_hg.py
index 017f1734..932f604e 100644
--- a/tests/test_scm/test_hg.py
+++ b/tests/test_scm/test_hg.py
@@ -1,21 +1,233 @@
+"""Tests for the mercurial SCM."""
+
+import json
 import shutil
+import subprocess
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
+from pytest import param
 
-from bumpversion.scm.hg import Mercurial
-from bumpversion.scm.models import SCMConfig
+from bumpversion import cli
+from bumpversion.scm.hg import Mercurial, tag, assert_nondirty
+from bumpversion.scm.models import LatestTagInfo, SCMConfig
+from bumpversion.exceptions import BumpVersionError, DirtyWorkingDirectoryError
+from bumpversion.utils import run_command
 from tests.conftest import inside_dir
 
+BASIC_CONFIG = """
+[tool.bumpversion]
+commit = true
+tag = true
+current_version = "1.0.0"
+parse = "(?P<major>\\\\d+)\\\\.(?P<minor>\\\\d+)\\\\.(?P<patch>\\\\d+)(\\\\-(?P<release>[a-z]+))?"
+serialize = [
+    "{major}.{minor}.{patch}-{release}",
+    "{major}.{minor}.{patch}"
+]
+"""
+
+
+@pytest.fixture
+def hg_instance(scm_config: SCMConfig):
+    """Return a Mercurial instance."""
+    return Mercurial(scm_config)
+
+
+def commit_readme(message: str) -> str:
+    """Commit the readme file and return the commit sha1, so we can tag it later on."""
+    subprocess.run(["hg", "add", "readme.md"], check=True)  # noqa: S603, S607
+    subprocess.run(["hg", "commit", "-m", message], check=True)  # noqa: S603, S607
+    result = subprocess.run(  # noqa: S603
+        ["hg", "id", "-r", "tip"], capture_output=True, check=True, text=True  # noqa: S607
+    )
+    return result.stdout.strip()
+
+
+class TestMercurial:
+    """Tests of the Mercurial class."""
+
+    class TestIsAvailable:
+        """Tests of the is_available method."""
+
+        def test_hg_is_not_available(self, tmp_path: Path, hg_instance: Mercurial) -> None:
+            """Should return false if it is not a mercurial repo."""
+            with inside_dir(tmp_path):
+                assert not hg_instance.is_available()
+
+        def test_hg_is_available(self, hg_repo: Path, hg_instance: Mercurial) -> None:
+            """Should return false if it is not a mercurial repo."""
+            with inside_dir(hg_repo):
+                assert hg_instance.is_available()
+
+    class TestLatestTagInfo:
+        """Test for the Mercurial.latest_tag_info() function."""
+
+        def test_returns_default_if_not_available(self, tmp_path: Path, hg_instance: Mercurial) -> None:
+            """If git is not available, it returns the default LatestTagInfo object, with all None values."""
+            # Arrange
+            expected = LatestTagInfo()
+
+            # Act
+            with inside_dir(tmp_path):
+                info = hg_instance.latest_tag_info()
+
+            # Assert
+            assert info == expected
+
+        def test_returns_correct_commit_and_tag_info(self, hg_repo: Path, hg_instance: Mercurial) -> None:
+            """Should return information that it is a git repo."""
+            # Arrange
+            tag_prefix = "app/"
+            readme = hg_repo.joinpath("readme.md")
+            readme.touch()
+            hg_instance.config.tag_name = f"{tag_prefix}{{new_version}}"
+
+            # Act
+            with inside_dir(hg_repo):
+                commit_readme("first")
+                tag(f"{tag_prefix}0.1.0", sign=False, message="bumpversion")
+                tag_info = hg_instance.latest_tag_info()
+
+            # Assert
+            assert tag_info.commit_sha is not None
+            assert tag_info.current_version == "0.1.0"
+            assert tag_info.current_tag == f"{tag_prefix}0.1.0"
+            assert tag_info.distance_to_latest_tag == 0
+            assert tag_info.branch_name == "default"
+            assert tag_info.short_branch_name == "default"
+            assert tag_info.repository_root == hg_repo
+            assert tag_info.dirty is False
+
+    class TestAddPath:
+        """Tests for the add_path method in the Mercurial class."""
+
+        def test_adds_to_repo_if_file_is_within_repo(self, hg_repo: Path, hg_instance: Mercurial):
+            """The file is added to the repo when it is within the repo root."""
+            # Arrange
+            new_file = hg_repo / "newfile.txt"
+            new_file.write_text("new file")
+
+            # Act
+            with inside_dir(hg_repo):
+                hg_instance.add_path(new_file)
+                result = run_command(["hg", "status"])
 
-def test_hg_is_not_available(tmp_path: Path, scm_config: SCMConfig) -> None:
-    """Should return false if it is not a mercurial repo."""
-    with inside_dir(tmp_path):
-        assert not Mercurial(scm_config).is_available()
+            # Assert
+            assert result.stdout == "A newfile.txt\n"
+
+        @patch("bumpversion.scm.hg.run_command")
+        def test_does_not_add_path_when_not_in_repository(
+            self, mock_run_command, hg_repo: Path, hg_instance: Mercurial, fixtures_path: Path
+        ):
+            """The file is not added to the repo when it is outside the repo root."""
+            # Arrange
+            new_file = fixtures_path / "basic_cfg.toml"
+
+            # Act
+            with inside_dir(hg_repo):
+                hg_instance.add_path(new_file)
+
+            mock_run_command.assert_called_once_with(["hg", "root"])
+
+    class TestGetAllTags:
+        """Tests for the get_all_tags method."""
+
+        def test_returns_correct_tags(self, hg_repo: Path, hg_instance: Mercurial):
+            """Returns all the tags in the repo."""
+            # Arrange
+            readme = hg_repo.joinpath("readme.md")
+            readme.touch()
+            file1 = hg_repo.joinpath("file1.txt")
+            file1.touch()
+            file2 = hg_repo.joinpath("file2.txt")
+            file2.touch()
+            with inside_dir(hg_repo):
+                commit_readme("first")
+                tag("tag1")
+                hg_instance.add_path(file1)
+                run_command(["hg", "commit", "-m", "add file1"])
+                tag("tag2")
+                hg_instance.add_path(file2)
+                run_command(["hg", "commit", "-m", "add file2"])
+                tag("tag3")
+
+                # Act
+                tags = hg_instance.get_all_tags()
+
+            # Assert
+            assert set(tags) == {"tip", "tag1", "tag2", "tag3"}
+
+        def test_handles_empty_tags(self, hg_repo: Path, hg_instance: Mercurial):
+            """When there are no tags, an empty list is returned."""
+            # Arrange
+            with inside_dir(hg_repo):
+                hg_repo.joinpath("readme.md").touch()
+                commit_readme("first")
+
+                # Act
+                tags = hg_instance.get_all_tags()
+
+            # Assert
+            assert set(tags) == {"tip"}
+
+        @pytest.mark.parametrize(
+            ["side_effect"],
+            [
+                param(subprocess.CalledProcessError(1, "cmd"), id="command failure"),
+                param(FileNotFoundError, id="file not found"),
+                param(PermissionError, id="permission error"),
+            ],
+        )
+        def test_failure_raises_error(self, hg_instance, side_effect, mocker):
+            """On failure, it raises a BumpVersionError exception."""
+            # Arrange
+            mocker.patch("bumpversion.utils.run_command", side_effect=side_effect)
+
+            # Act, Assert
+            with pytest.raises(BumpVersionError):
+                hg_instance.get_all_tags()
 
 
 @pytest.mark.skipif(not shutil.which("hg"), reason="Mercurial is not available.")
-def test_hg_is_available(hg_repo: Path, scm_config: SCMConfig) -> None:
-    """Should return false if it is not a mercurial repo."""
+def test_hg_bump_version_commits_changes(hg_repo: Path, fixtures_path: Path, runner) -> None:
+    """Should commit changes to the repo when configured."""
+    dest_cfg = hg_repo / ".bumpversion.toml"
+    dest_cfg.write_text(BASIC_CONFIG)
+
     with inside_dir(hg_repo):
-        assert Mercurial(scm_config).is_available()
+        run_command(["hg", "add", ".bumpversion.toml"])
+        run_command(["hg", "commit", "-m", "add configuration"])
+        runner.invoke(
+            cli.cli,
+            ["bump", "minor", "-vv"],
+        )
+        hg_result = run_command(["hg", "tags", "-T", "json"])
+
+    hg_logs = json.loads(hg_result.stdout)
+    tags = [item["tag"] for item in hg_logs]
+    assert "v1.1.0" in tags
+
+
+class TestAssertNonDirty:
+    """Tests for the assert_nondirty function."""
+
+    def test_recognizes_clean_working_directory(self, hg_repo: Path):
+        """A clean repo does not raise an exception."""
+        with inside_dir(hg_repo):
+            assert_nondirty()
+
+    def test_dirty_working_directory_raises_exception(self, hg_repo: Path):
+        """A dirty working directory raises an exception."""
+        # Arrange
+        readme = hg_repo.joinpath("readme.md")
+        readme.touch()
+        with inside_dir(hg_repo):
+            commit_readme("first")
+            readme.write_text("new text")
+            with pytest.raises(DirtyWorkingDirectoryError) as exc:
+                assert_nondirty()
+
+        assert "Mercurial working directory is not clean" in str(exc.value)
+        assert "M readme.md" in str(exc.value)