introduce run_hg and remove the need for module reloading for the hg command env var

Author: RonnyPfannschmidt

src/setuptools_scm/_file_finders/hg.py

@@ -7,19 +7,17 @@
 from .. import _types as _t
 from .._file_finders import is_toplevel_acceptable
 from .._file_finders import scm_find_files
-from .._run_cmd import run as _run
+from ..hg import run_hg
 from ..integration import data_from_mime
 from .pathtools import norm_real
 
 log = logging.getLogger(__name__)
 
-HG_COMMAND = os.environ.get("SETUPTOOLS_SCM_HG_COMMAND", "hg")
-
 
 def _hg_toplevel(path: str) -> str | None:
     try:
-        return _run(
-            [HG_COMMAND, "root"],
+        return run_hg(
+            ["root"],
             cwd=(path or "."),
             check=True,
         ).parse_success(norm_real)
@@ -34,7 +32,7 @@ def _hg_toplevel(path: str) -> str | None:
 def _hg_ls_files_and_dirs(toplevel: str) -> tuple[set[str], set[str]]:
     hg_files: set[str] = set()
     hg_dirs = {toplevel}
-    res = _run([HG_COMMAND, "files"], cwd=toplevel)
+    res = run_hg(["files"], cwd=toplevel)
     if res.returncode:
         return set(), set()
     for name in res.stdout.splitlines():

src/setuptools_scm/hg.py

@@ -6,6 +6,7 @@
 
 from pathlib import Path
 from typing import TYPE_CHECKING
+from typing import Any
 
 from . import Configuration
 from ._version_cls import Version
@@ -19,18 +20,28 @@
 if TYPE_CHECKING:
     from . import _types as _t
 
+from ._run_cmd import CompletedProcess
 from ._run_cmd import require_command as _require_command
 from ._run_cmd import run as _run
 
 log = logging.getLogger(__name__)
 
-HG_COMMAND = os.environ.get("SETUPTOOLS_SCM_HG_COMMAND", "hg")
+
+def _get_hg_command() -> str:
+    """Get the hg command from environment, allowing runtime configuration."""
+    return os.environ.get("SETUPTOOLS_SCM_HG_COMMAND", "hg")
+
+
+def run_hg(args: list[str], cwd: _t.PathT, **kwargs: Any) -> CompletedProcess:
+    """Run mercurial command with the configured hg executable."""
+    cmd = [_get_hg_command(), *args]
+    return _run(cmd, cwd=cwd, **kwargs)
 
 
 class HgWorkdir(Workdir):
     @classmethod
     def from_potential_worktree(cls, wd: _t.PathT) -> HgWorkdir | None:
-        res = _run([HG_COMMAND, "root"], wd)
+        res = run_hg(["root"], wd)
         if res.returncode:
             return None
         return cls(Path(res.stdout))
@@ -79,8 +90,8 @@ def _get_node_info(self) -> tuple[str, str, str] | None:
 
     def _get_branch_info(self) -> tuple[str, bool, str]:
         """Get branch name, dirty status, and dirty date."""
-        branch, dirty_str, dirty_date = _run(
-            [HG_COMMAND, "id", "-T", "{branch}\n{if(dirty, 1, 0)}\n{date|shortdate}"],
+        branch, dirty_str, dirty_date = run_hg(
+            ["id", "-T", "{branch}\n{if(dirty, 1, 0)}\n{date|shortdate}"],
             cwd=self.path,
             check=True,
         ).stdout.split("\n")
@@ -181,9 +192,9 @@ def _get_distance_based_version(
             return None
 
     def hg_log(self, revset: str, template: str) -> str:
-        cmd = [HG_COMMAND, "log", "-r", revset, "-T", template]
-
-        return _run(cmd, cwd=self.path, check=True).stdout
+        return run_hg(
+            ["log", "-r", revset, "-T", template], cwd=self.path, check=True
+        ).stdout
 
     def get_latest_normalizable_tag(self) -> str | None:
         # Gets all tags containing a '.' (see #229) from oldest to newest
@@ -223,12 +234,12 @@ def get_dirty_tag_date(self) -> datetime.date | None:
         """
         try:
             # Check if working directory is dirty first
-            res = _run([HG_COMMAND, "id", "-T", "{dirty}"], cwd=self.path)
+            res = run_hg(["id", "-T", "{dirty}"], cwd=self.path)
             if res.returncode != 0 or not bool(res.stdout):
                 return None
 
             # Get list of changed files using hg status
-            status_res = _run([HG_COMMAND, "status", "-m", "-a", "-r"], cwd=self.path)
+            status_res = run_hg(["status", "-m", "-a", "-r"], cwd=self.path)
             if status_res.returncode != 0:
                 return None
 
@@ -248,9 +259,10 @@ def get_dirty_tag_date(self) -> datetime.date | None:
 
 
 def parse(root: _t.PathT, config: Configuration) -> ScmVersion | None:
-    _require_command(HG_COMMAND)
+    hg_cmd = _get_hg_command()
+    _require_command(hg_cmd)
     if os.path.exists(os.path.join(root, ".hg/git")):
-        res = _run([HG_COMMAND, "path"], root)
+        res = run_hg(["path"], root)
         if not res.returncode:
             for line in res.stdout.split("\n"):
                 if line.startswith("default ="):

src/setuptools_scm/hg_git.py

@@ -9,10 +9,9 @@
 
 from . import _types as _t
 from ._run_cmd import CompletedProcess as _CompletedProcess
-from ._run_cmd import run as _run
 from .git import GitWorkdir
-from .hg import HG_COMMAND
 from .hg import HgWorkdir
+from .hg import run_hg
 from .scm_workdir import get_latest_file_mtime
 
 log = logging.getLogger(__name__)
@@ -27,25 +26,25 @@
 class GitWorkdirHgClient(GitWorkdir, HgWorkdir):
     @classmethod
     def from_potential_worktree(cls, wd: _t.PathT) -> GitWorkdirHgClient | None:
-        res = _run([HG_COMMAND, "root"], cwd=wd).parse_success(parse=Path)
+        res = run_hg(["root"], cwd=wd).parse_success(parse=Path)
         if res is None:
             return None
         return cls(res)
 
     def is_dirty(self) -> bool:
-        res = _run([HG_COMMAND, "id", "-T", "{dirty}"], cwd=self.path, check=True)
+        res = run_hg(["id", "-T", "{dirty}"], cwd=self.path, check=True)
         return bool(res.stdout)
 
     def get_branch(self) -> str | None:
-        res = _run([HG_COMMAND, "id", "-T", "{bookmarks}"], cwd=self.path)
+        res = run_hg(["id", "-T", "{bookmarks}"], cwd=self.path)
         if res.returncode:
             log.info("branch err %s", res)
             return None
         return res.stdout
 
     def get_head_date(self) -> date | None:
-        return _run(
-            [HG_COMMAND, "log", "-r", ".", "-T", "{shortdate(date)}"], cwd=self.path
+        return run_hg(
+            ["log", "-r", ".", "-T", "{shortdate(date)}"], cwd=self.path
         ).parse_success(parse=date.fromisoformat, error_msg="head date err")
 
     def get_dirty_tag_date(self) -> date | None:
@@ -59,7 +58,7 @@ def get_dirty_tag_date(self) -> date | None:
 
         try:
             # Get list of changed files using hg status
-            status_res = _run([HG_COMMAND, "status", "-m", "-a", "-r"], cwd=self.path)
+            status_res = run_hg(["status", "-m", "-a", "-r"], cwd=self.path)
             if status_res.returncode != 0:
                 return None
 
@@ -84,7 +83,7 @@ def fetch_shallow(self) -> None:
         pass
 
     def get_hg_node(self) -> str | None:
-        res = _run([HG_COMMAND, "log", "-r", ".", "-T", "{node}"], cwd=self.path)
+        res = run_hg(["log", "-r", ".", "-T", "{node}"], cwd=self.path)
         if res.returncode:
             return None
         else:
@@ -108,7 +107,7 @@ def node(self) -> str | None:
 
         if git_node is None:
             # trying again after hg -> git
-            _run([HG_COMMAND, "gexport"], cwd=self.path)
+            run_hg(["gexport"], cwd=self.path)
             git_node = self._hg2git(hg_node)
 
             if git_node is None:
@@ -123,7 +122,7 @@ def node(self) -> str | None:
         return git_node[:7]
 
     def count_all_nodes(self) -> int:
-        res = _run([HG_COMMAND, "log", "-r", "ancestors(.)", "-T", "."], cwd=self.path)
+        res = run_hg(["log", "-r", "ancestors(.)", "-T", "."], cwd=self.path)
         return len(res.stdout)
 
     def default_describe(self) -> _CompletedProcess:
@@ -133,9 +132,8 @@ def default_describe(self) -> _CompletedProcess:
         `git describe --dirty --tags --long --match *[0-9]*`
 
         """
-        res = _run(
+        res = run_hg(
             [
-                HG_COMMAND,
                 "log",
                 "-r",
                 "(reverse(ancestors(.)) and tag(r're:v?[0-9].*'))",
@@ -163,7 +161,7 @@ def default_describe(self) -> _CompletedProcess:
             logging.warning("tag not found hg=%s git=%s", hg_tags, git_tags)
             return _FAKE_GIT_DESCRIBE_ERROR
 
-        res = _run([HG_COMMAND, "log", "-r", f"'{tag}'::.", "-T", "."], cwd=self.path)
+        res = run_hg(["log", "-r", f"'{tag}'::.", "-T", "."], cwd=self.path)
         if res.returncode:
             return _FAKE_GIT_DESCRIBE_ERROR
         distance = len(res.stdout) - 1

testing/test_file_finder.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import importlib
 import os
 import sys
 
@@ -9,7 +8,6 @@
 import pytest
 
 from setuptools_scm._file_finders import find_files
-from setuptools_scm._file_finders import hg
 
 from .wd_wrapper import WorkDir
 
@@ -275,6 +273,5 @@ def test_hg_command_from_env(
     with monkeypatch.context() as m:
         m.setenv("SETUPTOOLS_SCM_HG_COMMAND", hg_exe)
         m.setenv("PATH", str(hg_wd.cwd / "not-existing"))
-        request.addfinalizer(lambda: importlib.reload(hg))
-        importlib.reload(hg)
+        # No module reloading needed - runtime configuration works immediately
         assert set(find_files()) == {"file"}

testing/test_hg_git.py

@@ -1,12 +1,8 @@
 from __future__ import annotations
 
-import importlib
-
 import pytest
 
 from setuptools_scm import Configuration
-from setuptools_scm import hg
-from setuptools_scm import hg_git
 from setuptools_scm._run_cmd import CommandNotFoundError
 from setuptools_scm._run_cmd import has_command
 from setuptools_scm._run_cmd import run
@@ -113,9 +109,6 @@ def test_hg_command_from_env(
     with monkeypatch.context() as m:
         m.setenv("SETUPTOOLS_SCM_HG_COMMAND", hg_exe)
         m.setenv("PATH", str(wd.cwd / "not-existing"))
-        request.addfinalizer(lambda: importlib.reload(hg))
-        request.addfinalizer(lambda: importlib.reload(hg_git))
-        importlib.reload(hg)
-        importlib.reload(hg_git)
+        # No module reloading needed - runtime configuration works immediately
         wd.write("pyproject.toml", "[tool.setuptools_scm]")
         assert wd.get_version().startswith("0.1.dev0+")

testing/test_mercurial.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import importlib
 import os
 
 from pathlib import Path
@@ -10,7 +9,6 @@
 import setuptools_scm._file_finders
 
 from setuptools_scm import Configuration
-from setuptools_scm import hg
 from setuptools_scm._run_cmd import CommandNotFoundError
 from setuptools_scm._run_cmd import has_command
 from setuptools_scm.hg import archival_to_version
@@ -78,19 +76,19 @@ def test_hg_command_from_env(
     with monkeypatch.context() as m:
         m.setenv("SETUPTOOLS_SCM_HG_COMMAND", hg_exe)
         m.setenv("PATH", str(wd.cwd / "not-existing"))
-        request.addfinalizer(lambda: importlib.reload(hg))
-        importlib.reload(hg)
         wd.write("pyproject.toml", "[tool.setuptools_scm]")
-        assert wd.get_version() == "0.0"
+        # Need to commit something first for versioning to work
+        wd.commit_testfile()
+        version = wd.get_version()
+        assert version.startswith("0.1.dev1+")
 
 
 def test_hg_command_from_env_is_invalid(
     wd: WorkDir, monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest
 ) -> None:
     with monkeypatch.context() as m:
         m.setenv("SETUPTOOLS_SCM_HG_COMMAND", str(wd.cwd / "not-existing"))
-        request.addfinalizer(lambda: importlib.reload(hg))
-        importlib.reload(hg)
+        # No module reloading needed - runtime configuration works immediately
         config = Configuration()
         wd.write("pyproject.toml", "[tool.setuptools_scm]")
         with pytest.raises(CommandNotFoundError, match=r"hg"):