diff --git a/tests/helpers/protocol.py b/tests/helpers/protocol.py
new file mode 100644
index 000000000..a61ed736e
--- /dev/null
+++ b/tests/helpers/protocol.py
@@ -0,0 +1,214 @@
+from collections import namedtuple
+import threading
+
+from . import socket
+
+
+class StreamFailure(Exception):
+    """Something went wrong while handling messages to/from a stream."""
+
+    def __init__(self, direction, msg, exception):
+        err = 'error while processing stream: {!r}'.format(exception)
+        super(StreamFailure, self).__init__(self, err)
+        self.direction = direction
+        self.msg = msg
+        self.exception = exception
+
+    def __repr__(self):
+        return '{}(direction={!r}, msg={!r}, exception={!r})'.format(
+            type(self).__name__,
+            self.direction,
+            self.msg,
+            self.exception,
+        )
+
+
+class MessageProtocol(namedtuple('Protocol', 'parse encode iter')):
+    """A basic abstraction of a message protocol.
+
+    parse(msg) - returns a message for the given data.
+    encode(msg) - returns the message, serialized to the line-format.
+    iter(stream, stop) - yield each message from the stream.  "stop"
+        is a function called with no args which returns True if the
+        iterator should stop.
+    """
+
+
+class Started(object):
+    """A simple wrapper around a started message protocol daemon."""
+
+    def __init__(self, fake):
+        self.fake = fake
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    def send_message(self, msg):
+        return self.fake.send_response(msg)
+
+    def close(self):
+        self.fake.close()
+
+
+class Daemon(object):
+    """A testing double for a protocol daemon."""
+
+    STARTED = Started
+
+    @classmethod
+    def validate_message(cls, msg):
+        """Ensure the message is legitimate."""
+        # By default check nothing.
+
+    def __init__(self, connect, protocol, handler):
+        self._connect = connect
+        self._protocol = protocol
+
+        self._closed = False
+        self._received = []
+        self._failures = []
+
+        self._handlers = []
+        self._default_handler = handler
+
+        # These are set when we start.
+        self._host = None
+        self._port = None
+        self._sock = None
+        self._server = None
+        self._listener = None
+
+    @property
+    def received(self):
+        """All the messages received thus far."""
+        return list(self._received)
+
+    @property
+    def failures(self):
+        """All send/recv failures thus far."""
+        return self._failures
+
+    def start(self, host, port):
+        """Start the fake daemon.
+
+        This calls the earlier provided connect() function.
+
+        A listener loop is started in another thread to handle incoming
+        messages from the socket.
+        """
+        self._host = host or None
+        self._port = port
+        self._start()
+        return self.STARTED(self)
+
+    def send_message(self, msg):
+        """Serialize msg to the line format and send it to the socket."""
+        if self._closed:
+            raise EOFError('closed')
+        self._validate_message(msg)
+        self._send_message(msg)
+
+    def close(self):
+        """Clean up the daemon's resources (e.g. sockets, files, listener)."""
+        if self._closed:
+            return
+
+        self._closed = True
+        self._close()
+
+    def add_handler(self, handler, oneoff=True):
+        """Add the given handler to the list of possible handlers."""
+        entry = (handler, 1 if oneoff else None)
+        self._handlers.append(entry)
+        return handler
+
+    def reset(self, force=False):
+        """Clear the recorded messages."""
+        if self._failures:
+            raise RuntimeError('have failures ({!r})'.format(self._failures))
+        if self._handlers:
+            if force:
+                self._handlers = []
+            else:
+                raise RuntimeError('have pending handlers')
+        self._received = []
+
+    def assert_received(self, case, expected):
+        """Ensure that the received messages match the expected ones."""
+        received = [self._protocol.parse(msg) for msg in self._received]
+        expected = [self._protocol.parse(msg) for msg in expected]
+        case.assertEqual(received, expected)
+
+    # internal methods
+
+    def _start(self, host=None):
+        self._sock, self._server = self._connect(
+            host or self._host,
+            self._port,
+        )
+
+        # TODO: make it a daemon thread?
+        self._listener = threading.Thread(target=self._listen)
+        self._listener.start()
+
+    def _listen(self):
+        with self._sock.makefile('rb') as sockfile:
+            for msg in self._protocol.iter(sockfile, lambda: self._closed):
+                if isinstance(msg, StreamFailure):
+                    self._failures.append(msg)
+                else:
+                    self._add_received(msg)
+
+    def _add_received(self, msg):
+        self._received.append(msg)
+        self._handle_message(msg)
+
+    def _handle_message(self, msg):
+        for i, entry in enumerate(list(self._handlers)):
+            handle_message, remaining = entry
+            handled = handle_message(msg, self._send_message)
+            if handled or handled is None:
+                if remaining is not None:
+                    if remaining == 1:
+                        self._handlers.pop(i)
+                    else:
+                        self._handlers[i] = (handle_message, remaining-1)
+                return handled
+        else:
+            if self._default_handler is not None:
+                return self._default_handler(msg, self._send_message)
+            return False
+
+    def _validate_message(self, msg):
+        return
+
+    def _send_message(self, msg):
+        msg = self._protocol.parse(msg)
+        raw = self._protocol.encode(msg)
+        try:
+            self._send(raw)
+        except Exception as exc:
+            failure = StreamFailure('send', msg, exc)
+            self._failures.append(failure)
+
+    def _send(self, raw):
+        while raw:
+            sent = self._sock.send(raw)
+            raw = raw[sent:]
+
+    def _close(self):
+        if self._sock is not None:
+            socket.close(self._sock)
+            self._sock = None
+        if self._server is not None:
+            socket.close(self._server)
+            self._server = None
+        if self._listener is not None:
+            self._listener.join(timeout=1)
+            # TODO: the listener isn't stopping!
+            #if self._listener.is_alive():
+            #    raise RuntimeError('timed out')
+            self._listener = None
diff --git a/tests/helpers/pydevd/_fake.py b/tests/helpers/pydevd/_fake.py
index 6e38b5d30..4e7d9bb30 100644
--- a/tests/helpers/pydevd/_fake.py
+++ b/tests/helpers/pydevd/_fake.py
@@ -1,33 +1,27 @@
-import socket
-import threading
+from _pydevd_bundle.pydevd_comm import (
+    CMD_VERSION,
+)
 
 from ptvsd.wrapper import start_server, start_client
+from ._pydevd import parse_message, encode_message, iter_messages, Message
+from tests.helpers import protocol
 
-from ._pydevd import parse_message, iter_messages, StreamFailure
 
-
-def socket_close(sock):
-    sock.shutdown(socket.SHUT_RDWR)
-    sock.close()
+PROTOCOL = protocol.MessageProtocol(
+    parse=parse_message,
+    encode=encode_message,
+    iter=iter_messages,
+)
 
 
 def _connect(host, port):
     if host is None:
-        return start_server(port)
+        return start_server(port), None
     else:
-        return start_client(host, port)
-
-
-class _Started(object):
-
-    def __init__(self, fake):
-        self.fake = fake
+        return start_client(host, port), None
 
-    def __enter__(self):
-        return self
 
-    def __exit__(self, *args):
-        self.close()
+class Started(protocol.Started):
 
     def send_response(self, msg):
         return self.fake.send_response(msg)
@@ -35,11 +29,8 @@ def send_response(self, msg):
     def send_event(self, msg):
         return self.fake.send_event(msg)
 
-    def close(self):
-        self.fake.close()
 
-
-class FakePyDevd(object):
+class FakePyDevd(protocol.Daemon):
     """A testing double for PyDevd.
 
     Note that you have the option to provide a handler function.  This
@@ -66,118 +57,67 @@ class FakePyDevd(object):
       https://github.com/fabioz/PyDev.Debugger/blob/master/_pydevd_bundle/pydevd_comm.py
     """  # noqa
 
-    CONNECT = staticmethod(_connect)
-
-    def __init__(self, handler=None, connect=None):
-        if connect is None:
-            connect = self.CONNECT
-
-        self._handler = handler
-        self._connect = connect
-
-        self._closed = False
-        self._received = []
-        self._failures = []
+    STARTED = Started
 
-        # These are set when we start.
-        self._host = None
-        self._port = None
-        self._sock = None
-        self._listener = None
+    VERSION = '1.1.1'
 
-    @property
-    def received(self):
-        """All the messages received thus far."""
-        return list(self._received)
+    @classmethod
+    def validate_message(cls, msg):
+        """Ensure the message is legitimate."""
+        # TODO: Check the message.
 
-    @property
-    def failures(self):
-        """All send/recv failures thus far."""
-        return self._failures
+    @classmethod
+    def handle_request(cls, req, send_message, handler=None):
+        """The default message handler."""
+        if handler is not None:
+            handler(req, send_message)
 
-    def start(self, host, port):
-        """Start the fake pydevd daemon.
+        resp = cls._get_response(req)
+        if resp is not None:
+            send_message(resp)
 
-        This calls the earlier provided connect() function.  By default
-        this calls either start_server() or start_client() (depending on
-        the host) from ptvsd.wrapper.  Thus the ptvsd message processor
-        is started and a PydevdSocket is used as the connection.
-
-        A listener loop is started in another thread to handle incoming
-        messages from the socket (i.e. from ptvsd).
-        """
-        self._host = host or None
-        self._port = port
-        self._sock = self._connect(self._host, self._port)
-
-        # TODO: make daemon?
-        self._listener = threading.Thread(target=self._listen)
-        self._listener.start()
-
-        return _Started(self)
+    @classmethod
+    def _get_response(cls, req):
+        try:
+            cmdid, seq, _ = req
+        except (IndexError, ValueError):
+            req = req.msg
+            cmdid, seq, _ = req
+
+        if cmdid == CMD_VERSION:
+            return Message(CMD_VERSION, seq, cls.VERSION)
+        else:
+            return None
+
+    def __init__(self, handler=None):
+        super(FakePyDevd, self).__init__(
+            _connect,
+            PROTOCOL,
+            (lambda msg, send: self.handle_request(msg, send, handler)),
+        )
 
     def send_response(self, msg):
         """Send a response message to the adapter (ptvsd)."""
+        # XXX Ensure it's a response?
         return self._send_message(msg)
 
     def send_event(self, msg):
         """Send an event message to the adapter (ptvsd)."""
-        return self._send_message(msg)
-
-    def close(self):
-        """If started, close the socket and wait for the listener to finish."""
-        if self._closed:
-            return
-
-        self._closed = True
-        if self._sock is not None:
-            socket_close(self._sock)
-            self._sock = None
-        if self._listener is not None:
-            self._listener.join(timeout=1)
-            # TODO: the listener isn't stopping!
-            #if self._listener.is_alive():
-            #    raise RuntimeError('timed out')
-            self._listener = None
-
-    def assert_received(self, case, expected):
-        """Ensure that the received messages match the expected ones."""
-        received = [parse_message(msg) for msg in self._received]
-        expected = [parse_message(msg) for msg in expected]
-        case.assertEqual(received, expected)
-
-    # internal methods
-
-    def _listen(self):
-        with self._sock.makefile('rb') as sockfile:
-            for msg in iter_messages(sockfile, lambda: self._closed):
-                if isinstance(msg, StreamFailure):
-                    self._failures.append(msg)
-                else:
-                    self._add_received(msg)
-
-    def _add_received(self, msg):
-        self._received.append(msg)
-
-        if self._handler is not None:
-            self._handler(msg, self._send_message)
-
-    def _send_message(self, msg):
-        """Serialize the message to the line format and send it to ptvsd.
-
-        If the message is bytes or a string then it is send as-is.
-        """
-        msg = parse_message(msg)
-        raw = msg.as_bytes()
-        if not raw.endswith(b'\n'):
-            raw += b'\n'
-        try:
-            self._send(raw)
-        except Exception as exc:
-            failure = StreamFailure('send', msg, exc)
-            self._failures.append(failure)
-
-    def _send(self, raw):
-        while raw:
-            sent = self._sock.send(raw)
-            raw = raw[sent:]
+        # XXX Ensure it's a request?
+        return self.send_message(msg)
+
+    def add_pending_response(self, cmdid, text):
+        """Add a response for a request."""
+        def handle_request(req, send_message, respid=cmdid):
+            try:
+                cmdid, seq, _ = req
+            except (IndexError, ValueError):
+                req = req.msg
+                cmdid, seq, _ = req
+            if cmdid != respid:
+                return False
+            resp = Message(cmdid, seq, text)
+            send_message(resp)
+            return True
+
+        self.add_handler(handle_request)
diff --git a/tests/helpers/pydevd/_pydevd.py b/tests/helpers/pydevd/_pydevd.py
index 691c6adac..68a6cfd05 100644
--- a/tests/helpers/pydevd/_pydevd.py
+++ b/tests/helpers/pydevd/_pydevd.py
@@ -1,25 +1,36 @@
 from collections import namedtuple
+try:
+    from urllib.parse import quote, unquote
+except ImportError:
+    from urllib import quote, unquote
+
+from tests.helpers.protocol import StreamFailure
 
 # TODO: Everything here belongs in a proper pydevd package.
 
 
-class StreamFailure(Exception):
-    """Something went wrong while handling messages to/from a stream."""
+def parse_message(msg):
+    """Return a message object for the given "msg" data."""
+    if type(msg) is bytes:
+        return Message.from_bytes(msg)
+    elif isinstance(msg, str):
+        return Message.from_bytes(msg)
+    elif type(msg) is RawMessage:
+        return msg.msg
+    elif type(msg) is Message:
+        return msg
+    elif isinstance(msg, tuple):
+        return Message(*msg)
+    else:
+        raise NotImplementedError
 
-    def __init__(self, direction, msg, exception):
-        err = 'error while processing stream: {!r}'.format(exception)
-        super(StreamFailure, self).__init__(self, err)
-        self.direction = direction
-        self.msg = msg
-        self.exception = exception
 
-    def __repr__(self):
-        return '{}(direction={!r}, msg={!r}, exception={!r})'.format(
-            type(self).__name__,
-            self.direction,
-            self.msg,
-            self.exception,
-        )
+def encode_message(msg):
+    """Return the message, serialized to the line-format."""
+    raw = msg.as_bytes()
+    if not raw.endswith(b'\n'):
+        raw += b'\n'
+    return raw
 
 
 def iter_messages(stream, stop=lambda: False):
@@ -36,18 +47,6 @@ def iter_messages(stream, stop=lambda: False):
             yield StreamFailure('recv', None, exc)
 
 
-def parse_message(msg):
-    """Return a message object for the given "msg" data."""
-    if type(msg) is bytes:
-        return RawMessage.from_bytes(msg)
-    elif isinstance(msg, str):
-        return RawMessage.from_bytes(msg)
-    elif type(msg) is RawMessage:
-        return msg
-    else:
-        raise NotImplementedError
-
-
 class RawMessage(namedtuple('RawMessage', 'bytes')):
     """A pydevd message class that leaves the raw bytes unprocessed."""
 
@@ -65,6 +64,68 @@ def __new__(cls, raw):
         self = super(RawMessage, cls).__new__(cls, raw)
         return self
 
+    @property
+    def msg(self):
+        try:
+            return self._msg
+        except AttributeError:
+            self._msg = Message.from_bytes(self.bytes)
+            return self._msg
+
     def as_bytes(self):
         """Return the line-formatted bytes corresponding to the message."""
         return self.bytes
+
+
+class Message(namedtuple('Message', 'cmdid seq payload')):
+    """A de-seralized PyDevd message."""
+
+    @classmethod
+    def from_bytes(cls, raw):
+        """Return a RawMessage corresponding to the given raw message."""
+        raw = RawMessage.from_bytes(raw)
+        parts = raw.bytes.split(b'\t', 2)
+        return cls(*parts)
+
+    @classmethod
+    def parse_payload(cls, payload):
+        """Return the de-serialized payload."""
+        if isinstance(payload, bytes):
+            payload = payload.decode('utf-8')
+        if isinstance(payload, str):
+            text = unquote(payload)
+            return cls._parse_payload_text(text)
+        elif hasattr(payload, 'as_text'):
+            return payload
+        else:
+            raise ValueError('unsupported payload {!r}'.format(payload))
+
+    @classmethod
+    def _parse_payload_text(cls, text):
+        # TODO: convert to the appropriate payload type.
+        return text
+
+    def __new__(cls, cmdid, seq, payload):
+        cmdid = int(cmdid) if cmdid or cmdid == 0 else None
+        seq = int(seq) if seq or seq == 0 else None
+        payload = cls.parse_payload(payload)
+        self = super(Message, cls).__new__(cls, cmdid, seq, payload)
+        return self
+
+    def __init__(self, *args, **kwargs):
+        if self.cmdid is None:
+            raise TypeError('missing cmdid')
+        if self.seq is None:
+            raise TypeError('missing seq')
+
+    def as_bytes(self):
+        """Return the line-formatted bytes corresponding to the message."""
+        try:
+            payload_as_text = self.payload.as_text
+        except AttributeError:
+            text = self.payload
+        else:
+            text = payload_as_text()
+        payload = quote(text)
+        data = '{}\t{}\t{}'.format(self.cmdid, self.seq, payload)
+        return data.encode('utf-8')
diff --git a/tests/helpers/socket.py b/tests/helpers/socket.py
new file mode 100644
index 000000000..298321543
--- /dev/null
+++ b/tests/helpers/socket.py
@@ -0,0 +1,38 @@
+from __future__ import absolute_import
+
+import socket
+
+
+def connect(host, port):
+    """Return (client, server) after connecting.
+
+    If host is None then it's a server, so it will wait for a connection
+    on localhost.  Otherwise it will connect to the remote host.
+    """
+    sock = socket.socket(
+        socket.AF_INET,
+        socket.SOCK_STREAM,
+        socket.IPPROTO_TCP,
+    )
+    sock.setsockopt(
+        socket.SOL_SOCKET,
+        socket.SO_REUSEADDR,
+        1,
+    )
+    if host is None:
+        addr = ('127.0.0.1', port)
+        server = sock
+        server.bind(addr)
+        server.listen(1)
+        sock, _ = server.accept()
+    else:
+        addr = (host, port)
+        sock.connect(addr)
+        server = None
+    return sock, server
+
+
+def close(sock):
+    """Shutdown and close the socket."""
+    sock.shutdown(socket.SHUT_RDWR)
+    sock.close()
diff --git a/tests/helpers/vsc/_fake.py b/tests/helpers/vsc/_fake.py
index 3b9d5c605..81405ce66 100644
--- a/tests/helpers/vsc/_fake.py
+++ b/tests/helpers/vsc/_fake.py
@@ -1,34 +1,24 @@
-import socket
+import contextlib
 import threading
 
-from ._vsc import StreamFailure, encode_message, iter_messages, parse_message
-from ._vsc import RawMessage  # noqa
+from tests.helpers import protocol, socket
+from ._vsc import encode_message, iter_messages, parse_message
 
 
-def socket_close(sock):
-    sock.shutdown(socket.SHUT_RDWR)
-    sock.close()
+PROTOCOL = protocol.MessageProtocol(
+    parse=parse_message,
+    encode=encode_message,
+    iter=iter_messages,
+)
 
 
-class _Started(object):
-
-    def __init__(self, fake):
-        self.fake = fake
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, *args):
-        self.close()
+class Started(protocol.Started):
 
     def send_request(self, msg):
         return self.fake.send_request(msg)
 
-    def close(self):
-        self.fake.close()
-
 
-class FakeVSC(object):
+class FakeVSC(protocol.Daemon):
     """A testing double for a VSC debugger protocol client.
 
     This class facilitates sending VSC debugger protocol messages over
@@ -58,52 +48,55 @@ class FakeVSC(object):
     """  # noqa
 
     def __init__(self, start_adapter, handler=None):
-        def start_adapter(host, port, start_adapter=start_adapter):
-            self._adapter = start_adapter(host, port)
+        super(FakeVSC, self).__init__(socket.connect, PROTOCOL, handler)
 
-        self._start_adapter = start_adapter
-        self._handler = handler
-
-        self._closed = False
-        self._received = []
-        self._failures = []
+        def start_adapter(host, port, _start_adapter=start_adapter):
+            self._adapter = _start_adapter(host, port)
 
-        # These are set when we start.
-        self._host = None
-        self._port = None
+        self._start_adapter = start_adapter
         self._adapter = None
-        self._sock = None
-        self._server = None
-        self._listener = None
-
-    @property
-    def addr(self):
-        host, port = self._host, self._port
-        if host is None:
-            host = '127.0.0.1'
-        return (host, port)
-
-    @property
-    def received(self):
-        """All the messages received thus far."""
-        return list(self._received)
-
-    @property
-    def failures(self):
-        """All send/recv failures thus far."""
-        return self._failures
 
     def start(self, host, port):
         """Start the fake and the adapter."""
-        if self._closed or self._adapter is not None:
+        if self._adapter is not None:
             raise RuntimeError('already started')
+        return super(FakeVSC, self).start(host, port)
+
+    def send_request(self, req):
+        """Send the given Request object."""
+        return self.send_message(req)
+
+    def wait_for_response(self, req, **kwargs):
+        reqseq = req['seq']
+        command = req['command']
+
+        def match(msg):
+            msg = msg.data
+            if msg['type'] != 'response' or msg['request_seq'] != reqseq:
+                return False
+            assert(msg['command'] == command)
+            return True
 
-        if not host:
+        return self._wait_for_message(match, req, **kwargs)
+
+    def wait_for_event(self, event, **kwargs):
+        def match(msg):
+            msg = msg.data
+            if msg['type'] != 'event' or msg['event'] != event:
+                return False
+            return True
+
+        return self._wait_for_message(match, req=None, **kwargs)
+
+    # internal methods
+
+    def _start(self, host=None):
+        start_adapter = (lambda: self._start_adapter(self._host, self._port))
+        if not self._host:
             # The adapter is the server so start it first.
-            t = threading.Thread(
-                target=lambda: self._start_adapter(host, port))
+            t = threading.Thread(target=start_adapter)
             t.start()
-            self._start('127.0.0.1', port)
+            super(FakeVSC, self)._start('127.0.0.1')
             t.join(timeout=1)
             if t.is_alive():
                 raise RuntimeError('timed out')
@@ -111,107 +104,34 @@ def start(self, host, port):
             # The adapter is the client so start it last.
             # TODO: For now don't use this.
             raise NotImplementedError
-            t = threading.Thread(
-                target=lambda: self._start(host, port))
+            t = threading.Thread(target=super(FakeVSC, self)._start)
             t.start()
-            self._start_adapter(host, port)
+            start_adapter()
             t.join(timeout=1)
             if t.is_alive():
                 raise RuntimeError('timed out')
 
-        return _Started(self)
-
-    def send_request(self, req):
-        """Send the given Request object."""
-        if self._closed:
-            raise EOFError('closed')
-        req = parse_message(req)
-        raw = encode_message(req)
-        try:
-            self._send(raw)
-        except Exception as exc:
-            failure = ('send', req, exc)
-            self._failures.append(failure)
-
-    def close(self):
-        """Close the fake's resources (e.g. socket, adapter)."""
-        if self._closed:
-            return
-
-        self._closed = True
-        self._close()
-
-    def assert_received(self, case, expected):
-        """Ensure that the received messages match the expected ones."""
-        received = [parse_message(msg) for msg in self._received]
-        expected = [parse_message(msg) for msg in expected]
-        case.assertEqual(received, expected)
-
-    # internal methods
-
-    def _start(self, host, port):
-        self._host = host
-        self._port = port
-        self._connect()
-
-        # TODO: make daemon?
-        self._listener = threading.Thread(target=self._listen)
-        self._listener.start()
-
-    def _connect(self):
-        sock = socket.socket(
-            socket.AF_INET,
-            socket.SOCK_STREAM,
-            socket.IPPROTO_TCP,
-        )
-        sock.setsockopt(
-            socket.SOL_SOCKET,
-            socket.SO_REUSEADDR,
-            1,
-        )
-        if self._host is None:
-            server = sock
-            server.bind(self.addr)
-            server.listen(1)
-            sock, _ = server.accept()
-        else:
-            sock.connect(self.addr)
-            server = None
-        self._server = server
-        self._sock = sock
-
-    def _listen(self):
-        with self._sock.makefile('rb') as sockfile:
-            for msg in iter_messages(sockfile, lambda: self._closed):
-                if isinstance(msg, StreamFailure):
-                    self._failures.append(msg)
-                else:
-                    self._add_received(msg)
-
-    def _add_received(self, msg):
-        self._received.append(msg)
-
-        if self._handler is not None:
-            self._handler(msg, self.send_request)
-
-    def _send(self, raw):
-        while raw:
-            sent = self._sock.send(raw)
-            raw = raw[sent:]
-
     def _close(self):
         if self._adapter is not None:
             self._adapter.close()
             self._adapter = None
-        if self._sock is not None:
-            socket_close(self._sock)
-            self._sock = None
-        if self._server is not None:
-            socket_close(self._server)
-            self._server = None
-        if self._listener is not None:
-            self._listener.join(timeout=1)
-            # TODO: the listener isn't stopping!
-            #if self._listener.is_alive():
-            #    raise RuntimeError('timed out')
-            self._listener = None
+        super(FakeVSC, self)._close()
+
+    @contextlib.contextmanager
+    def _wait_for_message(self, match, req=None, handler=None, timeout=1):
+        lock = threading.Lock()
+        lock.acquire()
+
+        def handle_message(msg, send_message):
+            if match(msg):
+                lock.release()
+                if handler is not None:
+                    handler(msg, send_message)
+            else:
+                return False
+        self.add_handler(handle_message)
+
+        yield req
+
+        lock.acquire(timeout=timeout)  # Wait for the message to match.
+        lock.release()
diff --git a/tests/helpers/vsc/_vsc.py b/tests/helpers/vsc/_vsc.py
index 5db4080b1..9b214a747 100644
--- a/tests/helpers/vsc/_vsc.py
+++ b/tests/helpers/vsc/_vsc.py
@@ -2,27 +2,22 @@
 import json
 
 from debugger_protocol.messages import wireformat
+from tests.helpers.protocol import StreamFailure
 
 # TODO: Use more of the code from debugger_protocol.
 
 
-class StreamFailure(Exception):
-    """Something went wrong while handling messages to/from a stream."""
-
-    def __init__(self, direction, msg, exception):
-        err = 'error while processing stream: {!r}'.format(exception)
-        super(StreamFailure, self).__init__(self, err)
-        self.direction = direction
-        self.msg = msg
-        self.exception = exception
-
-    def __repr__(self):
-        return '{}(direction={!r}, msg={!r}, exception={!r})'.format(
-            type(self).__name__,
-            self.direction,
-            self.msg,
-            self.exception,
-        )
+def parse_message(msg):
+    """Return a message object for the given "msg" data."""
+    if type(msg) is str:
+        data = json.loads(msg)
+    elif isinstance(msg, bytes):
+        data = json.loads(msg.decode('utf-8'))
+    elif type(msg) is RawMessage:
+        return msg
+    else:
+        data = msg
+    return RawMessage.from_data(**data)
 
 
 def encode_message(msg):
@@ -42,19 +37,6 @@ def iter_messages(stream, stop=lambda: False):
             yield StreamFailure('recv', None, exc)
 
 
-def parse_message(msg):
-    """Return a message object for the given "msg" data."""
-    if type(msg) is str:
-        data = json.loads(msg)
-    elif isinstance(msg, bytes):
-        data = json.loads(msg.decode('utf-8'))
-    elif type(msg) is RawMessage:
-        return msg
-    else:
-        data = msg
-    return RawMessage.from_data(**data)
-
-
 class RawMessage(namedtuple('RawMessage', 'data')):
     """A wrapper around a line-formatted debugger protocol message."""
 
diff --git a/tests/ptvsd/highlevel/__init__.py b/tests/ptvsd/highlevel/__init__.py
index 678df08cb..42e96272f 100644
--- a/tests/ptvsd/highlevel/__init__.py
+++ b/tests/ptvsd/highlevel/__init__.py
@@ -1,6 +1,12 @@
+import contextlib
 import platform
 import unittest
 
+from _pydevd_bundle.pydevd_comm import (
+    CMD_RUN,
+    CMD_VERSION,
+)
+
 from tests.helpers.pydevd import FakePyDevd
 from tests.helpers.vsc import FakeVSC
 
@@ -9,11 +15,126 @@
 
 
 class HighlevelTestCase(unittest.TestCase):
+    """The base class for high-level ptvsd tests."""
+
+    MIN_INITIALIZE_ARGS = {
+        'adapterID': 'spam',
+    }
+
+    def next_vsc_seq(self):
+        try:
+            seq = self._seq
+        except AttributeError:
+            seq = 0
+        self._seq = seq + 1
+        return seq
 
     def new_fake(self, pydevd=None, handler=None):
+        """Return a new fake VSC that may be used in tests."""
         if pydevd is None:
             pydevd = FakePyDevd()
         vsc = FakeVSC(pydevd.start, handler)
         self.addCleanup(vsc.close)
 
         return vsc, pydevd
+
+    @contextlib.contextmanager
+    def launched(self, vsc, pydevd, port=8888, **kwargs):
+        with vsc.start(None, port):
+            try:
+                self.launch(vsc, pydevd, **kwargs)
+                yield
+            finally:
+                self.disconnect(vsc)
+                vsc._received.pop(-1)
+
+    def attach(self, vsc, pydevd, **kwargs):
+        """Initialize the debugger protocol and then attach."""
+        self._handshake(vsc, pydevd, 'attach', **kwargs)
+
+    def launch(self, vsc, pydevd, **kwargs):
+        """Initialize the debugger protocol and then launch."""
+        self._handshake(vsc, pydevd, 'launch', **kwargs)
+
+    def _handshake(self, vsc, pydevd, command,
+                   breakpoints=None, excbreakpoints=None,
+                   reset=True, **kwargs):
+        initargs = dict(
+            kwargs.pop('initargs', None) or {},
+            disconnect=kwargs.pop('disconnect', True),
+        )
+        with vsc.wait_for_event('initialized'):
+            self._initialize(vsc, pydevd, **initargs)
+            req = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': command,
+                'arguments': kwargs,
+            }
+            with vsc.wait_for_response(req):
+                vsc.send_request(req)
+
+        # Handle breakpoints
+        if breakpoints:
+            req = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'setBreakpoints',
+                'arguments': self._parse_breakpoints(breakpoints),
+            }
+            with vsc.wait_for_response(req):
+                vsc.send_request(req)
+        if excbreakpoints:
+            req = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'setExceptionBreakpoints',
+                'arguments': self._parse_breakpoints(excbreakpoints),
+            }
+            with vsc.wait_for_response(req):
+                vsc.send_request(req)
+        pydevd.add_pending_response(CMD_RUN, '')
+        req = {
+            'type': 'request',
+            'seq': self.next_vsc_seq(),
+            'command': 'configurationDone',
+            'arguments': {}
+        }
+        with vsc.wait_for_response(req):
+            vsc.send_request(req)
+
+        if reset:
+            vsc.reset()
+            pydevd.reset()
+
+    def _initialize(self, vsc, pydevd, **reqargs):
+        """
+        See https://code.visualstudio.com/docs/extensionAPI/api-debugging#_the-vs-code-debug-protocol-in-a-nutshell
+        """  # noqa
+        def handle_response(resp, _):
+            self._capabilities = resp.data['body']
+        pydevd.add_pending_response(CMD_VERSION, pydevd.VERSION)
+        req = {
+            'type': 'request',
+            'seq': self.next_vsc_seq(),
+            'command': 'initialize',
+            'arguments':  dict(self.MIN_INITIALIZE_ARGS, **reqargs),
+        }
+        with vsc.wait_for_response(req, handler=handle_response):
+            vsc.send_request(req)
+
+    def _parse_breakpoints(self, breakpoints):
+        raise NotImplementedError
+        #for bp in breakpoints or ():
+
+    def disconnect(self, vsc, **reqargs):
+        req = {
+            'type': 'request',
+            'seq': self.next_vsc_seq(),
+            'command': 'disconnect',
+            'arguments': reqargs,
+        }
+        with vsc.wait_for_response(req):
+            vsc.send_request(req)
+        # TODO: wait for an exit event?
+#        vsc.close()
diff --git a/tests/ptvsd/highlevel/test_basic.py b/tests/ptvsd/highlevel/test_basic.py
deleted file mode 100644
index 10aa2c4b9..000000000
--- a/tests/ptvsd/highlevel/test_basic.py
+++ /dev/null
@@ -1,109 +0,0 @@
-import threading
-import unittest
-
-from _pydevd_bundle.pydevd_comm import (
-    CMD_VERSION,
-)
-
-from tests.helpers.pydevd import FakePyDevd
-from . import OS_ID, HighlevelTestCase
-
-
-@unittest.skip('finish!')
-class LivecycleTests(HighlevelTestCase):
-    """
-    See https://code.visualstudio.com/docs/extensionAPI/api-debugging#_the-vs-code-debug-protocol-in-a-nutshell
-    """  # noqa
-
-    def test_startup(self):
-        raise NotImplementedError
-
-    def test_shutdown(self):
-        raise NotImplementedError
-
-
-class MessageTests(HighlevelTestCase):
-
-    def test_initialize(self):
-        self.pseq = -1
-        plock = threading.Lock()
-        plock.acquire()
-
-        def handle_pydevd(msg, _):
-            try:
-                seq = msg.bytes.split(b'\t')[1]
-            except IndexError:
-                return
-            self.pseq = int(seq.decode('utf-8'))
-            plock.release()
-        pydevd = FakePyDevd(handle_pydevd)
-
-        self.num_left_vsc = 2
-        vlock = threading.Lock()
-        vlock.acquire()
-
-        def handle_vsp(msg, _):
-            if self.num_left_vsc == 0:
-                return
-            self.num_left_vsc -= 1
-            if self.num_left_vsc == 0:
-                vlock.release()
-        vsc, _ = self.new_fake(pydevd, handle_vsp)
-        with vsc.start(None, 8888):
-            vsc.send_request({
-                'type': 'request',
-                'seq': 42,
-                'command': 'initialize',
-                'arguments': {
-                    'adapterID': 'spam',
-                },
-            })
-            plock.acquire(timeout=1)
-            pydevd.send_response(
-                '{}\t{}\t<VERSION>'.format(CMD_VERSION, self.pseq))
-            plock.release()
-            vlock.acquire(timeout=1)  # wait for 2 messages to come back
-            vlock.release()
-
-        self.maxDiff = None
-        self.assertFalse(pydevd.failures)
-        self.assertFalse(vsc.failures)
-        vsc.assert_received(self, [
-            {
-                'type': 'response',
-                'seq': 0,
-                'request_seq': 42,
-                'command': 'initialize',
-                'success': True,
-                'message': '',
-                'body': dict(
-                    supportsExceptionInfoRequest=True,
-                    supportsConfigurationDoneRequest=True,
-                    supportsConditionalBreakpoints=True,
-                    supportsSetVariable=True,
-                    supportsExceptionOptions=True,
-                    exceptionBreakpointFilters=[
-                        {
-                            'filter': 'raised',
-                            'label': 'Raised Exceptions',
-                            'default': 'true'
-                        },
-                        {
-                            'filter': 'uncaught',
-                            'label': 'Uncaught Exceptions',
-                            'default': 'true'
-                        },
-                    ],
-                ),
-            },
-            {
-                'type': 'event',
-                'seq': 1,
-                'event': 'initialized',
-                'body': {},
-            },
-        ])
-        seq = 1000000000
-        pydevd.assert_received(self, [
-            '{}\t{}\t1.1\t{}\tID'.format(CMD_VERSION, seq, OS_ID),
-        ])
diff --git a/tests/ptvsd/highlevel/test_lifecycle.py b/tests/ptvsd/highlevel/test_lifecycle.py
new file mode 100644
index 000000000..2c75b3be8
--- /dev/null
+++ b/tests/ptvsd/highlevel/test_lifecycle.py
@@ -0,0 +1,283 @@
+import os
+import sys
+
+from _pydevd_bundle.pydevd_comm import (
+    CMD_RUN,
+    CMD_VERSION,
+)
+
+from . import OS_ID, HighlevelTestCase
+
+
+class LifecycleTests(HighlevelTestCase):
+    """
+    See https://code.visualstudio.com/docs/extensionAPI/api-debugging#_the-vs-code-debug-protocol-in-a-nutshell
+    """  # noqa
+
+    def test_attach(self):
+        vsc, pydevd = self.new_fake()
+
+        with vsc.start(None, 8888):
+            with vsc.wait_for_event('initialized'):
+                # initialize
+                pydevd.add_pending_response(CMD_VERSION, pydevd.VERSION)
+                req_initialize = {
+                    'type': 'request',
+                    'seq': self.next_vsc_seq(),
+                    'command': 'initialize',
+                    'arguments': {
+                        'adapterID': 'spam',
+                    },
+                }
+                with vsc.wait_for_response(req_initialize):
+                    vsc.send_request(req_initialize)
+
+                # attach
+                req_attach = {
+                    'type': 'request',
+                    'seq': self.next_vsc_seq(),
+                    'command': 'attach',
+                    'arguments': {},
+                }
+                with vsc.wait_for_response(req_attach):
+                    vsc.send_request(req_attach)
+
+            # configuration
+            pydevd.add_pending_response(CMD_RUN, '')
+            req_config = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'configurationDone',
+                'arguments': {}
+            }
+            with vsc.wait_for_response(req_config):
+                vsc.send_request(req_config)
+
+            # end
+            req_disconnect = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'disconnect',
+                'arguments': {},
+            }
+            with vsc.wait_for_response(req_disconnect):
+                vsc.send_request(req_disconnect)
+
+        self.assertFalse(pydevd.failures)
+        self.assertFalse(vsc.failures)
+        vsc.assert_received(self, [
+            {
+                'type': 'response',
+                'seq': 0,
+                'request_seq': req_initialize['seq'],
+                'command': 'initialize',
+                'success': True,
+                'message': '',
+                'body': dict(
+                    supportsExceptionInfoRequest=True,
+                    supportsConfigurationDoneRequest=True,
+                    supportsConditionalBreakpoints=True,
+                    supportsSetVariable=True,
+                    supportsExceptionOptions=True,
+                    exceptionBreakpointFilters=[
+                        {
+                            'filter': 'raised',
+                            'label': 'Raised Exceptions',
+                            'default': 'true'
+                        },
+                        {
+                            'filter': 'uncaught',
+                            'label': 'Uncaught Exceptions',
+                            'default': 'true'
+                        },
+                    ],
+                ),
+            },
+            {
+                'type': 'event',
+                'seq': 1,
+                'event': 'initialized',
+                'body': {},
+            },
+            {
+                'type': 'response',
+                'seq': 2,
+                'request_seq': req_attach['seq'],
+                'command': 'attach',
+                'success': True,
+                'message': '',
+                'body': {},
+            },
+            {
+                'type': 'response',
+                'seq': 3,
+                'request_seq': req_config['seq'],
+                'command': 'configurationDone',
+                'success': True,
+                'message': '',
+                'body': {},
+            },
+            {
+                'type': 'event',
+                'seq': 4,
+                'event': 'process',
+                'body': {
+                    'name': sys.argv[0],
+                    'systemProcessId': os.getpid(),
+                    'isLocalProcess': True,
+                    'startMethod': 'attach',
+                },
+            },
+            {
+                'type': 'response',
+                'seq': 5,
+                'request_seq': req_disconnect['seq'],
+                'command': 'disconnect',
+                'success': True,
+                'message': '',
+                'body': {},
+            },
+        ])
+        pydevd.assert_received(self, [
+            # (cmdid, seq, text)
+            (
+                CMD_VERSION,
+                1000000000,
+                '\t'.join(['1.1', OS_ID, 'ID']),
+            ),
+            (CMD_RUN, 1000000001, ''),
+        ])
+
+    def test_launch(self):
+        vsc, pydevd = self.new_fake()
+
+        with vsc.start(None, 8888):
+            with vsc.wait_for_event('initialized'):
+                # initialize
+                pydevd.add_pending_response(CMD_VERSION, pydevd.VERSION)
+                req_initialize = {
+                    'type': 'request',
+                    'seq': self.next_vsc_seq(),
+                    'command': 'initialize',
+                    'arguments': {
+                        'adapterID': 'spam',
+                    },
+                }
+                with vsc.wait_for_response(req_initialize):
+                    vsc.send_request(req_initialize)
+
+                # launch
+                req_launch = {
+                    'type': 'request',
+                    'seq': self.next_vsc_seq(),
+                    'command': 'launch',
+                    'arguments': {},
+                }
+                with vsc.wait_for_response(req_launch):
+                    vsc.send_request(req_launch)
+
+            # configuration
+            pydevd.add_pending_response(CMD_RUN, '')
+            req_config = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'configurationDone',
+                'arguments': {}
+            }
+            with vsc.wait_for_response(req_config):
+                vsc.send_request(req_config)
+
+            # end
+            req_disconnect = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'disconnect',
+                'arguments': {},
+            }
+            with vsc.wait_for_response(req_disconnect):
+                vsc.send_request(req_disconnect)
+
+        self.assertFalse(pydevd.failures)
+        self.assertFalse(vsc.failures)
+        vsc.assert_received(self, [
+            {
+                'type': 'response',
+                'seq': 0,
+                'request_seq': req_initialize['seq'],
+                'command': 'initialize',
+                'success': True,
+                'message': '',
+                'body': dict(
+                    supportsExceptionInfoRequest=True,
+                    supportsConfigurationDoneRequest=True,
+                    supportsConditionalBreakpoints=True,
+                    supportsSetVariable=True,
+                    supportsExceptionOptions=True,
+                    exceptionBreakpointFilters=[
+                        {
+                            'filter': 'raised',
+                            'label': 'Raised Exceptions',
+                            'default': 'true'
+                        },
+                        {
+                            'filter': 'uncaught',
+                            'label': 'Uncaught Exceptions',
+                            'default': 'true'
+                        },
+                    ],
+                ),
+            },
+            {
+                'type': 'event',
+                'seq': 1,
+                'event': 'initialized',
+                'body': {},
+            },
+            {
+                'type': 'response',
+                'seq': 2,
+                'request_seq': req_launch['seq'],
+                'command': 'launch',
+                'success': True,
+                'message': '',
+                'body': {},
+            },
+            {
+                'type': 'response',
+                'seq': 3,
+                'request_seq': req_config['seq'],
+                'command': 'configurationDone',
+                'success': True,
+                'message': '',
+                'body': {},
+            },
+            {
+                'type': 'event',
+                'seq': 4,
+                'event': 'process',
+                'body': {
+                    'name': sys.argv[0],
+                    'systemProcessId': os.getpid(),
+                    'isLocalProcess': True,
+                    'startMethod': 'launch',
+                },
+            },
+            {
+                'type': 'response',
+                'seq': 5,
+                'request_seq': req_disconnect['seq'],
+                'command': 'disconnect',
+                'success': True,
+                'message': '',
+                'body': {},
+            },
+        ])
+        pydevd.assert_received(self, [
+            # (cmdid, seq, text)
+            (
+                CMD_VERSION,
+                1000000000,
+                '\t'.join(['1.1', OS_ID, 'ID']),
+            ),
+            (CMD_RUN, 1000000001, ''),
+        ])
diff --git a/tests/ptvsd/highlevel/test_messages.py b/tests/ptvsd/highlevel/test_messages.py
new file mode 100644
index 000000000..421bb1743
--- /dev/null
+++ b/tests/ptvsd/highlevel/test_messages.py
@@ -0,0 +1,143 @@
+from _pydevd_bundle.pydevd_comm import (
+    CMD_LIST_THREADS,
+    CMD_VERSION,
+)
+
+from . import OS_ID, HighlevelTestCase
+
+
+class RequestTests(HighlevelTestCase):
+    """
+    lifecycle (in order):
+
+    initialize
+    attach
+    launch
+    setBreakpoints
+    setExceptionBreakpoints
+    configurationDone
+    disconnect
+
+    normal operation:
+
+    threads
+    stackTrace
+    scopes
+    variables
+    setVariable
+    evaluate
+    pause
+    continue
+    next
+    stepIn
+    stepOut
+    setBreakpoints
+    setExceptionBreakpoints
+    exceptionInfo
+    """
+
+    def test_initialize(self):
+        vsc, pydevd = self.new_fake()
+
+        with vsc.start(None, 8888):
+            try:
+                pydevd.add_pending_response(CMD_VERSION, pydevd.VERSION)
+                req = {
+                    'type': 'request',
+                    'seq': self.next_vsc_seq(),
+                    'command': 'initialize',
+                    'arguments': {
+                        'adapterID': 'spam',
+                    },
+                }
+                with vsc.wait_for_response(req):
+                    vsc.send_request(req)
+            finally:
+                self.disconnect(vsc)
+                vsc._received.pop(-1)
+
+        self.assertFalse(pydevd.failures)
+        self.assertFalse(vsc.failures)
+        vsc.assert_received(self, [
+            {
+                'type': 'response',
+                'seq': 0,
+                'request_seq': req['seq'],
+                'command': 'initialize',
+                'success': True,
+                'message': '',
+                'body': dict(
+                    supportsExceptionInfoRequest=True,
+                    supportsConfigurationDoneRequest=True,
+                    supportsConditionalBreakpoints=True,
+                    supportsSetVariable=True,
+                    supportsExceptionOptions=True,
+                    exceptionBreakpointFilters=[
+                        {
+                            'filter': 'raised',
+                            'label': 'Raised Exceptions',
+                            'default': 'true'
+                        },
+                        {
+                            'filter': 'uncaught',
+                            'label': 'Uncaught Exceptions',
+                            'default': 'true'
+                        },
+                    ],
+                ),
+            },
+            {
+                'type': 'event',
+                'seq': 1,
+                'event': 'initialized',
+                'body': {},
+            },
+        ])
+        seq = 1000000000
+        text = '\t'.join(['1.1', OS_ID, 'ID'])
+        pydevd.assert_received(self, [
+            (CMD_VERSION, seq, text),
+        ])
+
+    def test_threads(self):
+        vsc, pydevd = self.new_fake()
+        with self.launched(vsc, pydevd):
+            pydevd.add_pending_response(CMD_LIST_THREADS, """
+                <xml>
+                <thread name="spam" id="10" />
+                <thread name="pydevd.spam" id="11" />
+                <thread name="" id="12" />
+                </xml>
+            """.strip().replace('\n', ''))
+            req = {
+                'type': 'request',
+                'seq': self.next_vsc_seq(),
+                'command': 'threads',
+                'arguments': {},
+            }
+            with vsc.wait_for_response(req):
+                vsc.send_request(req)
+
+        self.maxDiff = None
+        self.assertFalse(pydevd.failures)
+        self.assertFalse(vsc.failures)
+        vsc.assert_received(self, [
+            {
+                'type': 'response',
+                'seq': 5,
+                'request_seq': req['seq'],
+                'command': 'threads',
+                'success': True,
+                'message': '',
+                'body': {
+                    'threads': [
+                        {'id': 1, 'name': 'spam'},
+                        {'id': 3, 'name': ''},
+                    ],
+                },
+            },
+        ])
+        pydevd.assert_received(self, [
+            # (cmdid, seq, text)
+            (CMD_LIST_THREADS, 1000000002, ''),
+        ])