diff --git a/src/_pytest/capture.py b/src/_pytest/capture.py index a1335d654f0..dcbceeef55e 100644 --- a/src/_pytest/capture.py +++ b/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( diff --git a/src/_pytest/pytester.py b/src/_pytest/pytester.py index 9df3ed779d5..df834a4eafc 100644 --- a/src/_pytest/pytester.py +++ b/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,7 +539,8 @@ class Testdir: __test__ = False - CLOSE_STDIN = object + CLOSE_STDIN = CLOSE_STDIN + """Sentinel to close stdin.""" class TimeoutExpired(Exception): pass @@ -1089,14 +1096,42 @@ 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 @@ -1111,36 +1146,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: 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 @@ -1163,6 +1207,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(): @@ -1411,7 +1456,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``. diff --git a/testing/test_pytester.py b/testing/test_pytester.py index 959000061a5..119fc7c5d55 100644 --- a/testing/test_pytester.py +++ b/testing/test_pytester.py @@ -648,7 +648,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, @@ -662,7 +662,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, @@ -675,7 +675,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()`. @@ -694,6 +694,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