forward-port draft76 websockets

from tornado-3.2.2

only enabled during testing with phantomjs

IPython/html/allow76.py

@@ -0,0 +1,312
+"""WebsocketProtocol76 from tornado 3.2.2 for tornado >= 4.0
+
+The contents of this file are Copyright (c) Tornado
+Used under the Apache 2.0 license
+"""
+
+
+from __future__ import absolute_import, division, print_function, with_statement
+# Author: Jacob Kristhammar, 2010
+
+import functools
+import hashlib
+import struct
+import time
+import tornado.escape
+import tornado.web
+
+from tornado.log import gen_log, app_log
+from tornado.util import bytes_type, unicode_type
+
+from tornado.websocket import WebSocketHandler, WebSocketProtocol13
+
+class AllowDraftWebSocketHandler(WebSocketHandler):
+    """Restore Draft76 support for tornado 4
+    
+    Remove when we can run tests without phantomjs + qt4
+    """
+    
+    # get is unmodified except between the BEGIN/END PATCH lines
+    @tornado.web.asynchronous
+    def get(self, *args, **kwargs):
+        self.open_args = args
+        self.open_kwargs = kwargs
+
+        # Upgrade header should be present and should be equal to WebSocket
+        if self.request.headers.get("Upgrade", "").lower() != 'websocket':
+            self.set_status(400)
+            self.finish("Can \"Upgrade\" only to \"WebSocket\".")
+            return
+
+        # Connection header should be upgrade. Some proxy servers/load balancers
+        # might mess with it.
+        headers = self.request.headers
+        connection = map(lambda s: s.strip().lower(), headers.get("Connection", "").split(","))
+        if 'upgrade' not in connection:
+            self.set_status(400)
+            self.finish("\"Connection\" must be \"Upgrade\".")
+            return
+
+        # Handle WebSocket Origin naming convention differences
+        # The difference between version 8 and 13 is that in 8 the
+        # client sends a "Sec-Websocket-Origin" header and in 13 it's
+        # simply "Origin".
+        if "Origin" in self.request.headers:
+            origin = self.request.headers.get("Origin")
+        else:
+            origin = self.request.headers.get("Sec-Websocket-Origin", None)
+
+
+        # If there was an origin header, check to make sure it matches
+        # according to check_origin. When the origin is None, we assume it
+        # did not come from a browser and that it can be passed on.
+        if origin is not None and not self.check_origin(origin):
+            self.set_status(403)
+            self.finish("Cross origin websockets not allowed")
+            return
+
+        self.stream = self.request.connection.detach()
+        self.stream.set_close_callback(self.on_connection_close)
+
+        if self.request.headers.get("Sec-WebSocket-Version") in ("7", "8", "13"):
+            self.ws_connection = WebSocketProtocol13(
+                self, compression_options=self.get_compression_options())
+            self.ws_connection.accept_connection()
+        #--------------- BEGIN PATCH ----------------
+        elif (self.allow_draft76() and
+              "Sec-WebSocket-Version" not in self.request.headers):
+            self.ws_connection = WebSocketProtocol76(self)
+            self.ws_connection.accept_connection()
+        #--------------- END PATCH ----------------
+        else:
+            if not self.stream.closed():
+                self.stream.write(tornado.escape.utf8(
+                    "HTTP/1.1 426 Upgrade Required\r\n"
+                    "Sec-WebSocket-Version: 8\r\n\r\n"))
+                self.stream.close()
+    
+    # 3.2 methods removed in 4.0:
+    def allow_draft76(self):
+        """Using this class allows draft76 connections by default"""
+        return True
+    
+    def get_websocket_scheme(self):
+        """Return the url scheme used for this request, either "ws" or "wss".
+        This is normally decided by HTTPServer, but applications
+        may wish to override this if they are using an SSL proxy
+        that does not provide the X-Scheme header as understood
+        by HTTPServer.
+        Note that this is only used by the draft76 protocol.
+        """
+        return "wss" if self.request.protocol == "https" else "ws"
+    
+
+
+# No modifications from tornado-3.2.2 below this line
+
+class WebSocketProtocol(object):
+    """Base class for WebSocket protocol versions.
+    """
+    def __init__(self, handler):
+        self.handler = handler
+        self.request = handler.request
+        self.stream = handler.stream
+        self.client_terminated = False
+        self.server_terminated = False
+
+    def async_callback(self, callback, *args, **kwargs):
+        """Wrap callbacks with this if they are used on asynchronous requests.
+
+        Catches exceptions properly and closes this WebSocket if an exception
+        is uncaught.
+        """
+        if args or kwargs:
+            callback = functools.partial(callback, *args, **kwargs)
+
+        def wrapper(*args, **kwargs):
+            try:
+                return callback(*args, **kwargs)
+            except Exception:
+                app_log.error("Uncaught exception in %s",
+                              self.request.path, exc_info=True)
+                self._abort()
+        return wrapper
+
+    def on_connection_close(self):
+        self._abort()
+
+    def _abort(self):
+        """Instantly aborts the WebSocket connection by closing the socket"""
+        self.client_terminated = True
+        self.server_terminated = True
+        self.stream.close()  # forcibly tear down the connection
+        self.close()  # let the subclass cleanup
+
+
+class WebSocketProtocol76(WebSocketProtocol):
+    """Implementation of the WebSockets protocol, version hixie-76.
+
+    This class provides basic functionality to process WebSockets requests as
+    specified in
+    http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-76
+    """
+    def __init__(self, handler):
+        WebSocketProtocol.__init__(self, handler)
+        self.challenge = None
+        self._waiting = None
+
+    def accept_connection(self):
+        try:
+            self._handle_websocket_headers()
+        except ValueError:
+            gen_log.debug("Malformed WebSocket request received")
+            self._abort()
+            return
+
+        scheme = self.handler.get_websocket_scheme()
+
+        # draft76 only allows a single subprotocol
+        subprotocol_header = ''
+        subprotocol = self.request.headers.get("Sec-WebSocket-Protocol", None)
+        if subprotocol:
+            selected = self.handler.select_subprotocol([subprotocol])
+            if selected:
+                assert selected == subprotocol
+                subprotocol_header = "Sec-WebSocket-Protocol: %s\r\n" % selected
+
+        # Write the initial headers before attempting to read the challenge.
+        # This is necessary when using proxies (such as HAProxy), which
+        # need to see the Upgrade headers before passing through the
+        # non-HTTP traffic that follows.
+        self.stream.write(tornado.escape.utf8(
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n"
+            "Upgrade: WebSocket\r\n"
+            "Connection: Upgrade\r\n"
+            "Server: TornadoServer/%(version)s\r\n"
+            "Sec-WebSocket-Origin: %(origin)s\r\n"
+            "Sec-WebSocket-Location: %(scheme)s://%(host)s%(uri)s\r\n"
+            "%(subprotocol)s"
+            "\r\n" % (dict(
+                version=tornado.version,
+                origin=self.request.headers["Origin"],
+                scheme=scheme,
+                host=self.request.host,
+                uri=self.request.uri,
+                subprotocol=subprotocol_header))))
+        self.stream.read_bytes(8, self._handle_challenge)
+
+    def challenge_response(self, challenge):
+        """Generates the challenge response that's needed in the handshake
+
+        The challenge parameter should be the raw bytes as sent from the
+        client.
+        """
+        key_1 = self.request.headers.get("Sec-Websocket-Key1")
+        key_2 = self.request.headers.get("Sec-Websocket-Key2")
+        try:
+            part_1 = self._calculate_part(key_1)
+            part_2 = self._calculate_part(key_2)
+        except ValueError:
+            raise ValueError("Invalid Keys/Challenge")
+        return self._generate_challenge_response(part_1, part_2, challenge)
+
+    def _handle_challenge(self, challenge):
+        try:
+            challenge_response = self.challenge_response(challenge)
+        except ValueError:
+            gen_log.debug("Malformed key data in WebSocket request")
+            self._abort()
+            return
+        self._write_response(challenge_response)
+
+    def _write_response(self, challenge):
+        self.stream.write(challenge)
+        self.async_callback(self.handler.open)(*self.handler.open_args, **self.handler.open_kwargs)
+        self._receive_message()
+
+    def _handle_websocket_headers(self):
+        """Verifies all invariant- and required headers
+
+        If a header is missing or have an incorrect value ValueError will be
+        raised
+        """
+        fields = ("Origin", "Host", "Sec-Websocket-Key1",
+                  "Sec-Websocket-Key2")
+        if not all(map(lambda f: self.request.headers.get(f), fields)):
+            raise ValueError("Missing/Invalid WebSocket headers")
+
+    def _calculate_part(self, key):
+        """Processes the key headers and calculates their key value.
+
+        Raises ValueError when feed invalid key."""
+        # pyflakes complains about variable reuse if both of these lines use 'c'
+        number = int(''.join(c for c in key if c.isdigit()))
+        spaces = len([c2 for c2 in key if c2.isspace()])
+        try:
+            key_number = number // spaces
+        except (ValueError, ZeroDivisionError):
+            raise ValueError
+        return struct.pack(">I", key_number)
+
+    def _generate_challenge_response(self, part_1, part_2, part_3):
+        m = hashlib.md5()
+        m.update(part_1)
+        m.update(part_2)
+        m.update(part_3)
+        return m.digest()
+
+    def _receive_message(self):
+        self.stream.read_bytes(1, self._on_frame_type)
+
+    def _on_frame_type(self, byte):
+        frame_type = ord(byte)
+        if frame_type == 0x00:
+            self.stream.read_until(b"\xff", self._on_end_delimiter)
+        elif frame_type == 0xff:
+            self.stream.read_bytes(1, self._on_length_indicator)
+        else:
+            self._abort()
+
+    def _on_end_delimiter(self, frame):
+        if not self.client_terminated:
+            self.async_callback(self.handler.on_message)(
+                frame[:-1].decode("utf-8", "replace"))
+        if not self.client_terminated:
+            self._receive_message()
+
+    def _on_length_indicator(self, byte):
+        if ord(byte) != 0x00:
+            self._abort()
+            return
+        self.client_terminated = True
+        self.close()
+
+    def write_message(self, message, binary=False):
+        """Sends the given message to the client of this Web Socket."""
+        if binary:
+            raise ValueError(
+                "Binary messages not supported by this version of websockets")
+        if isinstance(message, unicode_type):
+            message = message.encode("utf-8")
+        assert isinstance(message, bytes_type)
+        self.stream.write(b"\x00" + message + b"\xff")
+
+    def write_ping(self, data):
+        """Send ping frame."""
+        raise ValueError("Ping messages not supported by this version of websockets")
+
+    def close(self):
+        """Closes the WebSocket connection."""
+        if not self.server_terminated:
+            if not self.stream.closed():
+                self.stream.write("\xff\x00")
+            self.server_terminated = True
+        if self.client_terminated:
+            if self._waiting is not None:
+                self.stream.io_loop.remove_timeout(self._waiting)
+            self._waiting = None
+            self.stream.close()
+        elif self._waiting is None:
+            self._waiting = self.stream.io_loop.add_timeout(
+                time.time() + 5, self._abort)
+

IPython/html/base/zmqhandlers.py

@@ -1,271 +1,272
 # coding: utf-8
 """Tornado handlers for WebSocket <-> ZMQ sockets."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
+import os
 import json
 import struct
+import warnings
 
 try:
     from urllib.parse import urlparse # Py 3
 except ImportError:
     from urlparse import urlparse # Py 2
 
 import tornado
-from tornado import gen, ioloop, web, websocket
+from tornado import gen, ioloop, web
+from tornado.websocket import WebSocketHandler
 
 from IPython.kernel.zmq.session import Session
 from IPython.utils.jsonutil import date_default, extract_dates
 from IPython.utils.py3compat import cast_unicode
 
 from .handlers import IPythonHandler
 
-
 def serialize_binary_message(msg):
     """serialize a message as a binary blob
 
     Header:
 
     4 bytes: number of msg parts (nbufs) as 32b int
     4 * nbufs bytes: offset for each buffer as integer as 32b int
 
     Offsets are from the start of the buffer, including the header.
 
     Returns
     -------
 
     The message serialized to bytes.
 
     """
     # don't modify msg or buffer list in-place
     msg = msg.copy()
     buffers = list(msg.pop('buffers'))
     bmsg = json.dumps(msg, default=date_default).encode('utf8')
     buffers.insert(0, bmsg)
     nbufs = len(buffers)
     offsets = [4 * (nbufs + 1)]
     for buf in buffers[:-1]:
         offsets.append(offsets[-1] + len(buf))
     offsets_buf = struct.pack('!' + 'I' * (nbufs + 1), nbufs, *offsets)
     buffers.insert(0, offsets_buf)
     return b''.join(buffers)
 
 
 def deserialize_binary_message(bmsg):
     """deserialize a message from a binary blog
 
     Header:
 
     4 bytes: number of msg parts (nbufs) as 32b int
     4 * nbufs bytes: offset for each buffer as integer as 32b int
 
     Offsets are from the start of the buffer, including the header.
 
     Returns
     -------
 
     message dictionary
     """
     nbufs = struct.unpack('!i', bmsg[:4])[0]
     offsets = list(struct.unpack('!' + 'I' * nbufs, bmsg[4:4*(nbufs+1)]))
     offsets.append(None)
     bufs = []
     for start, stop in zip(offsets[:-1], offsets[1:]):
         bufs.append(bmsg[start:stop])
     msg = json.loads(bufs[0].decode('utf8'))
     msg['header'] = extract_dates(msg['header'])
     msg['parent_header'] = extract_dates(msg['parent_header'])
     msg['buffers'] = bufs[1:]
     return msg
 
+# ping interval for keeping websockets alive (30 seconds)
+WS_PING_INTERVAL = 30000
+
+if os.environ.get('IPYTHON_ALLOW_DRAFT_WEBSOCKETS_FOR_PHANTOMJS', False):
+    warnings.warn("""Allowing draft76 websocket connections!
+    This should only be done for testing with phantomjs!""")
+    from IPython.html import allow76
+    WebSocketHandler = allow76.AllowDraftWebSocketHandler
+    # draft 76 doesn't support ping
+    WS_PING_INTERVAL = 0
 
-class ZMQStreamHandler(websocket.WebSocketHandler):
+class ZMQStreamHandler(WebSocketHandler):
     
     def check_origin(self, origin):
         """Check Origin == Host or Access-Control-Allow-Origin.
         
         Tornado >= 4 calls this method automatically, raising 403 if it returns False.
         We call it explicitly in `open` on Tornado < 4.
         """
         if self.allow_origin == '*':
             return True
 
         host = self.request.headers.get("Host")
 
         # If no header is provided, assume we can't verify origin
         if origin is None:
             self.log.warn("Missing Origin header, rejecting WebSocket connection.")
             return False
         if host is None:
             self.log.warn("Missing Host header, rejecting WebSocket connection.")
             return False
         
         origin = origin.lower()
         origin_host = urlparse(origin).netloc
         
         # OK if origin matches host
         if origin_host == host:
             return True
         
         # Check CORS headers
         if self.allow_origin:
             allow = self.allow_origin == origin
         elif self.allow_origin_pat:
             allow = bool(self.allow_origin_pat.match(origin))
         else:
             # No CORS headers deny the request
             allow = False
         if not allow:
             self.log.warn("Blocking Cross Origin WebSocket Attempt.  Origin: %s, Host: %s",
                 origin, host,
             )
         return allow
 
     def clear_cookie(self, *args, **kwargs):
         """meaningless for websockets"""
         pass
 
     def _reserialize_reply(self, msg_list):
         """Reserialize a reply message using JSON.
 
         This takes the msg list from the ZMQ socket, deserializes it using
         self.session and then serializes the result using JSON. This method
         should be used by self._on_zmq_reply to build messages that can
         be sent back to the browser.
         """
         idents, msg_list = self.session.feed_identities(msg_list)
         msg = self.session.deserialize(msg_list)
         if msg['buffers']:
             buf = serialize_binary_message(msg)
             return buf
         else:
             smsg = json.dumps(msg, default=date_default)
             return cast_unicode(smsg)
 
     def _on_zmq_reply(self, msg_list):
         # Sometimes this gets triggered when the on_close method is scheduled in the
         # eventloop but hasn't been called.
         if self.stream.closed(): return
         try:
             msg = self._reserialize_reply(msg_list)
         except Exception:
             self.log.critical("Malformed message: %r" % msg_list, exc_info=True)
         else:
             self.write_message(msg, binary=isinstance(msg, bytes))
 
-    def allow_draft76(self):
-        """Allow draft 76, until browsers such as Safari update to RFC 6455.
-        
-        This has been disabled by default in tornado in release 2.2.0, and
-        support will be removed in later versions.
-        """
-        return True
-
-# ping interval for keeping websockets alive (30 seconds)
-WS_PING_INTERVAL = 30000
-
 class AuthenticatedZMQStreamHandler(ZMQStreamHandler, IPythonHandler):
     ping_callback = None
     last_ping = 0
     last_pong = 0
     
     @property
     def ping_interval(self):
         """The interval for websocket keep-alive pings.
         
         Set ws_ping_interval = 0 to disable pings.
         """
         return self.settings.get('ws_ping_interval', WS_PING_INTERVAL)
     
     @property
     def ping_timeout(self):
         """If no ping is received in this many milliseconds,
         close the websocket connection (VPNs, etc. can fail to cleanly close ws connections).
         Default is max of 3 pings or 30 seconds.
         """
         return self.settings.get('ws_ping_timeout',
             max(3 * self.ping_interval, WS_PING_INTERVAL)
         )
 
     def set_default_headers(self):
         """Undo the set_default_headers in IPythonHandler
         
         which doesn't make sense for websockets
         """
         pass
     
     def pre_get(self):
         """Run before finishing the GET request
         
         Extend this method to add logic that should fire before
         the websocket finishes completing.
         """
         # Check to see that origin matches host directly, including ports
         # Tornado 4 already does CORS checking
         if tornado.version_info[0] < 4:
             if not self.check_origin(self.get_origin()):
                 raise web.HTTPError(403)
         
         # authenticate the request before opening the websocket
         if self.get_current_user() is None:
             self.log.warn("Couldn't authenticate WebSocket connection")
             raise web.HTTPError(403)
         
         if self.get_argument('session_id', False):
             self.session.session = cast_unicode(self.get_argument('session_id'))
         else:
             self.log.warn("No session ID specified")
     
     @gen.coroutine
     def get(self, *args, **kwargs):
         # pre_get can be a coroutine in subclasses
         # assign and yield in two step to avoid tornado 3 issues
         res = self.pre_get()
         yield gen.maybe_future(res)
         # FIXME: only do super get on tornado ≥ 4
         # tornado 3 has no get, will raise 405
         if tornado.version_info >= (4,):
             super(AuthenticatedZMQStreamHandler, self).get(*args, **kwargs)
     
     def initialize(self):
         self.log.debug("Initializing websocket connection %s", self.request.path)
         self.session = Session(config=self.config)
     
     def open(self, *args, **kwargs):
         self.log.debug("Opening websocket %s", self.request.path)
         if tornado.version_info < (4,):
             try:
                 self.get(*self.open_args, **self.open_kwargs)
             except web.HTTPError:
                 self.close()
                 raise
         
         # start the pinging
         if self.ping_interval > 0:
             self.last_ping = ioloop.IOLoop.instance().time()  # Remember time of last ping
             self.last_pong = self.last_ping
             self.ping_callback = ioloop.PeriodicCallback(self.send_ping, self.ping_interval)
             self.ping_callback.start()
 
     def send_ping(self):
         """send a ping to keep the websocket alive"""
         if self.stream.closed() and self.ping_callback is not None:
             self.ping_callback.stop()
             return
         
         # check for timeout on pong.  Make sure that we really have sent a recent ping in
         # case the machine with both server and client has been suspended since the last ping.
         now = ioloop.IOLoop.instance().time()
         since_last_pong = 1e3 * (now - self.last_pong)
         since_last_ping = 1e3 * (now - self.last_ping)
         if since_last_ping < 2*self.ping_interval and since_last_pong > self.ping_timeout:
             self.log.warn("WebSocket ping timeout after %i ms.", since_last_pong)
             self.close()
             return
 
         self.ping(b'')
         self.last_ping = now
 
     def on_pong(self, data):
         self.last_pong = ioloop.IOLoop.instance().time()

IPython/testing/iptestcontroller.py

@@ -1,705 +1,713
 # -*- coding: utf-8 -*-
 """IPython Test Process Controller
 
 This module runs one or more subprocesses which will actually run the IPython
 test suite.
 
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from __future__ import print_function
 
 import argparse
 import json
 import multiprocessing.pool
 import os
 import re
 import requests
 import shutil
 import signal
 import sys
 import subprocess
 import time
 
 from .iptest import (
     have, test_group_names as py_test_group_names, test_sections, StreamCapturer,
     test_for,
 )
 from IPython.utils.path import compress_user
 from IPython.utils.py3compat import bytes_to_str
 from IPython.utils.sysinfo import get_sys_info
 from IPython.utils.tempdir import TemporaryDirectory
 from IPython.utils.text import strip_ansi
 
 try:
     # Python >= 3.3
     from subprocess import TimeoutExpired
     def popen_wait(p, timeout):
         return p.wait(timeout)
 except ImportError:
     class TimeoutExpired(Exception):
         pass
     def popen_wait(p, timeout):
         """backport of Popen.wait from Python 3"""
         for i in range(int(10 * timeout)):
             if p.poll() is not None:
                 return
             time.sleep(0.1)
         if p.poll() is None:
             raise TimeoutExpired
 
 NOTEBOOK_SHUTDOWN_TIMEOUT = 10
 
 class TestController(object):
     """Run tests in a subprocess
     """
     #: str, IPython test suite to be executed.
     section = None
     #: list, command line arguments to be executed
     cmd = None
     #: dict, extra environment variables to set for the subprocess
     env = None
     #: list, TemporaryDirectory instances to clear up when the process finishes
     dirs = None
     #: subprocess.Popen instance
     process = None
     #: str, process stdout+stderr
     stdout = None
 
     def __init__(self):
         self.cmd = []
         self.env = {}
         self.dirs = []
 
     def setup(self):
         """Create temporary directories etc.
         
         This is only called when we know the test group will be run. Things
         created here may be cleaned up by self.cleanup().
         """
         pass
 
     def launch(self, buffer_output=False, capture_output=False):
         # print('*** ENV:', self.env)  # dbg
         # print('*** CMD:', self.cmd)  # dbg
         env = os.environ.copy()
         env.update(self.env)
         if buffer_output:
             capture_output = True
         self.stdout_capturer = c = StreamCapturer(echo=not buffer_output)
         c.start()
         stdout = c.writefd if capture_output else None
         stderr = subprocess.STDOUT if capture_output else None
         self.process = subprocess.Popen(self.cmd, stdout=stdout,
                 stderr=stderr, env=env)
 
     def wait(self):
         self.process.wait()
         self.stdout_capturer.halt()
         self.stdout = self.stdout_capturer.get_buffer()
         return self.process.returncode
 
     def print_extra_info(self):
         """Print extra information about this test run.
         
         If we're running in parallel and showing the concise view, this is only
         called if the test group fails. Otherwise, it's called before the test
         group is started.
         
         The base implementation does nothing, but it can be overridden by
         subclasses.
         """
         return
 
     def cleanup_process(self):
         """Cleanup on exit by killing any leftover processes."""
         subp = self.process
         if subp is None or (subp.poll() is not None):
             return  # Process doesn't exist, or is already dead.
 
         try:
             print('Cleaning up stale PID: %d' % subp.pid)
             subp.kill()
         except: # (OSError, WindowsError) ?
             # This is just a best effort, if we fail or the process was
             # really gone, ignore it.
             pass
         else:
             for i in range(10):
                 if subp.poll() is None:
                     time.sleep(0.1)
                 else:
                     break
 
         if subp.poll() is None:
             # The process did not die...
             print('... failed. Manual cleanup may be required.')
 
     def cleanup(self):
         "Kill process if it's still alive, and clean up temporary directories"
         self.cleanup_process()
         for td in self.dirs:
             td.cleanup()
 
     __del__ = cleanup
 
 
 class PyTestController(TestController):
     """Run Python tests using IPython.testing.iptest"""
     #: str, Python command to execute in subprocess
     pycmd = None
 
     def __init__(self, section, options):
         """Create new test runner."""
         TestController.__init__(self)
         self.section = section
         # pycmd is put into cmd[2] in PyTestController.launch()
         self.cmd = [sys.executable, '-c', None, section]
         self.pycmd = "from IPython.testing.iptest import run_iptest; run_iptest()"
         self.options = options
 
     def setup(self):
         ipydir = TemporaryDirectory()
         self.dirs.append(ipydir)
         self.env['IPYTHONDIR'] = ipydir.name
         self.workingdir = workingdir = TemporaryDirectory()
         self.dirs.append(workingdir)
         self.env['IPTEST_WORKING_DIR'] = workingdir.name
         # This means we won't get odd effects from our own matplotlib config
         self.env['MPLCONFIGDIR'] = workingdir.name
 
         # From options:
         if self.options.xunit:
             self.add_xunit()
         if self.options.coverage:
             self.add_coverage()
         self.env['IPTEST_SUBPROC_STREAMS'] = self.options.subproc_streams
         self.cmd.extend(self.options.extra_args)
 
     @property
     def will_run(self):
         try:
             return test_sections[self.section].will_run
         except KeyError:
             return True
 
     def add_xunit(self):
         xunit_file = os.path.abspath(self.section + '.xunit.xml')
         self.cmd.extend(['--with-xunit', '--xunit-file', xunit_file])
 
     def add_coverage(self):
         try:
             sources = test_sections[self.section].includes
         except KeyError:
             sources = ['IPython']
 
         coverage_rc = ("[run]\n"
                        "data_file = {data_file}\n"
                        "source =\n"
                        "  {source}\n"
                       ).format(data_file=os.path.abspath('.coverage.'+self.section),
                                source="\n  ".join(sources))
         config_file = os.path.join(self.workingdir.name, '.coveragerc')
         with open(config_file, 'w') as f:
             f.write(coverage_rc)
 
         self.env['COVERAGE_PROCESS_START'] = config_file
         self.pycmd = "import coverage; coverage.process_startup(); " + self.pycmd
 
     def launch(self, buffer_output=False):
         self.cmd[2] = self.pycmd
         super(PyTestController, self).launch(buffer_output=buffer_output)
 
 
 js_prefix = 'js/'
 
 def get_js_test_dir():
     import IPython.html.tests as t
     return os.path.join(os.path.dirname(t.__file__), '')
 
 def all_js_groups():
     import glob
     test_dir = get_js_test_dir()
     all_subdirs = glob.glob(test_dir + '[!_]*/')
     return [js_prefix+os.path.relpath(x, test_dir) for x in all_subdirs]
 
 class JSController(TestController):
     """Run CasperJS tests """
     
     requirements =  ['zmq', 'tornado', 'jinja2', 'casperjs', 'sqlite3',
                      'jsonschema']
 
     def __init__(self, section, xunit=True, engine='phantomjs', url=None):
         """Create new test runner."""
         TestController.__init__(self)
         self.engine = engine
         self.section = section
         self.xunit = xunit
         self.url = url
         self.slimer_failure = re.compile('^FAIL.*', flags=re.MULTILINE)
         js_test_dir = get_js_test_dir()
         includes = '--includes=' + os.path.join(js_test_dir,'util.js')
         test_cases = os.path.join(js_test_dir, self.section[len(js_prefix):])
         self.cmd = ['casperjs', 'test', includes, test_cases, '--engine=%s' % self.engine]
 
     def setup(self):
         self.ipydir = TemporaryDirectory()
         self.nbdir = TemporaryDirectory()
         self.dirs.append(self.ipydir)
         self.dirs.append(self.nbdir)
         os.makedirs(os.path.join(self.nbdir.name, os.path.join(u'sub ∂ir1', u'sub ∂ir 1a')))
         os.makedirs(os.path.join(self.nbdir.name, os.path.join(u'sub ∂ir2', u'sub ∂ir 1b')))
 
         if self.xunit:
             self.add_xunit()
 
         # If a url was specified, use that for the testing.
         if self.url:
             try:
                 alive = requests.get(self.url).status_code == 200
             except:
                 alive = False
 
             if alive:
                 self.cmd.append("--url=%s" % self.url)
             else:
                 raise Exception('Could not reach "%s".' % self.url)
         else:
             # start the ipython notebook, so we get the port number
             self.server_port = 0
             self._init_server()
             if self.server_port:
                 self.cmd.append("--port=%i" % self.server_port)
             else:
                 # don't launch tests if the server didn't start
                 self.cmd = [sys.executable, '-c', 'raise SystemExit(1)']
 
     def add_xunit(self):
         xunit_file = os.path.abspath(self.section.replace('/','.') + '.xunit.xml')
         self.cmd.append('--xunit=%s' % xunit_file)
 
     def launch(self, buffer_output):
         # If the engine is SlimerJS, we need to buffer the output because
         # SlimerJS does not support exit codes, so CasperJS always returns 0.
         if self.engine == 'slimerjs' and not buffer_output:
             return super(JSController, self).launch(capture_output=True)
 
         else:
             return super(JSController, self).launch(buffer_output=buffer_output)
 
     def wait(self, *pargs, **kwargs):
         """Wait for the JSController to finish"""
         ret = super(JSController, self).wait(*pargs, **kwargs)
         # If this is a SlimerJS controller, check the captured stdout for
         # errors.  Otherwise, just return the return code.
         if self.engine == 'slimerjs':
             stdout = bytes_to_str(self.stdout)
             if ret != 0:
                 # This could still happen e.g. if it's stopped by SIGINT
                 return ret
             return bool(self.slimer_failure.search(strip_ansi(stdout)))
         else:
             return ret
 
     def print_extra_info(self):
         print("Running tests with notebook directory %r" % self.nbdir.name)
 
     @property
     def will_run(self):
         should_run = all(have[a] for a in self.requirements + [self.engine])
         return should_run
 
     def _init_server(self):
         "Start the notebook server in a separate process"
         self.server_command = command = [sys.executable,
             '-m', 'IPython.html',
             '--no-browser',
             '--ipython-dir', self.ipydir.name,
             '--notebook-dir', self.nbdir.name,
         ]
         # ipc doesn't work on Windows, and darwin has crazy-long temp paths,
         # which run afoul of ipc's maximum path length.
         if sys.platform.startswith('linux'):
             command.append('--KernelManager.transport=ipc')
         self.stream_capturer = c = StreamCapturer()
         c.start()
-        self.server = subprocess.Popen(command, stdout=c.writefd, stderr=subprocess.STDOUT, cwd=self.nbdir.name)
+        env = os.environ.copy()
+        if self.engine == 'phantomjs':
+            env['IPYTHON_ALLOW_DRAFT_WEBSOCKETS_FOR_PHANTOMJS'] = '1'
+        self.server = subprocess.Popen(command,
+            stdout=c.writefd,
+            stderr=subprocess.STDOUT,
+            cwd=self.nbdir.name,
+            env=env,
+        )
         self.server_info_file = os.path.join(self.ipydir.name,
             'profile_default', 'security', 'nbserver-%i.json' % self.server.pid
         )
         self._wait_for_server()
     
     def _wait_for_server(self):
         """Wait 30 seconds for the notebook server to start"""
         for i in range(300):
             if self.server.poll() is not None:
                 return self._failed_to_start()
             if os.path.exists(self.server_info_file):
                 try:
                     self._load_server_info()
                 except ValueError:
                     # If the server is halfway through writing the file, we may
                     # get invalid JSON; it should be ready next iteration.
                     pass
                 else:
                     return
             time.sleep(0.1)
         print("Notebook server-info file never arrived: %s" % self.server_info_file,
             file=sys.stderr
         )
     
     def _failed_to_start(self):
         """Notebook server exited prematurely"""
         captured = self.stream_capturer.get_buffer().decode('utf-8', 'replace')
         print("Notebook failed to start: ", file=sys.stderr)
         print(self.server_command)
         print(captured, file=sys.stderr)
     
     def _load_server_info(self):
         """Notebook server started, load connection info from JSON"""
         with open(self.server_info_file) as f:
             info = json.load(f)
         self.server_port = info['port']
 
     def cleanup(self):
         try:
             self.server.terminate()
         except OSError:
             # already dead
             pass
         # wait 10s for the server to shutdown
         try:
             popen_wait(self.server, NOTEBOOK_SHUTDOWN_TIMEOUT)
         except TimeoutExpired:
             # server didn't terminate, kill it
             try:
                 print("Failed to terminate notebook server, killing it.",
                     file=sys.stderr
                 )
                 self.server.kill()
             except OSError:
                 # already dead
                 pass
         # wait another 10s
         try:
             popen_wait(self.server, NOTEBOOK_SHUTDOWN_TIMEOUT)
         except TimeoutExpired:
             print("Notebook server still running (%s)" % self.server_info_file,
                 file=sys.stderr
             )
             
         self.stream_capturer.halt()
         TestController.cleanup(self)
 
 
 def prepare_controllers(options):
     """Returns two lists of TestController instances, those to run, and those
     not to run."""
     testgroups = options.testgroups
     if testgroups:
         if 'js' in testgroups:
             js_testgroups = all_js_groups()
         else:
             js_testgroups = [g for g in testgroups if g.startswith(js_prefix)]
         
         py_testgroups = [g for g in testgroups if g not in ['js'] + js_testgroups]
     else:
         py_testgroups = py_test_group_names
         if not options.all:
             js_testgroups = []
             test_sections['parallel'].enabled = False
         else:
             js_testgroups = all_js_groups()
 
     engine = 'slimerjs' if options.slimerjs else 'phantomjs'
     c_js = [JSController(name, xunit=options.xunit, engine=engine, url=options.url) for name in js_testgroups]
     c_py = [PyTestController(name, options) for name in py_testgroups]
 
     controllers = c_py + c_js
     to_run = [c for c in controllers if c.will_run]
     not_run = [c for c in controllers if not c.will_run]
     return to_run, not_run
 
 def do_run(controller, buffer_output=True):
     """Setup and run a test controller.
     
     If buffer_output is True, no output is displayed, to avoid it appearing
     interleaved. In this case, the caller is responsible for displaying test
     output on failure.
     
     Returns
     -------
     controller : TestController
       The same controller as passed in, as a convenience for using map() type
       APIs.
     exitcode : int
       The exit code of the test subprocess. Non-zero indicates failure.
     """
     try:
         try:
             controller.setup()
             if not buffer_output:
                 controller.print_extra_info()
             controller.launch(buffer_output=buffer_output)
         except Exception:
             import traceback
             traceback.print_exc()
             return controller, 1  # signal failure
 
         exitcode = controller.wait()
         return controller, exitcode
 
     except KeyboardInterrupt:
         return controller, -signal.SIGINT
     finally:
         controller.cleanup()
 
 def report():
     """Return a string with a summary report of test-related variables."""
     inf = get_sys_info()
     out = []
     def _add(name, value):
         out.append((name, value))
 
     _add('IPython version', inf['ipython_version'])
     _add('IPython commit', "{} ({})".format(inf['commit_hash'], inf['commit_source']))
     _add('IPython package', compress_user(inf['ipython_path']))
     _add('Python version', inf['sys_version'].replace('\n',''))
     _add('sys.executable', compress_user(inf['sys_executable']))
     _add('Platform', inf['platform'])
 
     width = max(len(n) for (n,v) in out)
     out = ["{:<{width}}: {}\n".format(n, v, width=width) for (n,v) in out]
 
     avail = []
     not_avail = []
 
     for k, is_avail in have.items():
         if is_avail:
             avail.append(k)
         else:
             not_avail.append(k)
 
     if avail:
         out.append('\nTools and libraries available at test time:\n')
         avail.sort()
         out.append('   ' + ' '.join(avail)+'\n')
 
     if not_avail:
         out.append('\nTools and libraries NOT available at test time:\n')
         not_avail.sort()
         out.append('   ' + ' '.join(not_avail)+'\n')
 
     return ''.join(out)
 
 def run_iptestall(options):
     """Run the entire IPython test suite by calling nose and trial.
 
     This function constructs :class:`IPTester` instances for all IPython
     modules and package and then runs each of them.  This causes the modules
     and packages of IPython to be tested each in their own subprocess using
     nose.
 
     Parameters
     ----------
 
     All parameters are passed as attributes of the options object.
 
     testgroups : list of str
       Run only these sections of the test suite. If empty, run all the available
       sections.
 
     fast : int or None
       Run the test suite in parallel, using n simultaneous processes. If None
       is passed, one process is used per CPU core. Default 1 (i.e. sequential)
 
     inc_slow : bool
       Include slow tests, like IPython.parallel. By default, these tests aren't
       run.
 
     slimerjs : bool
       Use slimerjs if it's installed instead of phantomjs for casperjs tests.
 
     url : unicode
       Address:port to use when running the JS tests.
 
     xunit : bool
       Produce Xunit XML output. This is written to multiple foo.xunit.xml files.
 
     coverage : bool or str
       Measure code coverage from tests. True will store the raw coverage data,
       or pass 'html' or 'xml' to get reports.
 
     extra_args : list
       Extra arguments to pass to the test subprocesses, e.g. '-v'
     """
     to_run, not_run = prepare_controllers(options)
 
     def justify(ltext, rtext, width=70, fill='-'):
         ltext += ' '
         rtext = (' ' + rtext).rjust(width - len(ltext), fill)
         return ltext + rtext
 
     # Run all test runners, tracking execution time
     failed = []
     t_start = time.time()
 
     print()
     if options.fast == 1:
         # This actually means sequential, i.e. with 1 job
         for controller in to_run:
             print('Test group:', controller.section)
             sys.stdout.flush()  # Show in correct order when output is piped
             controller, res = do_run(controller, buffer_output=False)
             if res:
                 failed.append(controller)
                 if res == -signal.SIGINT:
                     print("Interrupted")
                     break
             print()
 
     else:
         # Run tests concurrently
         try:
             pool = multiprocessing.pool.ThreadPool(options.fast)
             for (controller, res) in pool.imap_unordered(do_run, to_run):
                 res_string = 'OK' if res == 0 else 'FAILED'
                 print(justify('Test group: ' + controller.section, res_string))
                 if res:
                     controller.print_extra_info()
                     print(bytes_to_str(controller.stdout))
                     failed.append(controller)
                     if res == -signal.SIGINT:
                         print("Interrupted")
                         break
         except KeyboardInterrupt:
             return
 
     for controller in not_run:
         print(justify('Test group: ' + controller.section, 'NOT RUN'))
 
     t_end = time.time()
     t_tests = t_end - t_start
     nrunners = len(to_run)
     nfail = len(failed)
     # summarize results
     print('_'*70)
     print('Test suite completed for system with the following information:')
     print(report())
     took = "Took %.3fs." % t_tests
     print('Status: ', end='')
     if not failed:
         print('OK (%d test groups).' % nrunners, took)
     else:
         # If anything went wrong, point out what command to rerun manually to
         # see the actual errors and individual summary
         failed_sections = [c.section for c in failed]
         print('ERROR - {} out of {} test groups failed ({}).'.format(nfail,
                                   nrunners, ', '.join(failed_sections)), took)
         print()
         print('You may wish to rerun these, with:')
         print('  iptest', *failed_sections)
         print()
 
     if options.coverage:
         from coverage import coverage
         cov = coverage(data_file='.coverage')
         cov.combine()
         cov.save()
 
         # Coverage HTML report
         if options.coverage == 'html':
             html_dir = 'ipy_htmlcov'
             shutil.rmtree(html_dir, ignore_errors=True)
             print("Writing HTML coverage report to %s/ ... " % html_dir, end="")
             sys.stdout.flush()
 
             # Custom HTML reporter to clean up module names.
             from coverage.html import HtmlReporter
             class CustomHtmlReporter(HtmlReporter):
                 def find_code_units(self, morfs):
                     super(CustomHtmlReporter, self).find_code_units(morfs)
                     for cu in self.code_units:
                         nameparts = cu.name.split(os.sep)
                         if 'IPython' not in nameparts:
                             continue
                         ix = nameparts.index('IPython')
                         cu.name = '.'.join(nameparts[ix:])
 
             # Reimplement the html_report method with our custom reporter
             cov._harvest_data()
             cov.config.from_args(omit='*{0}tests{0}*'.format(os.sep), html_dir=html_dir,
                                  html_title='IPython test coverage',
                                 )
             reporter = CustomHtmlReporter(cov, cov.config)
             reporter.report(None)
             print('done.')
 
         # Coverage XML report
         elif options.coverage == 'xml':
             cov.xml_report(outfile='ipy_coverage.xml')
 
     if failed:
         # Ensure that our exit code indicates failure
         sys.exit(1)
 
 argparser = argparse.ArgumentParser(description='Run IPython test suite')
 argparser.add_argument('testgroups', nargs='*',
                     help='Run specified groups of tests. If omitted, run '
                     'all tests.')
 argparser.add_argument('--all', action='store_true',
                     help='Include slow tests not run by default.')
 argparser.add_argument('--slimerjs', action='store_true',
                     help="Use slimerjs if it's installed instead of phantomjs for casperjs tests.")
 argparser.add_argument('--url', help="URL to use for the JS tests.")
 argparser.add_argument('-j', '--fast', nargs='?', const=None, default=1, type=int,
                     help='Run test sections in parallel. This starts as many '
                     'processes as you have cores, or you can specify a number.')
 argparser.add_argument('--xunit', action='store_true',
                     help='Produce Xunit XML results')
 argparser.add_argument('--coverage', nargs='?', const=True, default=False,
                     help="Measure test coverage. Specify 'html' or "
                     "'xml' to get reports.")
 argparser.add_argument('--subproc-streams', default='capture',
                     help="What to do with stdout/stderr from subprocesses. "
                     "'capture' (default), 'show' and 'discard' are the options.")
 
 def default_options():
     """Get an argparse Namespace object with the default arguments, to pass to
     :func:`run_iptestall`.
     """
     options = argparser.parse_args([])
     options.extra_args = []
     return options
 
 def main():
     # iptest doesn't work correctly if the working directory is the
     # root of the IPython source tree. Tell the user to avoid
     # frustration.
     if os.path.exists(os.path.join(os.getcwd(),
                                    'IPython', 'testing', '__main__.py')):
         print("Don't run iptest from the IPython source directory",
               file=sys.stderr)
         sys.exit(1)
     # Arguments after -- should be passed through to nose. Argparse treats
     # everything after -- as regular positional arguments, so we separate them
     # first.
     try:
         ix = sys.argv.index('--')
     except ValueError:
         to_parse = sys.argv[1:]
         extra_args = []
     else:
         to_parse = sys.argv[1:ix]
         extra_args = sys.argv[ix+1:]
 
     options = argparser.parse_args(to_parse)
     options.extra_args = extra_args
 
     run_iptestall(options)
 
 
 if __name__ == '__main__':
     main()