Throw a TypeError if unexpected arguments are passed

Author: eric-wieser

src/subprocess_tee/__init__.py

@@ -25,12 +25,10 @@
 _logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    from subprocess_tee._types import SequenceNotStr
-
-    CompletedProcess = subprocess.CompletedProcess[Any]
     from collections.abc import Callable
-else:
-    CompletedProcess = subprocess.CompletedProcess
+
+    from subprocess_tee._types import SequenceNotStr
+CompletedProcess = subprocess.CompletedProcess
 
 STREAM_LIMIT = 2**23  # 8MB instead of default 64kb, override it if you need
 
@@ -46,53 +44,66 @@ async def _read_stream(stream: StreamReader, callback: Callable[..., Any]) -> No
 
 async def _stream_subprocess(  # noqa: C901
     args: str | tuple[str, ...],
+    *,
+    stdin=None,
+    tee=True,
+    quiet=False,
+    check=False,
+    executable=None,
+    capture_output=True,
     **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 '{list(kwargs.keys())[0]}'"
+        raise TypeError(msg)
+    if not capture_output:
+        msg = "Only capture_output=True is supported"
+        raise NotImplementedError(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")
 
         # Some users are reporting that default (undocumented) limit 64k is too
         # low
         if isinstance(args, str):
             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 +112,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 +137,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,
@@ -149,15 +159,15 @@ def tee_func(line: bytes, sink: list[str], pipe: Any | None) -> None:
 def run(
     args: str | SequenceNotStr[str] | None = None,
     bufsize: int = -1,
-    input: bytes | str | None = None,  # noqa: A002
     *,
-    capture_output: bool = False,
-    timeout: int | None = None,
+    capture_output: bool = True,
     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
@@ -177,16 +187,14 @@ def run(
     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)
     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):
         print(f"COMMAND: {cmd}")  # noqa: T201
 
     result = asyncio.run(_stream_subprocess(cmd, **kwargs))

test/test_unit.py

@@ -46,6 +46,23 @@ 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=sys.executable,
+        # shell=True,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    result = run(cmd)
+    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 +191,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 without 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)