diff --git a/CHANGES b/CHANGES index 9e5a8e321..96f515572 100644 --- 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 index d6d1226b3..1b09aec1e 100644 --- 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 new file mode 100644 index 000000000..4c562310f --- /dev/null +++ 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 new file mode 100644 index 000000000..f2998ce83 --- /dev/null +++ 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, + ) diff --git a/tests/utils/test_subprocess.py b/tests/utils/test_subprocess.py new file mode 100644 index 000000000..88bb5fff9 --- /dev/null +++ b/tests/utils/test_subprocess.py @@ -0,0 +1,129 @@ +import pathlib +import subprocess +from typing import Any + +import pytest + +from libvcs.utils.subprocess import SubprocessCommand + + +@pytest.fixture(autouse=True) +def cwd_default(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path): + monkeypatch.chdir(tmp_path) + + +@pytest.mark.parametrize( + "args,kwargs,expected_result", + [ + [["ls"], {}, SubprocessCommand("ls")], + [[["ls", "-l"]], {}, SubprocessCommand(["ls", "-l"])], + [[], {"args": ["ls", "-l"]}, SubprocessCommand(["ls", "-l"])], + [["ls -l"], {"shell": True}, SubprocessCommand("ls -l", shell=True)], + [[], {"args": "ls -l", "shell": True}, SubprocessCommand("ls -l", shell=True)], + [ + [], + {"args": ["ls", "-l"], "shell": True}, + SubprocessCommand(["ls", "-l"], shell=True), + ], + ], +) +def test_init(args: list, kwargs: dict, expected_result: Any): + """Test SubprocessCommand via list + kwargs, assert attributes""" + cmd = SubprocessCommand(*args, **kwargs) + assert cmd == expected_result + + # Attributes in cmd should match whats passed in + for k, v in kwargs.items(): + assert getattr(cmd, k) == v + + proc = cmd.Popen() + proc.communicate() + assert proc.returncode == 0 + + +FIXTURES = [ + [["ls"], {}, SubprocessCommand("ls")], + [[["ls", "-l"]], {}, SubprocessCommand(["ls", "-l"])], +] + + +@pytest.mark.parametrize( + "args,kwargs,expected_result", + FIXTURES, +) +def test_init_and_Popen(args: list, kwargs: dict, expected_result: Any): + """Test SubprocessCommand with Popen""" + cmd = SubprocessCommand(*args, **kwargs) + assert cmd == expected_result + + cmd_proc = cmd.Popen() + cmd_proc.communicate() + assert cmd_proc.returncode == 0 + + proc = subprocess.Popen(*args, **kwargs) + proc.communicate() + assert proc.returncode == 0 + + +@pytest.mark.parametrize( + "args,kwargs,expected_result", + FIXTURES, +) +def test_init_and_Popen_run(args: list, kwargs: dict, expected_result: Any): + """Test SubprocessCommand with run""" + cmd = SubprocessCommand(*args, **kwargs) + assert cmd == expected_result + + cmd_proc = cmd.Popen() + cmd_proc.communicate() + assert cmd_proc.returncode == 0 + + proc = subprocess.run(*args, **kwargs) + assert proc.returncode == 0 + + +@pytest.mark.parametrize( + "args,kwargs,expected_result", + FIXTURES, +) +def test_init_and_check_call(args: list, kwargs: dict, expected_result: Any): + """Test SubprocessCommand with Popen.check_call""" + cmd = SubprocessCommand(*args, **kwargs) + assert cmd == expected_result + + return_code = cmd.check_call() + assert return_code == 0 + + proc = subprocess.check_call(*args, **kwargs) + assert proc == return_code + + +@pytest.mark.parametrize( + "args,kwargs,expected_result", + FIXTURES, +) +def test_init_and_check_output(args: list, kwargs: dict, expected_result: Any): + """Test SubprocessCommand with Popen.check_output""" + cmd = SubprocessCommand(*args, **kwargs) + assert cmd == expected_result + + return_output = cmd.check_output() + assert isinstance(return_output, bytes) + + proc = subprocess.check_output(*args, **kwargs) + assert proc == return_output + + +@pytest.mark.parametrize( + "args,kwargs,run_kwargs", + [ + [["ls"], {}, {}], + [[["ls", "-l"]], {}, {}], + [[["ls", "-al"]], {}, {"stdout": subprocess.DEVNULL}], + ], +) +def test_run(tmp_path: pathlib.Path, args: list, kwargs: dict, run_kwargs: dict): + cmd = SubprocessCommand(*args, cwd=tmp_path, **kwargs) + response = cmd.run(**run_kwargs) + + assert response.returncode == 0