fix: throw a TypeError if unexpected arguments are passed

.config/dictionary.txt

@@ -5,6 +5,7 @@ codespell
 commitlint
 devel
 dists
+fspath
 instafail
 mkdocstrings
 nosetests

.pre-commit-config.yaml

@@ -80,6 +80,7 @@ repos:
       - id: mypy
         # empty args needed in order to match mypy cli behavior
         args: []
+        files: ^(src|test)/
         additional_dependencies:
           - pytest>=6.1.2
           - enrich>=1.2.5

src/subprocess_tee/__init__.py

@@ -25,12 +25,10 @@
 _logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    from subprocess_tee._types import SequenceNotStr
+    from collections.abc import Callable, Sequence
 
-    CompletedProcess = subprocess.CompletedProcess[Any]
-    from collections.abc import Callable
-else:
-    CompletedProcess = subprocess.CompletedProcess
+    from subprocess_tee._types import StrOrBytesPath
+CompletedProcess = subprocess.CompletedProcess
 
 STREAM_LIMIT = 2**23  # 8MB instead of default 64kb, override it if you need
 
@@ -44,55 +42,67 @@ async def _read_stream(stream: StreamReader, callback: Callable[..., Any]) -> No
             break
 
 
+# pylint: disable=too-many-arguments, too-many-locals
 async def _stream_subprocess(  # noqa: C901
-    args: str | tuple[str, ...],
+    args: StrOrBytesPath | Sequence[StrOrBytesPath],
+    *,
+    stdin=None,
+    tee=True,
+    quiet=False,
+    check=False,
+    executable=None,
     **kwargs: Any,
-) -> CompletedProcess:
+) -> subprocess.CompletedProcess[str]:
     platform_settings: dict[str, Any] = {}
     if platform.system() == "Windows":
         platform_settings["env"] = os.environ
 
-    # this part keeps behavior backwards compatible with subprocess.run
-    tee = kwargs.get("tee", True)
-    stdout = kwargs.get("stdout", sys.stdout)
+    # pop arguments so that we can ensure there are no unexpected arguments
+    stdout = kwargs.pop("stdout", sys.stdout)
+    stderr = kwargs.pop("stderr", sys.stderr)
+    for arg in ["cwd", "env"]:
+        if arg in kwargs:
+            platform_settings[arg] = kwargs.pop(arg)
+    if kwargs:
+        msg = f"Popen.__init__() got an unexpected keyword argument '{next(iter(kwargs.keys()))}'"
+        raise TypeError(msg)
+    del kwargs
 
     with Path(os.devnull).open("w", encoding="UTF-8") as devnull:
         if stdout == subprocess.DEVNULL or not tee:
             stdout = devnull
-        stderr = kwargs.get("stderr", sys.stderr)
         if stderr == subprocess.DEVNULL or not tee:
             stderr = devnull
 
         # We need to tell subprocess which shell to use when running shell-like
         # commands.
         # * SHELL is not always defined
         # * /bin/bash does not exit on alpine, /bin/sh seems bit more portable
-        if "executable" not in kwargs and isinstance(args, str) and " " in args:
-            platform_settings["executable"] = os.environ.get("SHELL", "/bin/sh")
-
-        # pass kwargs we know to be supported
-        for arg in ["cwd", "env"]:
-            if arg in kwargs:
-                platform_settings[arg] = kwargs[arg]
+        if executable is None and isinstance(args, str) and " " in args:
+            executable = os.environ.get("SHELL", "/bin/sh")
 
+        if isinstance(args, os.PathLike):
+            args = os.fspath(args)
         # Some users are reporting that default (undocumented) limit 64k is too
         # low
-        if isinstance(args, str):
+        if isinstance(args, (str, bytes)):
             process = await asyncio.create_subprocess_shell(
                 args,
                 limit=STREAM_LIMIT,
-                stdin=kwargs.get("stdin", False),
+                stdin=stdin,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
+                executable=executable,
                 **platform_settings,
             )
         else:
             process = await asyncio.create_subprocess_exec(
                 *args,
                 limit=STREAM_LIMIT,
-                stdin=kwargs.get("stdin", False),
+                stdin=stdin,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
+                executable=executable,
                 **platform_settings,
             )
         out: list[str] = []
@@ -101,7 +111,7 @@ async def _stream_subprocess(  # noqa: C901
         def tee_func(line: bytes, sink: list[str], pipe: Any | None) -> None:
             line_str = line.decode("utf-8").rstrip()
             sink.append(line_str)
-            if not kwargs.get("quiet"):
+            if not quiet:
                 if pipe and hasattr(pipe, "write"):
                     print(line_str, file=pipe)
                 else:
@@ -126,15 +136,14 @@ def tee_func(line: bytes, sink: list[str], pipe: Any | None) -> None:
 
         # We need to be sure we keep the stdout/stderr output identical with
         # the ones produced by subprocess.run(), at least when in text mode.
-        check = kwargs.get("check", False)
         stdout = None if check else ""
         stderr = None if check else ""
         if out:
             stdout = os.linesep.join(out) + os.linesep
         if err:
             stderr = os.linesep.join(err) + os.linesep
 
-        return CompletedProcess(
+        return subprocess.CompletedProcess(
             args=args,
             returncode=await process.wait(),
             stdout=stdout,
@@ -147,17 +156,19 @@ def tee_func(line: bytes, sink: list[str], pipe: Any | None) -> None:
 # pylint: disable=too-many-arguments
 # ruff: ignore=FBT001,ARG001
 def run(
-    args: str | SequenceNotStr[str] | None = None,
+    args: StrOrBytesPath | Sequence[StrOrBytesPath] | None = None,
     bufsize: int = -1,
     input: bytes | str | None = None,  # noqa: A002
     *,
-    capture_output: bool = False,
+    capture_output: bool = True,
     timeout: int | None = None,
     check: bool = False,
     **kwargs: Any,
-) -> CompletedProcess:
+) -> subprocess.CompletedProcess[str]:
     """Drop-in replacement for subprocess.run that behaves like tee.
 
+    Not all arguments to subprocess.run are supported.
+
     Extra arguments added by our version:
     echo: False - Prints command before executing it.
     quiet: False - Avoid printing output
@@ -174,29 +185,39 @@ def run(
         msg = "Popen.__init__() missing 1 required positional argument: 'args'"
         raise TypeError(msg)
 
-    cmd = args if isinstance(args, str) else join(args)
     # bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=None, startupinfo=None, creationflags=0, restore_signals=True, start_new_session=False, pass_fds=(), *, group=None, extra_groups=None, user=None, umask=-1, encoding=None, errors=None, text=None, pipesize=-1, process_group=None
     if bufsize != -1:
-        msg = "Ignored bufsize argument as it is not supported yet by __package__"
+        msg = f"Ignored bufsize argument as it is not supported yet by {__package__}"
+        _logger.warning(msg)
+    if input is not None:
+        msg = f"Ignored input argument as it is not supported yet by {__package__}"
+        _logger.warning(msg)
+    if timeout is not None:
+        msg = f"Ignored timeout argument as it is not supported yet by {__package__}"
+        _logger.warning(msg)
+    if not capture_output:
+        msg = f"Ignored capture_output argument as it is not supported yet by {__package__}"
         _logger.warning(msg)
     kwargs["check"] = check
-    kwargs["input"] = input
-    kwargs["timeout"] = timeout
-    kwargs["capture_output"] = capture_output
 
     check = kwargs.get("check", False)
 
-    if kwargs.get("echo"):
+    if kwargs.pop("echo", False):
+        cmd = (
+            args
+            if isinstance(args, (str, bytes, os.PathLike))
+            else join(str(s) for s in args)
+        )
         print(f"COMMAND: {cmd}")  # noqa: T201
 
-    result = asyncio.run(_stream_subprocess(cmd, **kwargs))
+    result = asyncio.run(_stream_subprocess(args, **kwargs))
     # we restore original args to mimic subprocess.run()
     result.args = args
 
     if check and result.returncode != 0:
         raise subprocess.CalledProcessError(
             result.returncode,
-            cmd,  # pyright: ignore[xxx]
+            args,
             output=result.stdout,
             stderr=result.stderr,
         )

src/subprocess_tee/_types.py

@@ -1,25 +1,6 @@
 """Internally used types."""
 
-# Source from https://github.com/python/typing/issues/256#issuecomment-1442633430
-from collections.abc import Iterator, Sequence
-from typing import Any, Protocol, SupportsIndex, TypeVar, overload
+from os import PathLike
+from typing import Union
 
-_T_co = TypeVar("_T_co", covariant=True)
-
-
-class SequenceNotStr(Protocol[_T_co]):
-    """Lists of strings which are not strings themselves."""
-
-    @overload
-    def __getitem__(self, index: SupportsIndex, /) -> _T_co: ...
-    @overload
-    def __getitem__(self, index: slice, /) -> Sequence[_T_co]: ...
-    def __contains__(self, value: object, /) -> bool: ...
-    def __len__(self) -> int: ...
-    def __iter__(self) -> Iterator[_T_co]: ...
-    def index(  # pylint: disable=C0116
-        self, value: Any, start: int = 0, stop: int = ..., /
-    ) -> int: ...
-    def count(self, value: Any, /) -> int: ...  # pylint: disable=C0116
-
-    def __reversed__(self) -> Iterator[_T_co]: ...
+StrOrBytesPath = Union[str, bytes, PathLike[str], PathLike[bytes]]

test/test_unit.py

@@ -46,6 +46,22 @@ def test_run_list() -> None:
     assert result.stderr == old_result.stderr
 
 
+def test_run_executable() -> None:
+    """Validate run call with a command made of list of strings and an executable."""
+    cmd = ["not a real executable", "-c", "import sys; print(sys.argv[0])"]
+    old_result = subprocess.run(
+        cmd,
+        executable=Path(sys.executable),
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    result = run(cmd, executable=Path(sys.executable))
+    assert result.returncode == old_result.returncode
+    assert result.stdout == old_result.stdout
+    assert result.stderr == old_result.stderr
+
+
 def test_run_echo(capsys: pytest.CaptureFixture[str]) -> None:
     """Validate run call with echo dumps command."""
     cmd = [sys.executable, "--version"]
@@ -174,3 +190,14 @@ def test_run_exc_no_args() -> None:
         subprocess.run(check=False)  # type: ignore[call-overload]
     with pytest.raises(TypeError, match=expected):
         subprocess_tee.run()
+
+
+def test_run_exc_extra_args() -> None:
+    """Checks that call with unrecognized arguments fails the same way as subprocess.run()."""
+    expected = re.compile(
+        r".*__init__\(\) got an unexpected keyword argument 'i_am_not_a_real_argument'"
+    )
+    with pytest.raises(TypeError, match=expected):
+        subprocess.run(["true"], i_am_not_a_real_argument=False, check=False)  # type: ignore[call-overload]
+    with pytest.raises(TypeError, match=expected):
+        subprocess_tee.run(["true"], i_am_not_a_real_argument=False, nor_am_i=True)

tox.ini

@@ -52,7 +52,7 @@ set_env =
     PIP_CONSTRAINT = {tox_root}/.config/constraints.txt
     PIP_DISABLE_PIP_VERSION_CHECK = 1
     PRE_COMMIT_COLOR = always
-    PYTEST_REQPASS = 18
+    PYTEST_REQPASS = 20
     PYTHONDONTWRITEBYTECODE = 1
     PYTHONUNBUFFERED = 1
     UV_CONSTRAINT = {tox_root}/.config/constraints.txt