Add types to run command (#149)

Author: ssbarnea

.config/constraints.txt

@@ -41,7 +41,7 @@ mkdocs==1.6.1             # via mkdocs-autorefs, mkdocs-git-revision-date-locali
 mkdocs-autorefs==1.2.0    # via mkdocstrings, mkdocstrings-python
 mkdocs-get-deps==0.2.0    # via mkdocs
 mkdocs-git-revision-date-localized-plugin==1.3.0  # via subprocess-tee (pyproject.toml)
-mkdocs-material==9.5.47   # via subprocess-tee (pyproject.toml)
+mkdocs-material==9.5.48   # via subprocess-tee (pyproject.toml)
 mkdocs-material-extensions==1.3.1  # via mkdocs-material, subprocess-tee (pyproject.toml)
 mkdocstrings==0.27.0      # via mkdocstrings-python, subprocess-tee (pyproject.toml)
 mkdocstrings-python==1.12.2  # via subprocess-tee (pyproject.toml)

.pre-commit-config.yaml

@@ -18,7 +18,7 @@ repos:
     hooks:
       - id: prettier
   - repo: https://github.com/streetsidesoftware/cspell-cli
-    rev: v8.16.0
+    rev: v8.16.1
     hooks:
       - id: cspell
         # entry: codespell --relative
@@ -93,7 +93,7 @@ repos:
           - typing
           - typing-extensions
   - repo: https://github.com/jendrikseipp/vulture
-    rev: v2.13
+    rev: v2.14
     hooks:
       - id: vulture
   - # keep at bottom as these are slower

pyproject.toml

@@ -97,7 +97,9 @@ disable = [
   # Disabled on purpose:
   "line-too-long", # covered by black
   "protected-access", # covered by ruff SLF001
+  "redefined-builtin", # covered by ruff
   "too-many-branches", # covered by ruff C901
+  "unused-argument", # covered vby ruff
   "wrong-import-order", # covered by ruff
   # TODO(ssbarnea): remove temporary skips adding during initial adoption:
   "duplicate-code",
@@ -195,6 +197,7 @@ exclude = [
   ".tox",
   "build",
   "venv",
-  "src/subprocess_tee/_version.py"
+  "src/subprocess_tee/_version.py",
+  "src/subprocess_tee/_types.py"
 ]
 paths = ["src", "test"]

src/subprocess_tee/__init__.py

@@ -1,8 +1,11 @@
 """tee like run implementation."""
 
+# cspell: ignore popenargs preexec startupinfo creationflags pipesize
+
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import platform
 import subprocess  # noqa: S404
@@ -19,9 +22,12 @@
     __version__ = "0.1.dev1"
 
 __all__ = ["CompletedProcess", "__version__", "run"]
+_logger = logging.getLogger(__name__)
 
 if TYPE_CHECKING:
-    CompletedProcess = subprocess.CompletedProcess[Any]  # pylint: disable=E1136
+    from subprocess_tee._types import SequenceNotStr
+
+    CompletedProcess = subprocess.CompletedProcess[Any]
     from collections.abc import Callable
 else:
     CompletedProcess = subprocess.CompletedProcess
@@ -39,7 +45,7 @@ async def _read_stream(stream: StreamReader, callback: Callable[..., Any]) -> No
 
 
 async def _stream_subprocess(  # noqa: C901
-    args: str | list[str],
+    args: str | tuple[str, ...],
     **kwargs: Any,
 ) -> CompletedProcess:
     platform_settings: dict[str, Any] = {}
@@ -136,7 +142,20 @@ def tee_func(line: bytes, sink: list[str], pipe: Any | None) -> None:
         )
 
 
-def run(args: str | list[str], **kwargs: Any) -> CompletedProcess:
+# signature is based on stdlib
+# subprocess.run()
+# pylint: disable=too-many-arguments
+# ruff: ignore=FBT001,ARG001
+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,
+    check: bool = False,
+    **kwargs: Any,
+) -> CompletedProcess:
     """Drop-in replacement for subprocess.run that behaves like tee.
 
     Extra arguments added by our version:
@@ -148,26 +167,36 @@ def run(args: str | list[str], **kwargs: Any) -> CompletedProcess:
 
     Raises:
         CalledProcessError: ...
+        TypeError: ...
 
     """
-    # run was called with a list instead of a single item but asyncio
-    # create_subprocess_shell requires command as a single string, so
-    # we need to convert it to string
+    if args is None:
+        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__"
+        _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", False):
         print(f"COMMAND: {cmd}")  # noqa: T201
 
-    result = asyncio.run(_stream_subprocess(args, **kwargs))
+    result = asyncio.run(_stream_subprocess(cmd, **kwargs))
     # we restore original args to mimic subprocess.run()
     result.args = args
 
     if check and result.returncode != 0:
         raise subprocess.CalledProcessError(
             result.returncode,
-            args,
+            cmd,  # pyright: ignore[xxx]
             output=result.stdout,
             stderr=result.stderr,
         )

src/subprocess_tee/_types.py

@@ -0,0 +1,25 @@
+"""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
+
+_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]: ...

test/test_unit.py

@@ -1,11 +1,13 @@
 """Unit tests."""
 
+import re
 import subprocess
 import sys
 from pathlib import Path
 
 import pytest
 
+import subprocess_tee
 from subprocess_tee import run
 
 
@@ -140,8 +142,35 @@ def test_run_compat() -> None:
     assert ours.args == original.args
 
 
+def test_run_compat2() -> None:
+    """Assure compatibility with subprocess.run()."""
+    cmd: tuple[str, int] = ("true", -1)
+    ours = run(*cmd)
+    original = subprocess.run(
+        *cmd,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert ours.returncode == original.returncode
+    assert ours.stdout == original.stdout
+    assert ours.stderr == original.stderr
+    assert ours.args == original.args
+
+
 def test_run_waits_for_completion(tmp_path: Path) -> None:
     """run() should always wait for the process to complete."""
     tmpfile = tmp_path / "output.txt"
     run(f"sleep 0.1 && echo 42 > {tmpfile!s}")
     assert tmpfile.read_text() == "42\n"
+
+
+def test_run_exc_no_args() -> None:
+    """Checks that call without arguments fails the same way as subprocess.run()."""
+    expected = re.compile(
+        r".*__init__\(\) missing 1 required positional argument: 'args'"
+    )
+    with pytest.raises(TypeError, match=expected):
+        subprocess.run(check=False)  # type: ignore[call-overload]
+    with pytest.raises(TypeError, match=expected):
+        subprocess_tee.run()

tox.ini

@@ -47,8 +47,7 @@ setenv =
     PIP_CONSTRAINT = {tox_root}/.config/constraints.txt
     devel,pkg,pre: PIP_CONSTRAINT = /dev/null
     PIP_DISABLE_VERSION_CHECK=1
-    PYTEST_REQPASS=16
-    py38: PYTEST_REQPASS=15
+    PYTEST_REQPASS=18
     PYTHONDONTWRITEBYTECODE=1
     PYTHONUNBUFFERED=1
 commands =