[RFC] pytester. support stdin with runpytest_inprocess

src/_pytest/capture.py

@@ -633,16 +633,20 @@ def snap(self):
 
 class SysCapture:
 
+    CLOSE_STDIN = object
     EMPTY_BUFFER = str()
     _state = None
 
-    def __init__(self, fd, tmpfile=None):
+    def __init__(self, fd, tmpfile=None, stdin=CLOSE_STDIN):
         name = patchsysdict[fd]
         self._old = getattr(sys, name)
         self.name = name
         if tmpfile is None:
             if name == "stdin":
-                tmpfile = DontReadFromInput()
+                if stdin is self.CLOSE_STDIN:
+                    tmpfile = DontReadFromInput()
+                else:
+                    tmpfile = stdin
             else:
                 tmpfile = CaptureIO()
         self.tmpfile = tmpfile

src/_pytest/pytester.py

@@ -847,7 +847,7 @@ class reprec:
             for finalizer in finalizers:
                 finalizer()
 
-    def runpytest_inprocess(self, *args, **kwargs):
+    def runpytest_inprocess(self, *args, stdin=CLOSE_STDIN, **kwargs):
         """Return result of running pytest in-process, providing a similar
         interface to what self.runpytest() provides.
         """
@@ -856,7 +856,26 @@ def runpytest_inprocess(self, *args, **kwargs):
         if syspathinsert:
             self.syspathinsert()
         now = time.time()
-        capture = MultiCapture(Capture=SysCapture)
+        if stdin is self.CLOSE_STDIN:
+            Capture = SysCapture
+        else:
+            old_stdin = sys.stdin
+            if isinstance(stdin, str):
+                import io
+                import functools
+
+                class EchoingInput(io.StringIO):
+                    def readline(self, *args, **kwargs):
+                        ret = super().readline(*args, **kwargs)
+                        if ret is not None:
+                            sys.stdout.write(ret)
+                        return ret
+
+                Capture = functools.partial(SysCapture, stdin=EchoingInput(stdin))
+            else:
+                Capture = stdin
+
+        capture = MultiCapture(Capture=Capture)
         capture.start_capturing()
         try:
             try:
@@ -878,6 +897,9 @@ class reprec:
             sys.stdout.write(out)
             sys.stderr.write(err)
 
+            if stdin is not self.CLOSE_STDIN:
+                sys.stdin = old_stdin
+
         res = RunResult(reprec.ret, out.split("\n"), err.split("\n"), time.time() - now)
         res.reprec = reprec
         return res

testing/test_pdb.py

@@ -724,17 +724,21 @@ def test_pdb_used_in_generate_tests(self, testdir):
             """
             import pytest
             def pytest_generate_tests(metafunc):
-                pytest.set_trace()
                 x = 5
+                pytest.set_trace()
+
             def test_foo(a):
                 pass
         """
         )
-        child = testdir.spawn_pytest(str(p1))
-        child.expect("x = 5")
-        child.expect("Pdb")
-        child.sendeof()
-        self.flush(child)
+        result = testdir.runpytest(str(p1), stdin="p 'x=' + str(x)\nq\n")
+        result.stdout.fnmatch_lines(
+            [
+                "*> PDB set_trace (IO-capturing turned off) >*",
+                "'x=5'",
+                "E   _pytest.outcomes.Exit: Quitting debugger",
+            ]
+        )
 
     def test_pdb_collection_failure_is_shown(self, testdir):
         p1 = testdir.makepyfile("xxx")