refactor: introduce a own CompletedProcess subclass with parse_success method

6 files changed, 94 insertions, 79 deletions

diff --git a/src/setuptools_scm/_run_cmd.py b/src/setuptools_scm/_run_cmd.py
index 5fa5f22..e40589f 100644
--- a/src/setuptools_scm/_run_cmd.py
+++ b/src/setuptools_scm/_run_cmd.py
@@ -9,17 +9,68 @@ from typing import Callable
 from typing import Mapping
 from typing import overload
 from typing import Sequence
+from typing import TYPE_CHECKING
 from typing import TypeVar
 
 from . import _log
 from . import _types as _t
 
+if TYPE_CHECKING:
+    BaseCompletedProcess = subprocess.CompletedProcess[str]
+else:
+    BaseCompletedProcess = subprocess.CompletedProcess
+
+
 log = _log.log.getChild("run_cmd")
 
 PARSE_RESULT = TypeVar("PARSE_RESULT")
 T = TypeVar("T")
 
 
+class CompletedProcess(BaseCompletedProcess):
+    @classmethod
+    def from_raw(
+        cls, input: BaseCompletedProcess, strip: bool = True
+    ) -> CompletedProcess:
+        return cls(
+            args=input.args,
+            returncode=input.returncode,
+            stdout=input.stdout.strip() if strip and input.stdout else input.stdout,
+            stderr=input.stderr.strip() if strip and input.stderr else input.stderr,
+        )
+
+    @overload
+    def parse_success(
+        self,
+        parse: Callable[[str], PARSE_RESULT],
+        default: None = None,
+        error_msg: str | None = None,
+    ) -> PARSE_RESULT | None:
+        ...
+
+    @overload
+    def parse_success(
+        self,
+        parse: Callable[[str], PARSE_RESULT],
+        default: T,
+        error_msg: str | None = None,
+    ) -> PARSE_RESULT | T:
+        ...
+
+    def parse_success(
+        self,
+        parse: Callable[[str], PARSE_RESULT],
+        default: T | None = None,
+        error_msg: str | None = None,
+    ) -> PARSE_RESULT | T | None:
+        if self.returncode:
+            if error_msg:
+                log.warning("%s %s", error_msg, self)
+            return default
+        else:
+            return parse(self.stdout)
+
+
 def no_git_env(env: Mapping[str, str]) -> dict[str, str]:
     # adapted from pre-commit
     # Too many bugs dealing with environment variables and GIT:
@@ -77,7 +128,7 @@ def run(
     trace: bool = True,
     timeout: int = 20,
     check: bool = False,
-) -> subprocess.CompletedProcess[str]:
+) -> CompletedProcess:
     if isinstance(cmd, str):
         cmd = shlex.split(cmd)
     else:
@@ -99,10 +150,8 @@ def run(
         text=True,
         timeout=timeout,
     )
-    if strip:
-        if res.stdout:
-            res.stdout = ensure_stripped_str(res.stdout)
-            res.stderr = ensure_stripped_str(res.stderr)
+
+    res = CompletedProcess.from_raw(res, strip=strip)
     if trace:
         if res.stdout:
             log.debug("out:\n%s", textwrap.indent(res.stdout, "    "))
@@ -115,43 +164,6 @@ def run(
     return res
 
 
-@overload
-def parse_success(
-    res: subprocess.CompletedProcess[str],
-    *,
-    parse: Callable[[str], PARSE_RESULT],
-    default: None = None,
-    error_msg: str | None = None,
-) -> PARSE_RESULT | None:
-    ...
-
-
-@overload
-def parse_success(
-    res: subprocess.CompletedProcess[str],
-    *,
-    parse: Callable[[str], PARSE_RESULT],
-    default: T,
-    error_msg: str | None = None,
-) -> PARSE_RESULT | T:
-    ...
-
-
-def parse_success(
-    res: subprocess.CompletedProcess[str],
-    *,
-    parse: Callable[[str], PARSE_RESULT],
-    default: T | None = None,
-    error_msg: str | None = None,
-) -> PARSE_RESULT | T | None:
-    if res.returncode:
-        if error_msg:
-            log.warning("%s %s", error_msg, res)
-        return default
-    else:
-        return parse(res.stdout)
-
-
 def _unsafe_quote_for_display(item: _t.PathT) -> str:
     # give better results than shlex.join in our cases
     text = os.fspath(item)
diff --git a/src/setuptools_scm/git.py b/src/setuptools_scm/git.py
index 6e5bb64..d1da7f8 100644
--- a/src/setuptools_scm/git.py
+++ b/src/setuptools_scm/git.py
@@ -10,7 +10,6 @@ from datetime import date
 from datetime import datetime
 from os.path import samefile
 from pathlib import Path
-from subprocess import CompletedProcess
 from typing import Callable
 from typing import Sequence
 from typing import TYPE_CHECKING
@@ -18,7 +17,7 @@ from typing import TYPE_CHECKING
 from . import _types as _t
 from . import Configuration
 from . import discover
-from ._run_cmd import parse_success as _parse_success
+from ._run_cmd import CompletedProcess as _CompletedProcess
 from ._run_cmd import require_command as _require_command
 from ._run_cmd import run as _run
 from .integration import data_from_mime
@@ -50,7 +49,7 @@ DEFAULT_DESCRIBE = [
 
 def run_git(
     args: Sequence[str | os.PathLike[str]], repo: Path, *, check: bool = False
-) -> CompletedProcess[str]:
+) -> _CompletedProcess:
     return _run(["git", "--git-dir", repo / ".git", *args], cwd=repo, check=check)
 
 
@@ -60,7 +59,7 @@ class GitWorkdir(Workdir):
     @classmethod
     def from_potential_worktree(cls, wd: _t.PathT) -> GitWorkdir | None:
         wd = Path(wd).resolve()
-        real_wd = _parse_success(run_git(["rev-parse", "--show-prefix"], wd), parse=str)
+        real_wd = run_git(["rev-parse", "--show-prefix"], wd).parse_success(parse=str)
         if real_wd is None:
             return None
         else:
@@ -82,21 +81,24 @@ class GitWorkdir(Workdir):
         return cls(Path(real_wd))
 
     def is_dirty(self) -> bool:
-        res = run_git(["status", "--porcelain", "--untracked-files=no"], self.path)
-
-        return _parse_success(
-            res,
+        return run_git(
+            ["status", "--porcelain", "--untracked-files=no"], self.path
+        ).parse_success(
             parse=bool,
             default=False,
         )
 
     def get_branch(self) -> str | None:
-        return _parse_success(
-            run_git(["rev-parse", "--abbrev-ref", "HEAD"], self.path),
+        return run_git(
+            ["rev-parse", "--abbrev-ref", "HEAD"],
+            self.path,
+        ).parse_success(
             parse=str,
             error_msg="branch err (abbrev-err)",
-        ) or _parse_success(
-            run_git(["symbolic-ref", "--short", "HEAD"], self.path),
+        ) or run_git(
+            ["symbolic-ref", "--short", "HEAD"],
+            self.path,
+        ).parse_success(
             parse=str,
             error_msg="branch err (symbolic-ref)",
         )
@@ -116,8 +118,7 @@ class GitWorkdir(Workdir):
             ],
             self.path,
         )
-        return _parse_success(
-            res,
+        return res.parse_success(
             parse=parse_timestamp,
             error_msg="logging the iso date for head failed",
             default=None,
@@ -133,8 +134,9 @@ class GitWorkdir(Workdir):
         def _unsafe_short_node(node: str) -> str:
             return node[:7]
 
-        return _parse_success(
-            run_git(["rev-parse", "--verify", "--quiet", "HEAD"], self.path),
+        return run_git(
+            ["rev-parse", "--verify", "--quiet", "HEAD"], self.path
+        ).parse_success(
             parse=_unsafe_short_node,
         )
 
@@ -142,7 +144,7 @@ class GitWorkdir(Workdir):
         res = run_git(["rev-list", "HEAD"], self.path)
         return res.stdout.count("\n") + 1
 
-    def default_describe(self) -> CompletedProcess[str]:
+    def default_describe(self) -> _CompletedProcess:
         return run_git(DEFAULT_DESCRIBE[1:], self.path)
 
 
@@ -229,7 +231,7 @@ def version_from_describe(
         tag, distance, node, dirty = _git_parse_describe(output)
         return meta(tag=tag, distance=distance, dirty=dirty, node=node, config=config)
 
-    return _parse_success(describe_res, parse=parse_describe)
+    return describe_res.parse_success(parse=parse_describe)
 
 
 def _git_parse_inner(
diff --git a/src/setuptools_scm/hg.py b/src/setuptools_scm/hg.py
index 44fa652..7feec51 100644
--- a/src/setuptools_scm/hg.py
+++ b/src/setuptools_scm/hg.py
@@ -17,17 +17,14 @@ from .version import tag_to_version
 if TYPE_CHECKING:
     from . import _types as _t
 
-from ._run_cmd import run as _run, require_command
+from ._run_cmd import run as _run, require_command as _require_command
 
 log = logging.getLogger(__name__)
 
 
 class HgWorkdir(Workdir):
-    COMMAND = "hg"
-
     @classmethod
     def from_potential_worktree(cls, wd: _t.PathT) -> HgWorkdir | None:
-        require_command(cls.COMMAND)
         res = _run(["hg", "root"], wd)
         if res.returncode:
             return None
@@ -142,6 +139,7 @@ class HgWorkdir(Workdir):
 
 
 def parse(root: _t.PathT, config: Configuration) -> ScmVersion | None:
+    _require_command("hg")
     if os.path.exists(os.path.join(root, ".hg/git")):
         res = _run(["hg", "path"], root)
         if not res.returncode:
diff --git a/src/setuptools_scm/hg_git.py b/src/setuptools_scm/hg_git.py
index 91f9ad1..b6c3036 100644
--- a/src/setuptools_scm/hg_git.py
+++ b/src/setuptools_scm/hg_git.py
@@ -5,9 +5,9 @@ import os
 from contextlib import suppress
 from datetime import date
 from pathlib import Path
-from subprocess import CompletedProcess
 
 from . import _types as _t
+from ._run_cmd import CompletedProcess as _CompletedProcess
 from ._run_cmd import require_command
 from ._run_cmd import run as _run
 from .git import GitWorkdir
@@ -15,7 +15,7 @@ from .hg import HgWorkdir
 
 log = logging.getLogger(__name__)
 
-_FAKE_GIT_DESCRIBE_ERROR = CompletedProcess(
+_FAKE_GIT_DESCRIBE_ERROR = _CompletedProcess(
     "fake git describe output for hg",
     1,
     "<>hg git failed to describe",
@@ -28,10 +28,10 @@ class GitWorkdirHgClient(GitWorkdir, HgWorkdir):
     @classmethod
     def from_potential_worktree(cls, wd: _t.PathT) -> GitWorkdirHgClient | None:
         require_command("hg")
-        res = _run(["hg", "root"], cwd=wd)
-        if res.returncode:
+        res = _run(["hg", "root"], cwd=wd).parse_success(parse=Path)
+        if res is None:
             return None
-        return cls(Path(res.stdout))
+        return cls(res)
 
     def is_dirty(self) -> bool:
         res = _run(["hg", "id", "-T", "{dirty}"], cwd=self.path, check=True)
@@ -45,11 +45,9 @@ class GitWorkdirHgClient(GitWorkdir, HgWorkdir):
         return res.stdout
 
     def get_head_date(self) -> date | None:
-        res = _run('hg log -r . -T "{shortdate(date)}"', cwd=self.path)
-        if res.returncode:
-            log.info("head date err %s", res)
-            return None
-        return date.fromisoformat(res.stdout)
+        return _run('hg log -r . -T "{shortdate(date)}"', cwd=self.path).parse_success(
+            parse=date.fromisoformat, error_msg="head date err"
+        )
 
     def is_shallow(self) -> bool:
         return False
@@ -100,7 +98,7 @@ class GitWorkdirHgClient(GitWorkdir, HgWorkdir):
         res = _run(["hg", "log", "-r", "ancestors(.)", "-T", "."], cwd=self.path)
         return len(res.stdout)
 
-    def default_describe(self) -> CompletedProcess[str]:
+    def default_describe(self) -> _CompletedProcess:
         """
         Tentative to reproduce the output of
 
@@ -149,7 +147,7 @@ class GitWorkdirHgClient(GitWorkdir, HgWorkdir):
         if self.is_dirty():
             desc += "-dirty"
         log.debug("faked describe %r", desc)
-        return CompletedProcess(
+        return _CompletedProcess(
             ["setuptools-scm", "faked", "describe"],
             returncode=0,
             stdout=desc,
diff --git a/src/setuptools_scm/scm_workdir.py b/src/setuptools_scm/scm_workdir.py
index ec6933e..9879549 100644
--- a/src/setuptools_scm/scm_workdir.py
+++ b/src/setuptools_scm/scm_workdir.py
@@ -3,7 +3,13 @@ from __future__ import annotations
 from dataclasses import dataclass
 from pathlib import Path
 
+from ._config import Configuration
+from .version import ScmVersion
+
 
 @dataclass()
 class Workdir:
     path: Path
+
+    def run_describe(self, config: Configuration) -> ScmVersion:
+        raise NotImplementedError(self.run_describe)
diff --git a/testing/test_git.py b/testing/test_git.py
index 53571cb..4ff0872 100644
--- a/testing/test_git.py
+++ b/testing/test_git.py
@@ -22,6 +22,7 @@ from setuptools_scm import Configuration
 from setuptools_scm import git
 from setuptools_scm import NonNormalizedVersion
 from setuptools_scm._file_finders.git import git_find_files
+from setuptools_scm._run_cmd import CompletedProcess
 from setuptools_scm._run_cmd import has_command
 from setuptools_scm._run_cmd import run
 from setuptools_scm.git import archival_to_version
@@ -475,9 +476,7 @@ def test_git_getdate_badgit(
 ) -> None:
     wd.commit_testfile()
     git_wd = git.GitWorkdir(wd.cwd)
-    fake_date_result = subprocess.CompletedProcess(
-        args=[], stdout="%cI", stderr="", returncode=0
-    )
+    fake_date_result = CompletedProcess(args=[], stdout="%cI", stderr="", returncode=0)
     with patch.object(
         git,
         "run_git",