Merge pull request #283 from ericsnowcurrently/fix-tests-race

Fix races in test code.

ptvsd/wrapper.py

@@ -21,6 +21,7 @@ try:
     urllib.unquote
 except Exception:
     import urllib.parse as urllib
+import warnings
 
 import _pydevd_bundle.pydevd_constants as pydevd_constants
 # Disable this, since we aren't packaging the Cython modules at the moment.
@@ -656,7 +657,39 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
             output='ptvsd',
             data={'version': __version__})
 
-    def _handle_exit(self):
+    # closing the adapter
+
+    def close(self):
+        """Stop the message processor and release its resources."""
+        if self._closed:
+            return
+        self._closed = True
+
+        # Stop the PyDevd message handler first.
+        self._stop_pydevd_message_loop()
+        # Treat PyDevd as effectively exited.
+        self._handle_pydevd_stopped()
+        # Close the editor-side socket.
+        self._stop_vsc_message_loop()
+
+    def _stop_pydevd_message_loop(self):
+        pydevd = self.pydevd
+        self.pydevd = None
+        pydevd.shutdown(socket.SHUT_RDWR)
+        pydevd.close()
+
+    def _stop_vsc_message_loop(self):
+        self.set_exit()
+        self.loop.stop()
+        self.event_loop_thread.join(WAIT_FOR_THREAD_FINISH_TIMEOUT)
+        if self.socket:
+            try:
+                self.socket.shutdown(socket.SHUT_RDWR)
+                self.socket.close()
+            except Exception:
+                pass
+
+    def _handle_pydevd_stopped(self):
         wait_on_normal_exit = self.debug_options.get(
             'WAIT_ON_NORMAL_EXIT', False)
         wait_on_abnormal_exit = self.debug_options.get(
@@ -672,37 +705,67 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
             return
         self._exited = True
 
+        # Notify the editor that the "debuggee" (e.g. script, app) exited.
         self.send_event('exited', exitCode=ptvsd_sys_exit_code)
+        # Notify the editor that the debugger has stopped.
         self.send_event('terminated')
 
-        self.disconnect_request_event.wait(WAIT_FOR_DISCONNECT_REQUEST_TIMEOUT)
+        # The editor will send a "disconnect" request at this point.
+        self._wait_for_disconnect()
+
+    def _wait_for_disconnect(self, timeout=None):
+        if timeout is None:
+            timeout = WAIT_FOR_DISCONNECT_REQUEST_TIMEOUT
+
+        if not self.disconnect_request_event.wait(timeout):
+            warnings.warn('timed out waiting for disconnect request')
         if self.disconnect_request is not None:
             self.send_response(self.disconnect_request)
             self.disconnect_request = None
 
-    def close(self):
-        """Stop the message processor and release its resources."""
-        if self._closed:
-            return
-        self._closed = True
+    def _handle_disconnect(self, request):
+        self.disconnect_request = request
+        self.disconnect_request_event.set()
+        killProcess = not self._closed
+        self.close()
+        # TODO: Move killing the process to close()?
+        if killProcess and self.killonclose:
+            os.kill(os.getpid(), signal.SIGTERM)
 
-        pydevd = self.pydevd
-        self.pydevd = None
-        pydevd.shutdown(socket.SHUT_RDWR)
-        pydevd.close()
+    # async helpers
 
-        self._handle_exit()
+    def async_method(m):
+        """Converts a generator method into an async one."""
+        m = futures.wrap_async(m)
 
-        self.set_exit()
-        self.loop.stop()
-        self.event_loop_thread.join(WAIT_FOR_THREAD_FINISH_TIMEOUT)
+        def f(self, *args, **kwargs):
+            return m(self, self.loop, *args, **kwargs)
 
-        if self.socket:
-            try:
-                self.socket.shutdown(socket.SHUT_RDWR)
-                self.socket.close()
-            except Exception:
-                pass
+        return f
+
+    def async_handler(m):
+        """Converts a generator method into a fire-and-forget async one."""
+        m = futures.wrap_async(m)
+
+        def f(self, *args, **kwargs):
+            fut = m(self, self.loop, *args, **kwargs)
+
+            def done(fut):
+                try:
+                    fut.result()
+                except BaseException:
+                    traceback.print_exc(file=sys.__stderr__)
+
+            fut.add_done_callback(done)
+
+        return f
+
+    def sleep(self):
+        fut = futures.Future(self.loop)
+        self.loop.call_soon(lambda: fut.set_result(None))
+        return fut
+
+    # PyDevd "socket" entry points (and related helpers)
 
     def pydevd_notify(self, cmd_id, args):
         # TODO: docstring
@@ -737,37 +800,6 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
             raise UnsupportedPyDevdCommandError(cmd_id)
         return f(self, seq, args)
 
-    def async_method(m):
-        """Converts a generator method into an async one."""
-        m = futures.wrap_async(m)
-
-        def f(self, *args, **kwargs):
-            return m(self, self.loop, *args, **kwargs)
-
-        return f
-
-    def async_handler(m):
-        """Converts a generator method into a fire-and-forget async one."""
-        m = futures.wrap_async(m)
-
-        def f(self, *args, **kwargs):
-            fut = m(self, self.loop, *args, **kwargs)
-
-            def done(fut):
-                try:
-                    fut.result()
-                except BaseException:
-                    traceback.print_exc(file=sys.__stderr__)
-
-            fut.add_done_callback(done)
-
-        return f
-
-    def sleep(self):
-        fut = futures.Future(self.loop)
-        self.loop.call_soon(lambda: fut.set_result(None))
-        return fut
-
     @staticmethod
     def parse_xml_response(args):
         return untangle.parse(io.BytesIO(args.encode('utf8'))).xml
@@ -785,6 +817,8 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
             provider._lock.release()
         yield futures.Result(context())
 
+    # VSC protocol handlers
+
     @async_handler
     def on_initialize(self, request, args):
         # TODO: docstring
@@ -896,18 +930,6 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
             options[key] = DEBUG_OPTIONS_PARSER[key](value)
         return options
 
-    def on_disconnect(self, request, args):
-        # TODO: docstring
-        if self.start_reason == 'launch':
-            self.disconnect_request = request
-            self.disconnect_request_event.set()
-            killProcess = not self._closed
-            self.close()
-            if killProcess and self.killonclose:
-                os.kill(os.getpid(), signal.SIGTERM)
-        else:
-            self.send_response(request)
-
     @async_handler
     def on_attach(self, request, args):
         # TODO: docstring
@@ -926,6 +948,13 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
             args.get('options', options))
         self.send_response(request)
 
+    def on_disconnect(self, request, args):
+        # TODO: docstring
+        if self.start_reason == 'launch':
+            self._handle_disconnect(request)
+        else:
+            self.send_response(request)
+
     def send_process_event(self, start_method):
         # TODO: docstring
         evt = {
@@ -1490,6 +1519,8 @@ class VSCodeMessageProcessor(ipcjson.SocketIO, ipcjson.IpcChannel):
         }
         self.send_response(request, **sys_info)
 
+    # PyDevd protocol event handlers
+
     @pydevd_events.handler(pydevd_comm.CMD_THREAD_CREATE)
     def on_pydevd_thread_create(self, seq, args):
         # If this is the first thread reported, report process creation

tests/helpers/protocol.py

@@ -180,6 +180,10 @@ class MessageDaemon(Daemon):
 
     STARTED = MessageDaemonStarted
 
+    EXTERNAL = None
+    PRINT_SENT_MESSAGES = False
+    PRINT_RECEIVED_MESSAGES = False
+
     @classmethod
     def validate_message(cls, msg):
         """Ensure the message is legitimate."""
@@ -267,6 +271,8 @@ class MessageDaemon(Daemon):
             raise
 
     def _add_received(self, msg):
+        if self.PRINT_RECEIVED_MESSAGES:
+            print('<--' if self.EXTERNAL else '-->', msg)
         self._received.append(msg)
         self._handle_message(msg)
 
@@ -290,6 +296,8 @@ class MessageDaemon(Daemon):
         return
 
     def _send_message(self, msg):
+        if self.PRINT_SENT_MESSAGES:
+            print('-->' if self.EXTERNAL else '<--', msg)
         msg = self._protocol.parse(msg)
         raw = self._protocol.encode(msg)
         try:

tests/helpers/pydevd/_binder.py

@@ -0,0 +1,172 @@
+from collections import namedtuple
+import threading
+import time
+
+from ptvsd import wrapper
+from tests.helpers import socket
+
+
+class PTVSD(namedtuple('PTVSD', 'client server proc fakesock')):
+    """A wrapper around a running "instance" of PTVSD.
+
+    "client" and "server" are the two ends of socket that PTVSD uses
+    to communicate with the editor (e.g. VSC) via the VSC debug adapter
+    protocol.  "server" will be None for a remote address.
+    "proc" is the wrapper around the VSC message handler.
+    "fakesock" is the socket-like object that PTVSD uses to communicate
+    with the debugger (e.g. PyDevd) via the PyDevd wire protocol.
+    """
+
+    @classmethod
+    def from_connect_func(cls, connect):
+        """Return a new instance using the socket returned by connect()."""
+        client, server = connect()
+        fakesock = wrapper._start(
+            client,
+            server,
+            killonclose=False,
+            addhandlers=False,
+        )
+        proc = fakesock._vscprocessor
+        proc._exit_on_unknown_command = False
+        return cls(client, server, proc, fakesock)
+
+    def close(self):
+        """Stop PTVSD and clean up.
+
+        This will trigger the VSC protocol end-of-debugging message flow
+        (e.g. "exited" and "terminated" events).  As part of that flow
+        this function may block while waiting for specific messages from
+        the editor (e.g. a "disconnect" request).  PTVSD also closes all
+        of its background threads and closes any sockets it controls.
+        """
+        self.proc.close()
+
+
+class BinderBase(object):
+    """Base class for one-off socket binders (for protocol daemons).
+
+    A "binder" facilitates separating the socket-binding behavior from
+    the socket-connecting behavior.  This matters because for server
+    sockets the connecting part is a blocking operation.
+
+    The bind method may be passed to protocol.Daemon() as the "bind"
+    argument.
+
+    Note that a binder starts up ptvsd using the connected socket and
+    runs the debugger in the background.
+    """
+
+    def __init__(self, address=None, ptvsd=None):
+        if address is not None or ptvsd is not None:
+            raise NotImplementedError
+
+        # Set when bind() called:
+        self.address = None
+        self._connect = None
+        self._waiter = None
+
+        # Set when ptvsd started:
+        self._thread = None
+        self.ptvsd = None
+
+    def __repr__(self):
+        return '{}(address={!r}, ptvsd={!r})'.format(
+            type(self).__name__,
+            self.address,
+            self.ptvsd,
+        )
+
+    def bind(self, address):
+        """Return (connect func, remote addr) after binding a socket.
+
+        A new client or server socket is immediately bound, depending on
+        the address.  Then the connect func is generated for that
+        socket.  The func takes no args and returns a client socket
+        connected to the original address.  In the case of a remote
+        address, that socket may be the one that was originally bound.
+
+        When the connect func is called, PTVSD is started up using the
+        socket.  Then some debugging operation (e.g. running a script
+        through pydevd) is started in a background thread.
+        """
+        if self._connect is not None:
+            raise RuntimeError('already bound')
+        self.address = address
+        self._connect, remote = socket.bind(address)
+        self._waiter = threading.Lock()
+        self._waiter.acquire()
+
+        def connect():
+            if self._thread is not None:
+                raise RuntimeError('already connected')
+            self._thread = threading.Thread(target=self._run)
+            self._thread.start()
+            # Wait for ptvsd to start up.
+            if self._waiter.acquire(timeout=1):
+                self._waiter.release()
+            else:
+                raise RuntimeError('timed out')
+            return self._wrap_sock()
+        return connect, remote
+
+    def wait_until_done(self):
+        """Wait for the started debugger operation to finish."""
+        if self._thread is None:
+            return
+        self._thread.join()
+
+    ####################
+    # for subclassing
+
+    def _run_debugger(self):
+        # Subclasses import this.  The method must directly or
+        # indirectly call self._start_ptvsd().
+        raise NotImplementedError
+
+    def _wrap_sock(self):
+        return socket.Connection(self.ptvsd.client, self.ptvsd.server)
+        #return socket.Connection(self.ptvsd.fakesock, self.ptvsd.server)
+
+    ####################
+    # internal methods
+
+    def _start_ptvsd(self):
+        if self.ptvsd is not None:
+            raise RuntimeError('already connected')
+        self.ptvsd = PTVSD.from_connect_func(self._connect)
+        self._waiter.release()
+
+    def _run(self):
+        try:
+            self._run_debugger()
+        except SystemExit as exc:
+            wrapper.ptvsd_sys_exit_code = int(exc.code)
+            raise
+        wrapper.ptvsd_sys_exit_code = 0
+        self.ptvsd.proc.close()
+
+
+class Binder(BinderBase):
+    """A "binder" that defers the debugging operation to an external func.
+
+    That function takes two arguments, "external" and "internal", and
+    returns nothing.  "external" is a socket that an editor (or fake)
+    may use to communicate with PTVSD over the VSC debug adapter
+    protocol.  "internal does the same for a debugger and the PyDevd
+    wire protocol.  The function should exit once debugging has
+    finished.
+    """
+
+    def __init__(self, do_debugging=None):
+        if do_debugging is None:
+            def do_debugging(external, internal):
+                time.sleep(5)
+        super(Binder, self).__init__()
+        self._do_debugging = do_debugging
+
+    def _run_debugger(self):
+        self._start_ptvsd()
+        external = self.ptvsd.server
+        internal = self.ptvsd.fakesock
+        self._do_debugging(external, internal)

tests/helpers/pydevd/_fake.py

@@ -1,10 +1,12 @@
+import threading
+
 from _pydevd_bundle.pydevd_comm import (
     CMD_VERSION,
 )
 
-import ptvsd.wrapper as _ptvsd
 from ._pydevd import parse_message, encode_message, iter_messages, Message
 from tests.helpers import protocol, socket
+from ._binder import BinderBase
 
 
 PROTOCOL = protocol.MessageProtocol(
@@ -14,28 +16,34 @@ PROTOCOL = protocol.MessageProtocol(
 )
 
 
-def _bind(address):
-    connect, remote = socket.bind(address)
+class Binder(BinderBase):
 
-    def connect(_connect=connect):
-        client, server = _connect()
-        pydevd = _ptvsd._start(client, server,
-                               killonclose=False,
-                               addhandlers=False)
-        pydevd._vscprocessor._exit_on_unknown_command = False
-        return socket.Connection(pydevd, server)
-    return connect, remote
+    def __init__(self):
+        super(Binder, self).__init__()
+        self._lock = threading.Lock()
+        self._lock.acquire()
+
+    def _run_debugger(self):
+        self._start_ptvsd()
+        # Block until "done" debugging.
+        self._lock.acquire()
+
+    def _wrap_sock(self):
+        return socket.Connection(self.ptvsd.fakesock, self.ptvsd.server)
+
+    def _done(self):
+        self._lock.release()
 
 
 class Started(protocol.MessageDaemonStarted):
 
     def send_response(self, msg):
         self.wait_until_connected()
-        return self.fake.send_response(msg)
+        return self.daemon.send_response(msg)
 
     def send_event(self, msg):
         self.wait_until_connected()
-        return self.fake.send_event(msg)
+        return self.daemon.send_event(msg)
 
 
 class FakePyDevd(protocol.MessageDaemon):
@@ -66,6 +74,7 @@ class FakePyDevd(protocol.MessageDaemon):
     """  # noqa
 
     STARTED = Started
+    EXTERNAL = False
 
     PROTOCOL = PROTOCOL
     VERSION = '1.1.1'
@@ -99,8 +108,10 @@ class FakePyDevd(protocol.MessageDaemon):
             return None
 
     def __init__(self, handler=None):
+        self.binder = Binder()
+
         super(FakePyDevd, self).__init__(
-            _bind,
+            self.binder.bind,
             PROTOCOL,
             (lambda msg, send: self.handle_request(msg, send, handler)),
         )
@@ -136,3 +147,7 @@ class FakePyDevd(protocol.MessageDaemon):
             send_message(resp)
             return True
         self.add_handler(handle_request, handlername)
+
+    def _close(self):
+        self.binder._done()
+        super(FakePyDevd, self)._close()

tests/helpers/pydevd/_live.py

@@ -2,79 +2,45 @@ import os
 import os.path
 import sys
 import threading
+import warnings
 
 import _pydevd_bundle.pydevd_comm as pydevd_comm
 
-from ptvsd import wrapper, debugger
-from tests.helpers import protocol, socket
+from ptvsd import debugger
+from tests.helpers import protocol
+from ._binder import BinderBase
 
 
-class Binder(object):
+class Binder(BinderBase):
 
     def __init__(self, filename, module):
+        super(Binder, self).__init__()
         self.filename = filename
         self.module = module
+        self._lock = threading.Lock()
+        self._lock.acquire()
 
-        self.address = None
-        self._waiter = None
-        self._connect = None
-
-        self._thread = None
-        self.client = None
-        self.server = None
-        self.fakesock = None
-        self.proc = None
-
-    def bind(self, address):
-        if self._connect is not None:
-            raise RuntimeError('already bound')
-        self.address = address
-        self._connect, remote = socket.bind(address)
-        self._waiter = threading.Lock()
-        self._waiter.acquire()
-
-        def connect():
-            self._thread = threading.Thread(target=self.run_pydevd)
-            self._thread.start()
-            if self._waiter.acquire(timeout=1):
-                self._waiter.release()
-            else:
-                raise RuntimeError('timed out')
-            return socket.Connection(self.client, self.server)
-            #return socket.Connection(self.fakesock, self.server)
-        return connect, remote
-
-    def new_pydevd_sock(self, *args):
-        if self.client is not None:
-            raise RuntimeError('already connected')
-        self.client, self.server = self._connect()
-        self.fakesock = wrapper._start(self.client, self.server,
-                                       killonclose=False,
-                                       addhandlers=False)
-        self.proc = self.fakesock._vscprocessor
-        self._waiter.release()
-        return self.fakesock
-
-    def run_pydevd(self):
-        pydevd_comm.start_server = self.new_pydevd_sock
-        pydevd_comm.start_client = self.new_pydevd_sock
+    def _run_debugger(self):
+        def new_pydevd_sock(*args):
+            self._start_ptvsd()
+            return self.ptvsd.fakesock
+        pydevd_comm.start_server = new_pydevd_sock
+        pydevd_comm.start_client = new_pydevd_sock
         # Force a fresh pydevd.
         sys.modules.pop('pydevd', None)
-        try:
-            if self.module is None:
-                debugger._run_file(self.address, self.filename)
-            else:
-                debugger._run_module(self.address, self.module)
-        except SystemExit as exc:
-            wrapper.ptvsd_sys_exit_code = int(exc.code)
-            raise
-        wrapper.ptvsd_sys_exit_code = 0
-        self.proc.close()
+        if self.module is None:
+            debugger._run_file(self.address, self.filename)
+        else:
+            debugger._run_module(self.address, self.module)
 
-    def wait_until_done(self):
-        if self._thread is None:
-            return
-        self._thread.join()
+        # Block until "done" debugging.
+        if not self._lock.acquire(timeout=3):
+            # This shouldn't happen since the timeout on event waiting
+            # is this long.
+            warnings.warn('timeout out waiting for "done"')
+
+    def done(self):
+        self._lock.release()
 
 
 class LivePyDevd(protocol.Daemon):
@@ -101,6 +67,9 @@ class LivePyDevd(protocol.Daemon):
         super(LivePyDevd, self).__init__(self.binder.bind)
 
     def _close(self):
+        # Note that we do not call self.binder.done() here, though it
+        # might make sense as a fallback.  Instead, we do so directly
+        # in the relevant test cases.
         super(LivePyDevd, self)._close()
         # TODO: Close pydevd somehow?
 

tests/helpers/vsc/_fake.py

@@ -26,7 +26,7 @@ class Started(protocol.MessageDaemonStarted):
 
     def send_request(self, msg):
         self.wait_until_connected()
-        return self.fake.send_request(msg)
+        return self.daemon.send_request(msg)
 
 
 class FakeVSC(protocol.MessageDaemon):
@@ -59,6 +59,7 @@ class FakeVSC(protocol.MessageDaemon):
     """  # noqa
 
     STARTED = Started
+    EXTERNAL = True
 
     PROTOCOL = PROTOCOL
 

tests/highlevel/__init__.py

@@ -1,10 +1,12 @@
 from collections import namedtuple
 import contextlib
 import platform
+import threading
 try:
     import urllib.parse as urllib
 except ImportError:
     import urllib
+import warnings
 
 from _pydevd_bundle import pydevd_xml
 from _pydevd_bundle.pydevd_comm import (
@@ -29,6 +31,11 @@ from tests.helpers.vsc import FakeVSC
 OS_ID = 'WINDOWS' if platform.system() == 'Windows' else 'UNIX'
 
 
+@contextlib.contextmanager
+def noop_cm(*args, **kwargs):
+    yield
+
+
 class Thread(namedtuple('Thread', 'id name')):
     """Information about a thread."""
 
@@ -305,15 +312,27 @@ class VSCLifecycle(object):
         self._pydevd = pydevd
         self._hidden = hidden or fix.hidden
 
-    def launched(self, port=None, hidedisconnect=False, **kwargs):
-        def start():
-            self.launch(**kwargs)
-        return self._started(start, port, hidedisconnect=hidedisconnect)
+    @contextlib.contextmanager
+    def daemon_running(self, port=None, hide=False):
+        with self._fix.hidden() if hide else noop_cm():
+            daemon = self._start_daemon(port)
+        try:
+            yield
+        finally:
+            with self._fix.hidden() if hide else noop_cm():
+                self._stop_daemon(daemon)
 
-    def attached(self, port=None, hidedisconnect=False, **kwargs):
-        def start():
+    @contextlib.contextmanager
+    def launched(self, port=None, hide=False, **kwargs):
+        with self.daemon_running(port, hide=hide):
+            self.launch(**kwargs)
+            yield
+
+    @contextlib.contextmanager
+    def attached(self, port=None, hide=False, **kwargs):
+        with self.daemon_running(port, hide=hide):
             self.attach(**kwargs)
-        return self._started(start, port, hidedisconnect=hidedisconnect)
+            yield
 
     def launch(self, **kwargs):
         """Initialize the debugger protocol and then launch."""
@@ -325,30 +344,42 @@ class VSCLifecycle(object):
         with self._hidden():
             self._handshake('attach', **kwargs)
 
-    def disconnect(self, **reqargs):
-        self._send_request('disconnect', reqargs)
+    def disconnect(self, exitcode=0, **reqargs):
+        wrapper.ptvsd_sys_exit_code = exitcode
+        self._fix.send_request('disconnect', reqargs)
         # TODO: wait for an exit event?
         # TODO: call self._fix.vsc.close()?
 
     # internal methods
 
-    @contextlib.contextmanager
-    def _started(self, start, port, hidedisconnect=False):
+    def _start_daemon(self, port):
         if port is None:
             port = self.PORT
         addr = (None, port)
-        with self._fix.fake.start(addr):
-            with self._fix.disconnect_when_done(hide=hidedisconnect):
-                start()
-                yield
+        daemon = self._fix.fake.start(addr)
+        with self._hidden():
+            with self._fix.wait_for_event('output'):
+                daemon.wait_until_connected()
+        return daemon
+
+    def _stop_daemon(self, daemon):
+        # We must close ptvsd directly (rather than closing the external
+        # socket (i.e. "daemon").  This is because cloing ptvsd blocks,
+        # keeping us from sending the disconnect request we need to send
+        # at the end.
+        t = threading.Thread(target=self._fix.close_ptvsd)
+        with self._fix.wait_for_events(['exited', 'terminated']):
+            # The thread runs close_ptvsd(), which sends the two
+            # events and then waits for a "disconnect" request.  We send
+            # that after we receive the events.
+            t.start()
+        self.disconnect()
+        t.join()
+        daemon.close()
 
     def _handshake(self, command, threadnames=None, config=None,
                    default_threads=True, process=True, reset=True,
                    **kwargs):
-        with self._hidden():
-            with self._fix.wait_for_event('output'):
-                pass
-
         initargs = dict(
             kwargs.pop('initargs', None) or {},
             disconnect=kwargs.pop('disconnect', True),
@@ -433,6 +464,9 @@ class FixtureBase(object):
             return self._fake
         except AttributeError:
             self._fake = self.new_fake()
+            # Uncomment the following 2 lines to see all messages.
+            #self._fake.PRINT_SENT_MESSAGES = True
+            #self._fake.PRINT_RECEIVED_MESSAGES = True
             return self._fake
 
     @property
@@ -583,6 +617,15 @@ class VSCFixture(FixtureBase):
             self._lifecycle = self.LIFECYCLE(self)
             return self._lifecycle
 
+    @property
+    def _proc(self):
+        # This is used below in close_ptvsd().
+        # TODO: This is a horrendous use of internal details!
+        try:
+            return self.fake._adapter.daemon.binder.ptvsd.proc
+        except AttributeError:
+            return None
+
     def send_request(self, command, args=None, handle_response=None):
         kwargs = dict(args or {}, handler=handle_response)
         with self._wait_for_response(command, **kwargs) as req:
@@ -606,34 +649,19 @@ class VSCFixture(FixtureBase):
             self.msgs.next_event()
 
     @contextlib.contextmanager
-    def _wait_for_events(self, events):
+    def wait_for_events(self, events):
         if not events:
             yield
             return
-        with self._wait_for_events(events[1:]):
+        with self.wait_for_events(events[1:]):
             with self.wait_for_event(events[0]):
                 yield
 
-    @contextlib.contextmanager
-    def disconnect_when_done(self, hide=True):
-        try:
-            yield
-        finally:
-            if hide:
-                with self.hidden():
-                    self._disconnect()
-            else:
-                self._disconnect()
-
-    def _disconnect(self):
-        with self.exits_after(exitcode=0):
-            self.send_request('disconnect')
-
-    @contextlib.contextmanager
-    def exits_after(self, exitcode):
-        wrapper.ptvsd_sys_exit_code = exitcode
-        with self._wait_for_events(['exited', 'terminated']):
-            yield
+    def close_ptvsd(self):
+        if self._proc is None:
+            warnings.warn('"proc" not bound')
+        else:
+            self._proc.close()
 
 
 class HighlevelFixture(object):
@@ -752,6 +780,11 @@ class HighlevelFixture(object):
         with self._vsc.wait_for_event(event, *args, **kwargs):
             yield
 
+    @contextlib.contextmanager
+    def wait_for_events(self, events):
+        with self._vsc.wait_for_events(events):
+            yield
+
     @contextlib.contextmanager
     def expect_debugger_command(self, cmdid):
         with self._pydevd.expected_command(cmdid):
@@ -763,10 +796,8 @@ class HighlevelFixture(object):
     def send_debugger_event(self, cmdid, payload):
         self._pydevd.send_event(cmdid, payload)
 
-    @contextlib.contextmanager
-    def disconnect_when_done(self):
-        with self._vsc.disconnect_when_done():
-            yield
+    def close_ptvsd(self):
+        self._vsc.close_ptvsd()
 
     # combinations
 
@@ -805,7 +836,7 @@ class HighlevelFixture(object):
 
         # Send and handle messages.
         self._pydevd.set_threads_response()
-        with self._vsc._wait_for_events(['thread' for _ in newthreads]):
+        with self.wait_for_events(['thread' for _ in newthreads]):
             self.send_request('threads')
         self._known_threads.update(newthreads)
 

tests/highlevel/test_lifecycle.py

@@ -32,10 +32,10 @@ class LifecycleTests(HighlevelTest, unittest.TestCase):
     def test_attach(self):
         version = self.debugger.VERSION
         addr = (None, 8888)
-        with self.vsc.start(addr):
-            with self.vsc.wait_for_event('output'):
-                pass
-
+        daemon = self.vsc.start(addr)
+        with self.vsc.wait_for_event('output'):
+            daemon.wait_until_connected()
+        try:
             with self.vsc.wait_for_event('initialized'):
                 # initialize
                 self.set_debugger_response(CMD_VERSION, version)
@@ -53,7 +53,10 @@ class LifecycleTests(HighlevelTest, unittest.TestCase):
 
             # end
             req_disconnect = self.send_request('disconnect')
-        # An "exited" event comes once self.vsc closes.
+        finally:
+            with self._fix.wait_for_events(['exited', 'terminated']):
+                self.fix.close_ptvsd()
+            daemon.close()
 
         self.assert_received(self.vsc, [
             self.new_event(
@@ -95,6 +98,7 @@ class LifecycleTests(HighlevelTest, unittest.TestCase):
             #)),
             self.new_response(req_disconnect),
             self.new_event('exited', exitCode=0),
+            self.new_event('terminated'),
         ])
         self.assert_received(self.debugger, [
             self.debugger_msgs.new_request(CMD_VERSION,
@@ -126,8 +130,8 @@ class LifecycleTests(HighlevelTest, unittest.TestCase):
             # Normal ops would go here.
 
             # end
-            req_disconnect = self.send_request('disconnect')
-        # An "exited" event comes once self.vsc closes.
+            with self.fix.wait_for_events(['exited', 'terminated']):
+                req_disconnect = self.send_request('disconnect')
 
         self.assert_received(self.vsc, [
             self.new_event(

tests/highlevel/test_live_pydevd.py

@@ -20,6 +20,10 @@ class Fixture(VSCFixture):
             start_adapter=self._pydevd.start,
         )
 
+    @property
+    def _proc(self):
+        return self._pydevd.binder.ptvsd.proc
+
     @property
     def binder(self):
         return self._pydevd.binder
@@ -103,8 +107,9 @@ class LifecycleTests(TestBase, unittest.TestCase):
             # Normal ops would go here.
 
             # end
-            with self._wait_for_events(['exited', 'terminated']):
-                pass
+            with self.wait_for_events(['exited', 'terminated']):
+                self.fix.binder.done()
+            # TODO: Send a "disconnect" request?
             self.fix.binder.wait_until_done()
             received = self.vsc.received
 
@@ -153,7 +158,7 @@ class VSCFlowTest(TestBase):
     @contextlib.contextmanager
     def launched(self, port=8888, **kwargs):
         kwargs.setdefault('process', False)
-        with self.lifecycle.launched(port=port, hidedisconnect=True, **kwargs):
+        with self.lifecycle.launched(port=port, hide=True, **kwargs):
             #with self.fix.install_sig_handler():
                 yield
 
@@ -194,8 +199,9 @@ class BreakpointTests(VSCFlowTest, unittest.TestCase):
 
     def test_no_breakpoints(self):
         with self.launched():
-            # All the script to run to completion.
+            # Allow the script to run to completion.
             received = self.vsc.received
+            self.fix.binder.done()
 
         self.assert_received(self.vsc, [])
         self.assert_vsc_received(received, [])

tests/highlevel/test_messages.py

@@ -94,14 +94,12 @@ class InitializeTests(LifecycleTest, unittest.TestCase):
     @unittest.skip('tested via test_lifecycle.py')
     def test_basic(self):
         version = self.debugger.VERSION
-        addr = (None, 8888)
-        with self.vsc.start(addr):
-            with self.disconnect_when_done():
-                self.set_debugger_response(CMD_VERSION, version)
-                req = self.send_request('initialize', {
-                    'adapterID': 'spam',
-                })
-                received = self.vsc.received
+        with self.lifecycle.demon_running(port=8888):
+            self.set_debugger_response(CMD_VERSION, version)
+            req = self.send_request('initialize', {
+                'adapterID': 'spam',
+            })
+            received = self.vsc.received
 
         self.assert_vsc_received(received, [
             self.new_response(req, **dict(