Use enum, typing/docs for popen/run

src/_pytest/capture.py

@@ -4,6 +4,7 @@
 """
 import collections
 import contextlib
+import enum
 import io
 import os
 import sys
@@ -29,6 +30,14 @@
 patchsysdict = {0: "stdin", 1: "stdout", 2: "stderr"}
 
 
+class CloseStdinType(enum.Enum):
+    CLOSE_STDIN = 1
+
+
+CLOSE_STDIN = CloseStdinType.CLOSE_STDIN
+"""Sentinel to close stdin."""
+
+
 def pytest_addoption(parser):
     group = parser.getgroup("general")
     group._addoption(

src/_pytest/pytester.py

@@ -11,8 +11,11 @@
 import traceback
 from fnmatch import fnmatch
 from io import StringIO
+from typing import AnyStr
 from typing import Callable
 from typing import Dict
+from typing import Generic
+from typing import IO
 from typing import Iterable
 from typing import List
 from typing import Optional
@@ -25,8 +28,11 @@
 
 import pytest
 from _pytest._code import Source
+from _pytest.capture import CLOSE_STDIN
+from _pytest.capture import CloseStdinType
 from _pytest.capture import MultiCapture
 from _pytest.capture import SysCapture
+from _pytest.compat import overload
 from _pytest.compat import TYPE_CHECKING
 from _pytest.config import _PluggyPlugin
 from _pytest.config import ExitCode
@@ -513,7 +519,7 @@ def restore(self) -> None:
         sys.path[:], sys.meta_path[:] = self.__saved
 
 
-class Testdir:
+class Testdir(Generic[AnyStr]):
     """Temporary test directory with tools to test/run pytest itself.
 
     This is based on the ``tmpdir`` fixture but provides a number of methods
@@ -533,8 +539,8 @@ class Testdir:
 
     __test__ = False
 
-    class CLOSE_STDIN:
-        pass
+    CLOSE_STDIN = CLOSE_STDIN
+    """Sentinel to close stdin."""
 
     class TimeoutExpired(Exception):
         pass
@@ -1086,14 +1092,42 @@ def collect_by_name(
                 return colitem
         return None
 
+    @overload
     def popen(
         self,
         cmdargs,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        stdin=CLOSE_STDIN,
-        **kw
-    ):
+        stdout: Optional[Union[int, IO]] = subprocess.PIPE,
+        stderr: Optional[Union[int, IO]] = subprocess.PIPE,
+        stdin: Optional[Union[CloseStdinType, bytes, int, IO]] = CLOSE_STDIN,
+        *,
+        encoding: None = ...,
+        **kw,
+    ) -> "subprocess.Popen[bytes]":
+        ...
+
+    @overload
+    def popen(  # noqa: F811
+        self,
+        cmdargs,
+        stdout: Optional[Union[int, IO]] = subprocess.PIPE,
+        stderr: Optional[Union[int, IO]] = subprocess.PIPE,
+        stdin: Optional[Union[CloseStdinType, bytes, int, IO]] = CLOSE_STDIN,
+        *,
+        encoding: str,
+        **kw,
+    ) -> "subprocess.Popen[str]":
+        ...
+
+    def popen(  # noqa: F811
+        self,
+        cmdargs,
+        stdout: Optional[Union[int, IO]] = subprocess.PIPE,
+        stderr: Optional[Union[int, IO]] = subprocess.PIPE,
+        stdin: Optional[Union[CloseStdinType, bytes, int, IO]] = CLOSE_STDIN,
+        *,
+        encoding: Optional[str] = None,
+        **kw,
+    ) -> "Union[subprocess.Popen[bytes], subprocess.Popen[str]]":
         """Invoke subprocess.Popen.
 
         This calls subprocess.Popen making sure the current working directory
@@ -1108,36 +1142,45 @@ def popen(
         )
         kw["env"] = env
 
-        if stdin is Testdir.CLOSE_STDIN:
+        if stdin is CLOSE_STDIN:
             kw["stdin"] = subprocess.PIPE
         elif isinstance(stdin, bytes):
             kw["stdin"] = subprocess.PIPE
         else:
             kw["stdin"] = stdin
 
-        popen = subprocess.Popen(cmdargs, stdout=stdout, stderr=stderr, **kw)
-        if stdin is Testdir.CLOSE_STDIN:
+        popen = subprocess.Popen(
+            cmdargs, stdout=stdout, stderr=stderr, encoding=encoding, **kw
+        )
+        if stdin is CLOSE_STDIN:
+            assert popen.stdin
             popen.stdin.close()
         elif isinstance(stdin, bytes):
+            assert popen.stdin
             popen.stdin.write(stdin)
 
         return popen
 
-    def run(self, *cmdargs, timeout=None, stdin=CLOSE_STDIN) -> RunResult:
+    def run(
+        self,
+        *cmdargs,
+        timeout=None,
+        stdin: Optional[Union[CloseStdinType, bytes, int, IO]] = CLOSE_STDIN,
+    ) -> RunResult:
         """Run a command with arguments.
 
-        Run a process using subprocess.Popen saving the stdout and stderr.
+        Run a process using :class:<python:subprocess.Popen> saving the stdout and stderr.
 
         :param args: the sequence of arguments to pass to `subprocess.Popen()`
         :kwarg timeout: the period in seconds after which to timeout and raise
             :py:class:`Testdir.TimeoutExpired`
         :kwarg stdin: optional standard input.  Bytes are being send, closing
             the pipe, otherwise it is passed through to ``popen``.
-            Defaults to ``CLOSE_STDIN``, which translates to using a pipe
-            (``subprocess.PIPE``) that gets closed.
 
-        Returns a :py:class:`RunResult`.
+            Defaults to :attr:`CLOSE_STDIN`, which translates to using a pipe
+            (:data:`python:subprocess.PIPE`) that gets closed.
 
+        Returns a :py:class:`RunResult`.
         """
         __tracebackhide__ = True
 
@@ -1160,6 +1203,7 @@ def run(self, *cmdargs, timeout=None, stdin=CLOSE_STDIN) -> RunResult:
                 close_fds=(sys.platform != "win32"),
             )
             if isinstance(stdin, bytes):
+                assert popen.stdin
                 popen.stdin.close()
 
             def handle_timeout():
@@ -1408,7 +1452,7 @@ def _match_lines(
         match_func: Callable[[str, str], bool],
         match_nickname: str,
         *,
-        consecutive: bool = False
+        consecutive: bool = False,
     ) -> None:
         """Underlying implementation of ``fnmatch_lines`` and ``re_match_lines``.
 

testing/test_pytester.py

@@ -625,7 +625,7 @@ def test_run_stdin(testdir) -> None:
     assert result.ret == 0
 
 
-def test_popen_stdin_pipe(testdir) -> None:
+def test_popen_stdin_pipe(testdir: Testdir) -> None:
     proc = testdir.popen(
         [sys.executable, "-c", "import sys; print(sys.stdin.read())"],
         stdout=subprocess.PIPE,
@@ -639,7 +639,7 @@ def test_popen_stdin_pipe(testdir) -> None:
     assert proc.returncode == 0
 
 
-def test_popen_stdin_bytes(testdir) -> None:
+def test_popen_stdin_bytes(testdir: Testdir) -> None:
     proc = testdir.popen(
         [sys.executable, "-c", "import sys; print(sys.stdin.read())"],
         stdout=subprocess.PIPE,
@@ -652,7 +652,7 @@ def test_popen_stdin_bytes(testdir) -> None:
     assert proc.returncode == 0
 
 
-def test_popen_default_stdin_stderr_and_stdin_None(testdir) -> None:
+def test_popen_default_stdin_stderr_and_stdin_None(testdir: Testdir) -> None:
     # stdout, stderr default to pipes,
     # stdin can be None to not close the pipe, avoiding
     # "ValueError: flush of closed file" with `communicate()`.
@@ -671,6 +671,24 @@ def test_popen_default_stdin_stderr_and_stdin_None(testdir) -> None:
     assert proc.returncode == 0
 
 
+def test_popen_encoding_and_close_stdin_sentinel(testdir: Testdir) -> None:
+    p1 = testdir.makepyfile(
+        """
+        import sys
+        print('stdout')
+        sys.stderr.write('stderr')
+        """
+    )
+    proc = testdir.popen(
+        [sys.executable, str(p1)], stdin=testdir.CLOSE_STDIN, encoding="utf8"
+    )
+    assert proc.wait() == 0
+    assert proc.stdout and proc.stderr
+    assert proc.stdout.read().splitlines() == ["stdout"]
+    assert proc.stderr.read().splitlines() == ["stderr"]
+    assert proc.returncode == 0
+
+
 def test_spawn_uses_tmphome(testdir) -> None:
     tmphome = str(testdir.tmpdir)
     assert os.environ.get("HOME") == tmphome