pytester: allow passing in stdin to run/popen

Author: blueyed

changelog/5059.feature.rst

@@ -0,0 +1 @@
+Standard input (stdin) can be given to pytester's ``Testdir.run()`` and ``Testdir.popen()``.

src/_pytest/pytester.py

@@ -36,6 +36,8 @@
     u"/var/lib/sss/mc/passwd"
 ]
 
+CLOSE_STDIN = object
+
 
 def pytest_addoption(parser):
     parser.addoption(
@@ -1032,7 +1034,7 @@ def collect_by_name(self, modcol, name):
             if colitem.name == name:
                 return colitem
 
-    def popen(self, cmdargs, stdout, stderr, **kw):
+    def popen(self, cmdargs, stdout, stderr, stdin=CLOSE_STDIN, **kw):
         """Invoke subprocess.Popen.
 
         This calls subprocess.Popen making sure the current working directory
@@ -1050,10 +1052,18 @@ def popen(self, cmdargs, stdout, stderr, **kw):
         env["USERPROFILE"] = env["HOME"]
         kw["env"] = env
 
-        popen = subprocess.Popen(
-            cmdargs, stdin=subprocess.PIPE, stdout=stdout, stderr=stderr, **kw
-        )
-        popen.stdin.close()
+        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 CLOSE_STDIN:
+            popen.stdin.close()
+        elif isinstance(stdin, bytes):
+            popen.stdin.write(stdin)
 
         return popen
 
@@ -1065,15 +1075,24 @@ def run(self, *cmdargs, **kwargs):
         :param args: the sequence of arguments to pass to `subprocess.Popen()`
         :param timeout: the period in seconds after which to timeout and raise
             :py:class:`Testdir.TimeoutExpired`
+        :param 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`.
 
         """
         __tracebackhide__ = True
 
         timeout = kwargs.pop("timeout", None)
+        stdin = kwargs.pop("stdin", CLOSE_STDIN)
         raise_on_kwargs(kwargs)
 
+        popen_kwargs = {"stdin": stdin}
+        if isinstance(stdin, bytes):
+            popen_kwargs["stdin"] = subprocess.PIPE
+
         cmdargs = [
             str(arg) if isinstance(arg, py.path.local) else arg for arg in cmdargs
         ]
@@ -1086,8 +1105,15 @@ def run(self, *cmdargs, **kwargs):
         try:
             now = time.time()
             popen = self.popen(
-                cmdargs, stdout=f1, stderr=f2, close_fds=(sys.platform != "win32")
+                cmdargs,
+                stdout=f1,
+                stderr=f2,
+                close_fds=(sys.platform != "win32"),
+                **popen_kwargs
             )
+            if isinstance(stdin, bytes):
+                popen.stdin.write(stdin)
+                popen.stdin.close()
 
             def handle_timeout():
                 __tracebackhide__ = True

testing/test_pytester.py

@@ -4,6 +4,7 @@
 from __future__ import print_function
 
 import os
+import subprocess
 import sys
 import time
 
@@ -482,3 +483,60 @@ def test_pytester_addopts(request, monkeypatch):
         testdir.finalize()
 
     assert os.environ["PYTEST_ADDOPTS"] == "--orig-unused"
+
+
+def test_run_stdin(testdir):
+    with pytest.raises(testdir.TimeoutExpired):
+        testdir.run(
+            sys.executable,
+            "-c",
+            "import sys; print(sys.stdin.read())",
+            stdin=subprocess.PIPE,
+            timeout=0.1,
+        )
+
+    with pytest.raises(testdir.TimeoutExpired):
+        result = testdir.run(
+            sys.executable,
+            "-c",
+            "import sys, time; time.sleep(1); print(sys.stdin.read())",
+            stdin=b"input\n2ndline",
+            timeout=0.1,
+        )
+
+    result = testdir.run(
+        sys.executable,
+        "-c",
+        "import sys; print(sys.stdin.read())",
+        stdin=b"input\n2ndline",
+    )
+    assert result.stdout.lines == ["input", "2ndline"]
+    assert result.stderr.str() == ""
+    assert result.ret == 0
+
+
+def test_popen_stdin_pipe(testdir):
+    proc = testdir.popen(
+        [sys.executable, "-c", "import sys; print(sys.stdin.read())"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        stdin=subprocess.PIPE,
+    )
+    stdin = b"input\n2ndline"
+    stdout, stderr = proc.communicate(input=stdin)
+    assert stdout.decode("utf8").splitlines() == ["input", "2ndline"]
+    assert stderr == b""
+    assert proc.returncode == 0
+
+
+def test_popen_stdin_bytes(testdir):
+    proc = testdir.popen(
+        [sys.executable, "-c", "import sys; print(sys.stdin.read())"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        stdin=b"input\n2ndline",
+    )
+    stdout, stderr = proc.communicate()
+    assert stdout.decode("utf8").splitlines() == ["input", "2ndline"]
+    assert stderr == b""
+    assert proc.returncode == 0