Switch test output capture to use os.pipe(). Fixes #1819.

tests/debug/output.py

@@ -4,6 +4,7 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+import os
 import re
 import threading
 
@@ -14,45 +15,46 @@ class CapturedOutput(object):
     """Captures stdout and stderr of the debugged process.
     """
 
-    def __init__(self, session):
+    def __init__(self, session, **fds):
         self.session = session
         self._lock = threading.Lock()
         self._chunks = {}
         self._worker_threads = []
 
-        assert not len(session.captured_output - {"stdout", "stderr"})
-        for stream_name in session.captured_output:
+        for stream_name, fd in fds.items():
             log.info("Capturing {0} {1}", session.debuggee_id, stream_name)
-            stream = getattr(session.debuggee, stream_name)
-            self._capture(stream, stream_name)
+            self._capture(fd, stream_name)
 
     def __str__(self):
         return fmt("CapturedOutput({0})", self.session)
 
-    def _worker(self, pipe, name):
+    def _worker(self, fd, name):
         chunks = self._chunks[name]
-        while True:
-            try:
-                chunk = pipe.read(0x1000)
-            except Exception:
-                break
-            if not len(chunk):
-                break
+        try:
+            while True:
+                try:
+                    chunk = os.read(fd, 0x1000)
+                except Exception:
+                    break
+                if not len(chunk):
+                    break
 
-            lines = "\n".join(
-                repr(line) for line, _ in re.findall(b"(.+?(\n|$))", chunk)
-            )
-            log.info("{0} {1}:\n{2}", self.session.debuggee_id, name, lines)
+                lines = "\n".join(
+                    repr(line) for line, _ in re.findall(b"(.+?(\n|$))", chunk)
+                )
+                log.info("{0} {1}:\n{2}", self.session.debuggee_id, name, lines)
 
-            with self._lock:
-                chunks.append(chunk)
+                with self._lock:
+                    chunks.append(chunk)
+        finally:
+            os.close(fd)
 
-    def _capture(self, pipe, name):
+    def _capture(self, fd, name):
         assert name not in self._chunks
         self._chunks[name] = []
 
         thread = threading.Thread(
-            target=lambda: self._worker(pipe, name), name=fmt("{0} {1}", self, name)
+            target=lambda: self._worker(fd, name), name=fmt("{0} {1}", self, name)
         )
         thread.daemon = True
         thread.start()

tests/debug/session.py

@@ -310,6 +310,7 @@ class Session(object):
 
     def spawn_debuggee(self, args, cwd=None, exe=sys.executable, debug_me=None):
         assert self.debuggee is None
+        assert not len(self.captured_output - {"stdout", "stderr"})
 
         args = [exe] + [
             compat.filename_str(s.strpath if isinstance(s, py.path.local) else s)
@@ -333,20 +334,28 @@ class Session(object):
             args,
             env,
         )
+
+        popen_fds = {}
+        capture_fds = {}
+        for stream_name in self.captured_output:
+            rfd, wfd = os.pipe()
+            popen_fds[stream_name] = wfd
+            capture_fds[stream_name] = rfd
         self.debuggee = psutil.Popen(
             args,
             cwd=cwd,
             env=env.for_popen(),
             bufsize=0,
             stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
+            **popen_fds
         )
         log.info("Spawned {0} with PID={1}", self.debuggee_id, self.debuggee.pid)
         watchdog.register_spawn(self.debuggee.pid, self.debuggee_id)
 
-        if self.captured_output:
-            self.captured_output = output.CapturedOutput(self)
+        if len(capture_fds):
+            self.captured_output = output.CapturedOutput(self, **capture_fds)
+        for fd in popen_fds.values():
+            os.close(fd)
 
     def wait_for_enable_attach(self):
         log.info(
@@ -730,6 +739,8 @@ class Session(object):
                 pass
             finally:
                 watchdog.unregister_spawn(self.debuggee.pid, self.debuggee_id)
+            # if self.captured_output:
+            #     self.captured_output.wait()
 
         self.timeline.wait_until_realized(timeline.Event("terminated"))
 

tests/ptvsd/server/test_start_stop.py

@@ -34,9 +34,6 @@ def wait_and_press_key(session):
     session.debuggee.stdin.write(b"\n")
 
 
-@pytest.mark.skipif(
-    sys.version_info < (3, 0), reason="https://github.com/microsoft/ptvsd/issues/1819"
-)
 @pytest.mark.parametrize(
     "run", [runners.launch["integratedTerminal"], runners.launch["externalTerminal"]]
 )