Skip to content

Commit 58603a3

Browse files
committed
cmd(fix[run-args]): normalize scalar command inputs
why: Fix the mypy arg-type failure and prevent scalar command strings from being split into character args. what: - widen _internal.run typing to StrOrBytesPath-compatible inputs - normalize scalar and sequence command args before subprocess execution - update git, hg, and svn wrappers to use the shared normalizer - preserve path semantics with os.fspath and flatten hg enum flags
1 parent 6ee680e commit 58603a3

File tree

4 files changed

+51
-28
lines changed

4 files changed

+51
-28
lines changed

src/libvcs/_internal/run.py

Lines changed: 30 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,14 @@
1212

1313
import datetime
1414
import logging
15+
import os
1516
import subprocess
1617
import sys
1718
import typing as t
1819
from collections.abc import Iterable, Mapping, MutableMapping, Sequence
1920

2021
from libvcs import exc
21-
from libvcs._internal.types import StrPath
22+
from libvcs._internal.types import StrOrBytesPath
2223

2324
logger = logging.getLogger(__name__)
2425

@@ -97,23 +98,39 @@ def __call__(self, output: str, timestamp: datetime.datetime) -> None:
9798
if sys.platform == "win32":
9899
_ENV: t.TypeAlias = Mapping[str, str]
99100
else:
100-
_ENV: t.TypeAlias = Mapping[bytes, StrPath] | Mapping[str, StrPath]
101+
_ENV: t.TypeAlias = Mapping[bytes, StrOrBytesPath] | Mapping[str, StrOrBytesPath]
101102

102-
_CMD = StrPath | Sequence[StrPath]
103+
_CMD: t.TypeAlias = StrOrBytesPath | Sequence[StrOrBytesPath]
103104
_FILE: t.TypeAlias = int | t.IO[t.Any] | None
104105

105106

107+
def _normalize_command_args(args: _CMD) -> list[StrOrBytesPath]:
108+
"""Return subprocess arguments without splitting scalar strings or bytes."""
109+
if isinstance(args, (str, bytes, os.PathLike)):
110+
return [os.fspath(args)]
111+
112+
return [os.fspath(arg) for arg in args]
113+
114+
115+
def _stringify_command(args: _CMD) -> str | list[str]:
116+
"""Return a human-readable command for CommandError."""
117+
if isinstance(args, (str, bytes, os.PathLike)):
118+
return os.fsdecode(args)
119+
120+
return [os.fsdecode(arg) for arg in args]
121+
122+
106123
def run(
107124
args: _CMD,
108125
bufsize: int = -1,
109-
executable: StrPath | None = None,
126+
executable: StrOrBytesPath | None = None,
110127
stdin: _FILE | None = None,
111128
stdout: _FILE | None = None,
112129
stderr: _FILE | None = None,
113130
preexec_fn: t.Callable[[], t.Any] | None = None,
114131
close_fds: bool = True,
115132
shell: bool = False,
116-
cwd: StrPath | None = None,
133+
cwd: StrOrBytesPath | None = None,
117134
env: _ENV | None = None,
118135
startupinfo: t.Any | None = None,
119136
creationflags: int = 0,
@@ -173,8 +190,14 @@ def progress_cb(output, timestamp):
173190
----------------
174191
When minimum python >= 3.10, pipesize: int = -1 will be added after umask.
175192
"""
193+
normalized_args: _CMD
194+
if shell:
195+
normalized_args = os.fspath(args) if isinstance(args, os.PathLike) else args
196+
else:
197+
normalized_args = _normalize_command_args(args)
198+
176199
proc = subprocess.Popen(
177-
args,
200+
normalized_args,
178201
bufsize=bufsize,
179202
executable=executable,
180203
stdin=stdin,
@@ -246,8 +269,6 @@ def progress_cb(output: t.AnyStr, timestamp: datetime.datetime) -> None:
246269
raise exc.CommandError(
247270
output=output,
248271
returncode=code,
249-
cmd=[str(arg) for arg in args]
250-
if isinstance(args, (list, tuple))
251-
else str(args),
272+
cmd=_stringify_command(normalized_args),
252273
)
253274
return output

src/libvcs/cmd/git.py

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44

55
import dataclasses
66
import datetime
7+
import os
78
import pathlib
89
import re
910
import shlex
@@ -12,7 +13,7 @@
1213
from collections.abc import Sequence
1314

1415
from libvcs._internal.query_list import QueryList
15-
from libvcs._internal.run import ProgressCallbackProtocol, run
16+
from libvcs._internal.run import ProgressCallbackProtocol, _normalize_command_args, run
1617
from libvcs._internal.types import StrOrBytesPath, StrPath
1718

1819
_CMD = StrOrBytesPath | Sequence[StrOrBytesPath]
@@ -211,10 +212,10 @@ def run(
211212
>>> git.run(['help'])
212213
"usage: git [...--version] [...--help] [-C <path>]..."
213214
"""
214-
cli_args = ["git", *args] if isinstance(args, Sequence) else ["git", args]
215+
cli_args: list[StrOrBytesPath] = ["git", *_normalize_command_args(args)]
215216

216217
if "cwd" not in kwargs:
217-
kwargs["cwd"] = self.path
218+
kwargs["cwd"] = self.path if cwd is None else cwd
218219

219220
#
220221
# Print-and-exit
@@ -237,7 +238,7 @@ def run(
237238
if not isinstance(C, list):
238239
C = [C]
239240
for c in C:
240-
cli_args.extend(["-C", str(c)])
241+
cli_args.extend(["-C", os.fspath(c)])
241242
if config is not None:
242243
assert isinstance(config, dict)
243244

@@ -253,15 +254,15 @@ def stringify(v: t.Any) -> str:
253254
if config_env is not None:
254255
cli_args.append(f"--config-env={config_env}")
255256
if git_dir is not None:
256-
cli_args.extend(["--git-dir", str(git_dir)])
257+
cli_args.extend(["--git-dir", os.fspath(git_dir)])
257258
if work_tree is not None:
258-
cli_args.extend(["--work-tree", str(work_tree)])
259+
cli_args.extend(["--work-tree", os.fspath(work_tree)])
259260
if namespace is not None:
260-
cli_args.extend(["--namespace", namespace])
261+
cli_args.extend(["--namespace", os.fspath(namespace)])
261262
if super_prefix is not None:
262-
cli_args.extend(["--super-prefix", super_prefix])
263+
cli_args.extend(["--super-prefix", os.fspath(super_prefix)])
263264
if exec_path is not None:
264-
cli_args.extend(["--exec-path", exec_path])
265+
cli_args.extend(["--exec-path", os.fspath(exec_path)])
265266
if bare is True:
266267
cli_args.append("--bare")
267268
if no_replace_objects is True:

src/libvcs/cmd/hg.py

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -16,10 +16,10 @@
1616
import typing as t
1717
from collections.abc import Sequence
1818

19-
from libvcs._internal.run import ProgressCallbackProtocol, run
19+
from libvcs._internal.run import ProgressCallbackProtocol, _normalize_command_args, run
2020
from libvcs._internal.types import StrOrBytesPath, StrPath
2121

22-
_CMD = StrOrBytesPath | Sequence[StrOrBytesPath]
22+
_CMD: t.TypeAlias = StrOrBytesPath | Sequence[StrOrBytesPath]
2323

2424

2525
class HgColorType(enum.Enum):
@@ -156,7 +156,7 @@ def run(
156156
>>> hg.run(['help'])
157157
"Mercurial Distributed SCM..."
158158
"""
159-
cli_args = ["hg", *args] if isinstance(args, Sequence) else ["hg", args]
159+
cli_args: list[StrOrBytesPath] = ["hg", *_normalize_command_args(args)]
160160

161161
if "cwd" not in kwargs:
162162
kwargs["cwd"] = self.path
@@ -166,9 +166,9 @@ def run(
166166
if config is not None:
167167
cli_args.extend(["--config", config])
168168
if pager is not None:
169-
cli_args.append(["--pager", pager])
169+
cli_args.extend(["--pager", pager.value])
170170
if color is not None:
171-
cli_args.append(["--color", color])
171+
cli_args.extend(["--color", color.value])
172172
if verbose is True:
173173
cli_args.append("--verbose")
174174
if quiet is True:

src/libvcs/cmd/svn.py

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,16 @@
1010

1111
from __future__ import annotations
1212

13+
import os
1314
import pathlib
1415
import typing as t
1516
from collections.abc import Sequence
1617

1718
from libvcs import exc
18-
from libvcs._internal.run import ProgressCallbackProtocol, run
19+
from libvcs._internal.run import ProgressCallbackProtocol, _normalize_command_args, run
1920
from libvcs._internal.types import StrOrBytesPath, StrPath
2021

21-
_CMD = StrOrBytesPath | Sequence[StrOrBytesPath]
22+
_CMD: t.TypeAlias = StrOrBytesPath | Sequence[StrOrBytesPath]
2223

2324
DepthLiteral = t.Literal["infinity", "empty", "files", "immediates"] | None
2425
RevisionLiteral = t.Literal["HEAD", "BASE", "COMMITTED", "PREV"] | None
@@ -125,7 +126,7 @@ def run(
125126
>>> svn.run(['help'])
126127
"usage: svn <subcommand> [options] [args]..."
127128
"""
128-
cli_args = ["svn", *args] if isinstance(args, Sequence) else ["svn", args]
129+
cli_args: list[StrOrBytesPath] = ["svn", *_normalize_command_args(args)]
129130

130131
if "cwd" not in kwargs:
131132
kwargs["cwd"] = self.path
@@ -141,9 +142,9 @@ def run(
141142
if trust_server_cert is True:
142143
cli_args.append("--trust-server_cert")
143144
if config_dir is not None:
144-
cli_args.extend(["--config-dir", str(config_dir)])
145+
cli_args.extend(["--config-dir", os.fspath(config_dir)])
145146
if config_option is not None:
146-
cli_args.extend(["--config-option", str(config_option)])
147+
cli_args.extend(["--config-option", os.fspath(config_option)])
147148

148149
if self.progress_callback is not None:
149150
kwargs["callback"] = self.progress_callback

0 commit comments

Comments
 (0)