Repository: tornadoweb/tornado
Pull request #2307 (2018-03-12 01:48:54)

concurrent,httputil,websocket: small bug fixes


Patch:
diff --git a/tornado/concurrent.py b/tornado/concurrent.py
index 1a79f1e1..aa38bb99 100644
--- a/tornado/concurrent.py
+++ b/tornado/concurrent.py
@@ -421,6 +421,10 @@ def run_on_executor(*args, **kwargs):
 
     .. versionchanged:: 5.0
        Always uses the current IOLoop instead of ``self.io_loop``.
+
+    .. versionchanged:: 5.1
+       Returns a `.Future` compatible with ``await`` instead of a
+       `concurrent.futures.Future`.
     """
     def run_on_executor_decorator(fn):
         executor = kwargs.get("executor", "executor")
@@ -428,12 +432,14 @@ def run_on_executor(*args, **kwargs):
         @functools.wraps(fn)
         def wrapper(self, *args, **kwargs):
             callback = kwargs.pop("callback", None)
-            future = getattr(self, executor).submit(fn, self, *args, **kwargs)
+            async_future = Future()
+            conc_future = getattr(self, executor).submit(fn, self, *args, **kwargs)
+            chain_future(conc_future, async_future)
             if callback:
                 from tornado.ioloop import IOLoop
                 IOLoop.current().add_future(
-                    future, lambda future: callback(future.result()))
-            return future
+                    async_future, lambda future: callback(future.result()))
+            return async_future
         return wrapper
     if args and kwargs:
         raise ValueError("cannot combine positional and keyword args")
diff --git a/tornado/gen.py b/tornado/gen.py
index 0ef7c9fc..762b44f9 100644
--- a/tornado/gen.py
+++ b/tornado/gen.py
@@ -1292,12 +1292,7 @@ except ImportError:
                         break
         raise Return(_r)
 else:
-    try:
-        _wrap_awaitable = asyncio.ensure_future
-    except AttributeError:
-        # asyncio.ensure_future was introduced in Python 3.4.4, but
-        # Debian jessie still ships with 3.4.2 so try the old name.
-        _wrap_awaitable = getattr(asyncio, 'async')
+    _wrap_awaitable = asyncio.ensure_future
 
 
 def convert_yielded(yielded):
diff --git a/tornado/http1connection.py b/tornado/http1connection.py
index ed906223..0349e171 100644
--- a/tornado/http1connection.py
+++ b/tornado/http1connection.py
@@ -519,12 +519,7 @@ class HTTP1Connection(httputil.HTTPConnection):
         # RFC 7230 section allows for both CRLF and bare LF.
         eol = data.find("\n")
         start_line = data[:eol].rstrip("\r")
-        try:
-            headers = httputil.HTTPHeaders.parse(data[eol:])
-        except ValueError:
-            # probably form split() if there was no ':' in the line
-            raise httputil.HTTPInputError("Malformed HTTP headers: %r" %
-                                          data[eol:100])
+        headers = httputil.HTTPHeaders.parse(data[eol:])
         return start_line, headers
 
     def _read_body(self, code, headers, delegate):
diff --git a/tornado/httputil.py b/tornado/httputil.py
index ceff7350..3d2d3359 100644
--- a/tornado/httputil.py
+++ b/tornado/httputil.py
@@ -183,11 +183,16 @@ class HTTPHeaders(collections.MutableMapping):
         """
         if line[0].isspace():
             # continuation of a multi-line header
+            if self._last_key is None:
+                raise HTTPInputError("first header line cannot start with whitespace")
             new_part = ' ' + line.lstrip()
             self._as_list[self._last_key][-1] += new_part
             self._dict[self._last_key] += new_part
         else:
-            name, value = line.split(":", 1)
+            try:
+                name, value = line.split(":", 1)
+            except ValueError:
+                raise HTTPInputError("no colon in header line")
             self.add(name, value.strip())
 
     @classmethod
@@ -197,6 +202,12 @@ class HTTPHeaders(collections.MutableMapping):
         >>> h = HTTPHeaders.parse("Content-Type: text/html\\r\\nContent-Length: 42\\r\\n")
         >>> sorted(h.items())
         [('Content-Length', '42'), ('Content-Type', 'text/html')]
+
+        .. versionchanged:: 5.1
+
+           Raises `HTTPInputError` on malformed headers instead of a
+           mix of `KeyError`, and `ValueError`.
+
         """
         h = cls()
         for line in _CRLF_RE.split(headers):
diff --git a/tornado/ioloop.py b/tornado/ioloop.py
index 839e7ee5..b3fc3567 100644
--- a/tornado/ioloop.py
+++ b/tornado/ioloop.py
@@ -281,7 +281,7 @@ class IOLoop(Configurable):
         else:
             try:
                 loop = asyncio.get_event_loop()
-            except (RuntimeError, AssertionError):
+            except RuntimeError:
                 if not instance:
                     return None
                 raise
diff --git a/tornado/platform/asyncio.py b/tornado/platform/asyncio.py
index b2ad9fe6..0d7eea3e 100644
--- a/tornado/platform/asyncio.py
+++ b/tornado/platform/asyncio.py
@@ -104,7 +104,7 @@ class BaseAsyncIOLoop(IOLoop):
     def start(self):
         try:
             old_loop = asyncio.get_event_loop()
-        except (RuntimeError, AssertionError):
+        except RuntimeError:
             old_loop = None
         try:
             self._setup_logging()
@@ -211,7 +211,7 @@ class AsyncIOLoop(BaseAsyncIOLoop):
         if not self.is_current:
             try:
                 self.old_asyncio = asyncio.get_event_loop()
-            except (RuntimeError, AssertionError):
+            except RuntimeError:
                 self.old_asyncio = None
             self.is_current = True
         asyncio.set_event_loop(self.asyncio_loop)
@@ -270,9 +270,7 @@ class AnyThreadEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
     def get_event_loop(self):
         try:
             return super().get_event_loop()
-        except (RuntimeError, AssertionError):
-            # This was an AssertionError in python 3.4.2 (which ships with debian jessie)
-            # and changed to a RuntimeError in 3.4.3.
+        except RuntimeError:
             # "There is no current event loop in thread %r"
             loop = self.new_event_loop()
             self.set_event_loop(loop)
diff --git a/tornado/websocket.py b/tornado/websocket.py
index 1209bfbf..629e9417 100644
--- a/tornado/websocket.py
+++ b/tornado/websocket.py
@@ -312,8 +312,23 @@ class WebSocketHandler(tornado.web.RequestHandler):
         """
         raise NotImplementedError
 
-    def ping(self, data):
-        """Send ping frame to the remote end."""
+    def ping(self, data=b''):
+        """Send ping frame to the remote end.
+
+        The data argument allows a small amount of data (up to 125
+        bytes) to be sent as a part of the ping message. Note that not
+        all websocket implementations expose this data to
+        applications.
+
+        Consider using the ``websocket_ping_interval`` application
+        setting instead of sending pings manually.
+
+        .. versionchanged:: 5.1
+
+           The data argument is now optional.
+
+        """
+        data = utf8(data)
         if self.ws_connection is None:
             raise WebSocketClosedError()
         self.ws_connection.write_ping(data)
@@ -661,7 +676,8 @@ class WebSocketProtocol13(WebSocketProtocol):
             self.request.headers.get("Sec-Websocket-Key"))
 
     def _accept_connection(self):
-        subprotocols = [s.strip() for s in self.request.headers.get_list("Sec-WebSocket-Protocol")]
+        subprotocols = self.request.headers.get("Sec-WebSocket-Protocol", '')
+        subprotocols = [s.strip() for s in subprotocols.split(',')]
         if subprotocols:
             selected = self.handler.select_subprotocol(subprotocols)
             if selected:
@@ -755,12 +771,19 @@ class WebSocketProtocol13(WebSocketProtocol):
             **self._get_compressor_options(other_side, agreed_parameters, compression_options))
 
     def _write_frame(self, fin, opcode, data, flags=0):
+        data_len = len(data)
+        if opcode & 0x8:
+            # All control frames MUST have a payload length of 125
+            # bytes or less and MUST NOT be fragmented.
+            if not fin:
+                raise ValueError("control frames may not be fragmented")
+            if data_len > 125:
+                raise ValueError("control frame payloads may not exceed 125 bytes")
         if fin:
             finbit = self.FIN
         else:
             finbit = 0
         frame = struct.pack("B", finbit | opcode | flags)
-        data_len = len(data)
         if self.mask_outgoing:
             mask_bit = 0x80
         else:
@@ -1204,6 +1227,25 @@ class WebSocketClientConnection(simple_httpclient._HTTPConnection):
         else:
             self.read_queue.append(message)
 
+    def ping(self, data=b''):
+        """Send ping frame to the remote end.
+
+        The data argument allows a small amount of data (up to 125
+        bytes) to be sent as a part of the ping message. Note that not
+        all websocket implementations expose this data to
+        applications.
+
+        Consider using the ``ping_interval`` argument to
+        `websocket_connect` instead of sending pings manually.
+
+        .. versionadded:: 5.1
+
+        """
+        data = utf8(data)
+        if self.protocol is None:
+            raise WebSocketClosedError()
+        self.protocol.write_ping(data)
+
     def on_pong(self, data):
         pass
 


Test patch:
diff --git a/tornado/test/asyncio_test.py b/tornado/test/asyncio_test.py
index 41fda20d..bbabb521 100644
--- a/tornado/test/asyncio_test.py
+++ b/tornado/test/asyncio_test.py
@@ -150,7 +150,7 @@ class AnyThreadEventLoopPolicyTest(unittest.TestCase):
     def run_policy_test(self, accessor, expected_type):
         # With the default policy, non-main threads don't get an event
         # loop.
-        self.assertRaises((RuntimeError, AssertionError),
+        self.assertRaises(RuntimeError,
                           self.executor.submit(accessor).result)
         # Set the policy and we can get a loop.
         asyncio.set_event_loop_policy(AnyThreadEventLoopPolicy())
diff --git a/tornado/test/concurrent_test.py b/tornado/test/concurrent_test.py
index 1b6f8ee9..b00b80dd 100644
--- a/tornado/test/concurrent_test.py
+++ b/tornado/test/concurrent_test.py
@@ -31,7 +31,7 @@ from tornado.log import app_log
 from tornado import stack_context
 from tornado.tcpserver import TCPServer
 from tornado.testing import AsyncTestCase, ExpectLog, bind_unused_port, gen_test
-from tornado.test.util import unittest
+from tornado.test.util import unittest, skipBefore35, exec_test
 
 
 try:
@@ -429,6 +429,26 @@ class RunOnExecutorTest(AsyncTestCase):
         answer = yield o.f()
         self.assertEqual(answer, 42)
 
+    @skipBefore35
+    @gen_test
+    def test_async_await(self):
+        class Object(object):
+            def __init__(self):
+                self.executor = futures.thread.ThreadPoolExecutor(1)
+
+            @run_on_executor()
+            def f(self):
+                return 42
+
+        o = Object()
+        namespace = exec_test(globals(), locals(), """
+        async def f():
+            answer = await o.f()
+            return answer
+        """)
+        result = yield namespace['f']()
+        self.assertEqual(result, 42)
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tornado/test/httpserver_test.py b/tornado/test/httpserver_test.py
index a626369b..30696a04 100644
--- a/tornado/test/httpserver_test.py
+++ b/tornado/test/httpserver_test.py
@@ -425,7 +425,7 @@ class HTTPServerRawTest(AsyncHTTPTestCase):
             self.wait()
 
     def test_malformed_headers(self):
-        with ExpectLog(gen_log, '.*Malformed HTTP headers'):
+        with ExpectLog(gen_log, '.*Malformed HTTP message.*no colon in header line'):
             self.stream.write(b'GET / HTTP/1.0\r\nasdf\r\n\r\n')
             self.io_loop.add_timeout(datetime.timedelta(seconds=0.05),
                                      self.stop)
diff --git a/tornado/test/httputil_test.py b/tornado/test/httputil_test.py
index 5c064dde..bd5832bc 100644
--- a/tornado/test/httputil_test.py
+++ b/tornado/test/httputil_test.py
@@ -4,6 +4,7 @@ from __future__ import absolute_import, division, print_function
 from tornado.httputil import (
     url_concat, parse_multipart_form_data, HTTPHeaders, format_timestamp,
     HTTPServerRequest, parse_request_start_line, parse_cookie, qs_to_qsl,
+    HTTPInputError,
 )
 from tornado.escape import utf8, native_str
 from tornado.util import PY3
@@ -283,6 +284,13 @@ Foo: even
                           ("Foo", "bar baz"),
                           ("Foo", "even more lines")])
 
+    def test_malformed_continuation(self):
+        # If the first line starts with whitespace, it's a
+        # continuation line with nothing to continue, so reject it
+        # (with a proper error).
+        data = " Foo: bar"
+        self.assertRaises(HTTPInputError, HTTPHeaders.parse, data)
+
     def test_unicode_newlines(self):
         # Ensure that only \r\n is recognized as a header separator, and not
         # the other newline-like unicode characters.
diff --git a/tornado/test/websocket_test.py b/tornado/test/websocket_test.py
index 38b3211d..2e14342c 100644
--- a/tornado/test/websocket_test.py
+++ b/tornado/test/websocket_test.py
@@ -621,6 +621,34 @@ class ClientPeriodicPingTest(WebSocketBaseTestCase):
         # TODO: test that the connection gets closed if ping responses stop.
 
 
+class ManualPingTest(WebSocketBaseTestCase):
+    def get_app(self):
+        class PingHandler(TestWebSocketHandler):
+            def on_ping(self, data):
+                self.write_message(data, binary=isinstance(data, bytes))
+
+        self.close_future = Future()
+        return Application([
+            ('/', PingHandler, dict(close_future=self.close_future)),
+        ])
+
+    @gen_test
+    def test_manual_ping(self):
+        ws = yield self.ws_connect('/')
+
+        self.assertRaises(ValueError, ws.ping, 'a' * 126)
+
+        ws.ping('hello')
+        resp = yield ws.read_message()
+        # on_ping always sees bytes.
+        self.assertEqual(resp, b'hello')
+
+        ws.ping(b'binary hello')
+        resp = yield ws.read_message()
+        self.assertEqual(resp, b'binary hello')
+        yield self.close(ws)
+
+
 class MaxMessageSizeTest(WebSocketBaseTestCase):
     def get_app(self):
         self.close_future = Future()