New CLOSE_STDIN sentinel (#272)

.coveragerc

@@ -27,3 +27,4 @@ exclude_lines =
     ^\s*assert False(,|$)
 
     ^\s*if TYPE_CHECKING:
+    ^\s*\.\.\.

.travis.yml

@@ -20,7 +20,7 @@ jobs:
     # Coverage for Python 3.5.{0,1} specific code, mostly typing related.
     - env:
         - TOXENV=py35-coverage
-        - PYTEST_ADDOPTS="-k 'test_raises_cyclic_reference or test_supports_breakpoint_module_global'"
+        - PYTEST_ADDOPTS="-m py35_specific"
       before_install:
         - python -m pip install -U pip==19.3.1
         # Work around https://github.com/jaraco/zipp/issues/40.

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(
@@ -629,8 +638,7 @@ def snap(self):
 
 
 class SysCaptureBinary:
-    class CLOSE_STDIN:
-        pass
+    CLOSE_STDIN = CLOSE_STDIN
 
     EMPTY_BUFFER = b""
     _state = None

src/_pytest/pytester.py

@@ -12,9 +12,12 @@
 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 Generator
+from typing import Generic
+from typing import IO
 from typing import Iterable
 from typing import List
 from typing import Mapping
@@ -28,8 +31,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
@@ -570,7 +576,7 @@ def pytest_runtest_call(item: Function) -> Generator[None, None, None]:
         mp.undo()
 
 
-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
@@ -590,8 +596,8 @@ class Testdir:
 
     __test__ = False
 
-    class CLOSE_STDIN:
-        pass
+    CLOSE_STDIN = CLOSE_STDIN
+    """Sentinel to close stdin."""
 
     class TimeoutExpired(Exception):
         pass
@@ -983,11 +989,9 @@ def runpytest_inprocess(self, *args, tty=None, **kwargs) -> RunResult:
             if "-s" in args:
                 stdin = sys.stdin
             else:
-                stdin = self.CLOSE_STDIN
+                stdin = CLOSE_STDIN
 
-        if stdin is self.CLOSE_STDIN:
-            stdin = SysCapture.CLOSE_STDIN
-        elif isinstance(stdin, str):
+        if isinstance(stdin, str):
 
             class EchoingInput(StringIO):
                 def readline(self, *args, **kwargs):
@@ -1004,7 +1008,7 @@ def get_capture(fd):
             from _pytest.compat import CaptureIO
 
             if fd == 0:
-                if not stdin or stdin is SysCapture.CLOSE_STDIN:
+                if not stdin or stdin is CLOSE_STDIN:
                     tmpfile = None
                 else:
                     from _pytest.capture import safe_text_dupfile
@@ -1182,58 +1186,96 @@ def collect_by_name(
                 return colitem
         return None
 
+    @overload
     def popen(
         self,
         cmdargs,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        stdin=CLOSE_STDIN,
+        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
         is in the PYTHONPATH.
 
-        You probably want to use :py:meth:`run` instead.
+        `encoding` is only supported with Python 3.6+.
 
+        You probably want to use :py:meth:`run` instead.
         """
         env = os.environ.copy()
         env["PYTHONPATH"] = os.pathsep.join(
             filter(None, [os.getcwd(), env.get("PYTHONPATH", "")])
         )
         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
 
+        if encoding is not None:
+            kw["encoding"] = encoding
         popen = subprocess.Popen(cmdargs, stdout=stdout, stderr=stderr, **kw)
-        if stdin is Testdir.CLOSE_STDIN:
+        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
 
@@ -1255,6 +1297,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():

testing/python/raises.py

@@ -155,6 +155,7 @@ def test_no_raise_message(self):
         else:
             assert False, "Expected pytest.raises.Exception"
 
+    @pytest.mark.py35_specific
     @pytest.mark.parametrize("method", ["function", "function_match", "with"])
     def test_raises_cyclic_reference(self, method):
         """

testing/test_debugging.py

@@ -961,6 +961,7 @@ def test_foo():
 
 
 class TestDebuggingBreakpoints:
+    @pytest.mark.py35_specific
     def test_supports_breakpoint_module_global(self):
         """
         Test that supports breakpoint global marks on Python 3.7+ and not on

testing/test_pytester.py

@@ -901,7 +901,7 @@ def test():
     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,
@@ -915,7 +915,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,
@@ -928,7 +928,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()`.
@@ -947,6 +947,30 @@ def test_popen_default_stdin_stderr_and_stdin_None(testdir) -> None:
     assert proc.returncode == 0
 
 
+@pytest.mark.py35_specific
+def test_popen_encoding_and_close_stdin_sentinel(testdir: Testdir) -> None:
+    if sys.version_info < (3, 6):
+        with pytest.raises(TypeError):
+            testdir.popen([sys.executable], stdin=testdir.CLOSE_STDIN, encoding="utf8")
+        return
+
+    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

tox.ini

@@ -203,6 +203,7 @@ markers =
     # experimental mark for all tests using pexpect
     uses_pexpect
     pypy_specific
+    py35_specific
 
 [flake8]
 max-line-length = 120