vcs-python · tony · May 1, 2022 · May 1, 2022 · May 1, 2022 · May 1, 2022
diff --git a/CHANGES b/CHANGES
@@ -13,6 +13,12 @@ $ pip install --user --upgrade --pre libvcs
 
 - _Add your latest changes from PRs here_
 
+### Development
+
+- {issue}`#336`: {class}`~libvcs.utils.subprocess.SubprocessCommand`: Encapsulated {mod}`subprocess`
+  call in a {func}`dataclasses.dataclass` for introspecting, modifying, mocking and controlling
+  execution.
+
 ## libvcs 0.12.0, "Nimbus" (2022-04-24)
 
 ### Breaking

diff --git a/docs/contributing/internals/index.md b/docs/contributing/internals/index.md
@@ -13,4 +13,5 @@ stability policy.
 exc
 types
 query_list
+subprocess
 ```
diff --git a/docs/contributing/internals/subprocess.md b/docs/contributing/internals/subprocess.md
@@ -0,0 +1,13 @@
+# `libvcs.utils.subprocess`
+
+```{eval-rst}
+.. autoapimodule:: libvcs.utils.subprocess
+   :members:
+   :exclude-members:
+     StrOrBytesPath, F, args, bufsize,
+     executable, stdin, stdout, stderr, preexec_fn, cwd,
+     close_fds, shell, cmd, env, text, universal_newlines,
+     startupinfo, creationflags, restore_signals, start_new_session,
+     group, extra_groups, user, umask, pass_fds, encoding, errors,
+
+```
diff --git a/libvcs/utils/subprocess.py b/libvcs/utils/subprocess.py
@@ -0,0 +1,278 @@
+import dataclasses
+import subprocess
+import sys
+from typing import (
+    IO,
+    Any,
+    Callable,
+    Literal,
+    Mapping,
+    Optional,
+    Sequence,
+    TypeVar,
+    Union,
+    overload,
+)
+
+from typing_extensions import TypeAlias
+
+from ..types import StrOrBytesPath
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+
+if sys.platform == "win32":
+    _ENV: TypeAlias = Mapping[str, str]
+else:
+    _ENV: TypeAlias = Union[
+        Mapping[bytes, StrOrBytesPath], Mapping[str, StrOrBytesPath]
+    ]
+_FILE: TypeAlias = Union[None, int, IO[Any]]
+_TXT: TypeAlias = Union[bytes, str]
+#: Command
+_CMD: TypeAlias = Union[StrOrBytesPath, Sequence[StrOrBytesPath]]
+
+
+@dataclasses.dataclass
+class SubprocessCommand:
+    """Encapsulate a :mod:`subprocess` request. Inspect, mutate, control before invocation.
+
+    Attributes
+    ----------
+    args : _CMD
+        A string, or a sequence of program arguments.
+
+    bufsize : int
+        supplied as the buffering argument to the open() function when creating the
+        stdin/stdout/stderr pipe file objects
+
+    executable : Optional[StrOrBytesPath]
+        A replacement program to execute.
+
+    stdin : _FILE
+        standard output for executed program
+
+    stdout :
+        standard output for executed program
+
+    stderr :
+        standard output for executed program
+
+    close_fds : Controls closing or inheriting of file descriptors.
+
+    shell : If true, the command will be executed through the shell.
+
+    cwd : Sets the current directory before the child is executed.
+
+    env : Defines the environment variables for the new process.
+
+    text :
+        If ``True``, decode stdin, stdout and stderr using the given encoding (if set)
+        or the system default otherwise.
+
+    universal_newlines :
+        Alias of text, provided for backwards compatibility.
+
+    startupinfo :
+        Windows only
+
+    creationflags :
+        Windows only
+
+    preexec_fn :
+        (POSIX only) An object to be called in the child process just before the child
+        is executed.
+
+    restore_signals :
+        POSIX only
+
+    start_new_session :
+        POSIX only
+
+    group :
+        POSIX only
+
+    extra_groups :
+        POSIX only
+
+    user :
+        POSIX only
+
+    umask :
+        POSIX only
+
+    pass_fds :
+        POSIX only
+
+    encoding :
+        Text mode encoding to use for file objects stdin, stdout and stderr.
+
+    errors :
+        Text mode error handling to use for file objects stdin, stdout and stderr.
+
+    Examples
+    --------
+    >>> cmd = SubprocessCommand("ls")
+    >>> cmd.args
+    'ls'
+
+    With ``shell=True``:
+
+    >>> cmd = SubprocessCommand("ls -l", shell=True)
+    >>> cmd.shell
+    True
+    >>> cmd.args
+    'ls -l'
+    >>> cmd.check_call()
+    0
+    """
+
+    args: _CMD
+    bufsize: int = -1
+    executable: Optional[StrOrBytesPath] = None
+    stdin: _FILE = None
+    stdout: _FILE = None
+    stderr: _FILE = None
+    preexec_fn: Optional[Callable[[], Any]] = None
+    close_fds: bool = True
+    shell: bool = False
+    cwd: Optional[StrOrBytesPath] = None
+    env: Optional[_ENV] = None
+
+    # Windows
+    creationflags: int = 0
+    startupinfo: Optional[Any] = None
+
+    # POSIX-only
+    restore_signals: bool = True
+    start_new_session: bool = False
+    pass_fds: Any = ()
+    umask: int = -1
+    if sys.version_info >= (3, 10):
+        pipesize: int = -1
+    user: Optional[str] = None
+    group: Optional[str] = None
+    extra_groups: Optional[list[str]] = None
+
+    # Alias of text, for backwards compatibility
+    universal_newlines: Optional[bool] = None
+    text: Optional[Literal[True]] = None
+
+    # Text mode encoding and error handling to use for file objects
+    # stdin, stdout, stderr
+    encoding: Optional[str] = None
+    errors: Optional[str] = None
+
+    def Popen(self, **kwargs) -> subprocess.Popen:
+        """Run commands :class:`subprocess.Popen`, optionally overrides via kwargs.
+
+        Parameters
+        ----------
+        **kwargs : dict, optional
+            Overrides existing attributes for :class:`subprocess.Popen`
+
+        Examples
+        --------
+        >>> cmd = SubprocessCommand(args=['echo', 'hello'])
+        >>> proc = cmd.Popen(stdout=subprocess.PIPE)
+        >>> proc.communicate() # doctest: +SKIP
+        """
+        return subprocess.Popen(**dataclasses.replace(self, **kwargs).__dict__)
+
+    def check_call(self, **kwargs) -> int:
+        """Run command :func:`subprocess.check_call`, optionally overrides via kwargs.
+
+        Parameters
+        ----------
+        **kwargs : dict, optional
+            Overrides existing attributes for :func:`subprocess.check_call`
+
+        Examples
+        --------
+        >>> cmd = SubprocessCommand(args=['echo', 'hello'])
+        >>> cmd.check_call(stdout=subprocess.PIPE)
+        0
+        """
+        return subprocess.check_call(**dataclasses.replace(self, **kwargs).__dict__)
+
+    @overload
+    def check_output(self, input: Optional[str] = None, **kwargs) -> str:
+        ...
+
+    @overload
+    def check_output(self, input: Optional[bytes] = None, **kwargs) -> bytes:
+        ...
+
+    def check_output(
+        self, input: Optional[Union[str, bytes]] = None, **kwargs
+    ) -> Union[bytes, str]:
+        r"""Run command :func:`subprocess.check_output`, optionally overrides via kwargs.
+
+        Parameters
+        ----------
+        input : Union[bytes, str], optional
+            pass string to subprocess's stdin. Bytes by default, str in text mode.
+
+            Text mode is triggered by setting any of text, encoding, errors or
+            universal_newlines.
+        **kwargs : dict, optional
+            Overrides existing attributes for :func:`subprocess.check_output`
+
+        Examples
+        --------
+        >>> cmd = SubprocessCommand(args=['echo', 'hello'])
+        >>> proc = cmd.check_output(shell=True)
+
+        From :mod:`subprocess`:
+
+        >>> import subprocess
+        >>> cmd = SubprocessCommand(
+        ...     ["/bin/sh", "-c", "ls -l non_existent_file ; exit 0"])
+        >>> cmd.check_output(stderr=subprocess.STDOUT)
+        b"ls: ...non_existent_file...: No such file or directory\n"
+
+        >>> cmd = SubprocessCommand(["sed", "-e", "s/foo/bar/"])
+        >>> cmd.check_output(input=b"when in the course of fooman events\n")
+        b'when in the course of barman events\n'
+        """
+        params = dataclasses.replace(self, **kwargs).__dict__
+        params.pop("stdout")
+        return subprocess.check_output(input=input, **params)
+
+    def run(
+        self,
+        input: Optional[Union[str, bytes]] = None,
+        timeout: Optional[int] = None,
+        check: bool = False,
+        capture_output: bool = False,
+        **kwargs,
+    ) -> subprocess.CompletedProcess:
+        """Run command in :func:`subprocess.run`, optionally overrides via kwargs.
+
+        Parameters
+        ----------
+        input : Union[bytes, str], optional
+            pass string to subprocess's stdin. Bytes by default, str in text mode.
+
+            Text mode is triggered by setting any of text, encoding, errors or
+            universal_newlines.
+
+        check : bool
+            If True and the exit code was non-zero, it raises a
+            :exc:`subprocess.CalledProcessError`. The CalledProcessError object will
+            have the return code in the returncode attribute, and output & stderr
+            attributes if those streams were captured.
+
+        timeout : int
+            If given, and the process takes too long, a :exc:`subprocess.TimeoutExpired`
+
+        **kwargs : dict, optional
+            Overrides existing attributes for :func:`subprocess.run`
+        """
+        return subprocess.run(
+            **dataclasses.replace(self, **kwargs).__dict__,
+            input=input,
+            capture_output=capture_output,
+            timeout=timeout,
+            check=check,
+        )
-Original file line number
+Diff line change
@@ Expand Up / @@ -13,4 +13,5 @@ stability policy. @@
     exc
     types
     query_list
+    subprocess
     ```