Repository: tornadoweb/tornado
Pull request #2350 (2018-04-08 18:37:42)

httpclient: Deprecate fetch callback, narrow `raise_error`
In 6.0, the `callback` argument to `fetch` will be going away as a part of the general move towards pure coroutines. The behavior of `raise_error=False` will also change to reduce the use of fake 599 response codes (which were mainly used to get around the limitations of the callback interface).

Deprecation warnings were added for both of these changes. This caused a fair amount of churn in the test suite, although hopefully less in applications that aren't doing as much probing of HTTP edge cases. 


Patch:
diff --git a/docs/httpclient.rst b/docs/httpclient.rst
index 78231618..cf3bc8e7 100644
--- a/docs/httpclient.rst
+++ b/docs/httpclient.rst
@@ -24,9 +24,13 @@
 
    Exceptions
    ----------
-   .. autoexception:: HTTPError
+   .. autoexception:: HTTPClientError
       :members:
 
+   .. exception:: HTTPError
+
+      Alias for `HTTPClientError`.
+
    Command-line interface
    ----------------------
 
diff --git a/docs/releases.rst b/docs/releases.rst
index 6f87edc3..74afb455 100644
--- a/docs/releases.rst
+++ b/docs/releases.rst
@@ -4,7 +4,6 @@ Release notes
 .. toctree::
    :maxdepth: 2
 
-   releases/v5.0.2
    releases/v5.0.1
    releases/v5.0.0
    releases/v4.5.3
diff --git a/docs/releases/v5.0.2.rst b/docs/releases/v5.0.2.rst
deleted file mode 100644
index a5252529..00000000
--- a/docs/releases/v5.0.2.rst
+++ /dev/null
@@ -1,18 +0,0 @@
-What's new in Tornado 5.0.2
-===========================
-
-Apr 7, 2018
------------
-
-Bug fixes
-~~~~~~~~~
-
-- Fixed a memory leak when `.IOLoop` objects are created and destroyed.
-- If `.AsyncTestCase.get_new_ioloop` returns a reference to a
-  preexisting event loop (typically when it has been overridden to
-  return `.IOLoop.current()`), the test's ``tearDown`` method will not
-  close this loop.
-- Fixed a confusing error message when the synchronous `.HTTPClient`
-  fails to initialize because an event loop is already running.
-- `.PeriodicCallback` no longer executes twice in a row due to
-  backwards clock adjustments.
diff --git a/tornado/auth.py b/tornado/auth.py
index beee6453..0069efcb 100644
--- a/tornado/auth.py
+++ b/tornado/auth.py
@@ -202,9 +202,9 @@ class OpenIdMixin(object):
         url = self._OPENID_ENDPOINT
         if http_client is None:
             http_client = self.get_auth_http_client()
-        http_client.fetch(url, functools.partial(
-            self._on_authentication_verified, callback),
-            method="POST", body=urllib_parse.urlencode(args))
+        fut = http_client.fetch(url, method="POST", body=urllib_parse.urlencode(args))
+        fut.add_done_callback(functools.partial(
+            self._on_authentication_verified, callback))
 
     def _openid_args(self, callback_uri, ax_attrs=[], oauth_scope=None):
         url = urlparse.urljoin(self.request.full_url(), callback_uri)
@@ -254,11 +254,16 @@ class OpenIdMixin(object):
             })
         return args
 
-    def _on_authentication_verified(self, future, response):
-        if response.error or b"is_valid:true" not in response.body:
+    def _on_authentication_verified(self, future, response_fut):
+        try:
+            response = response_fut.result()
+        except Exception as e:
             future.set_exception(AuthError(
-                "Invalid OpenID response: %s" % (response.error or
-                                                 response.body)))
+                "Error response %s" % e))
+            return
+        if b"is_valid:true" not in response.body:
+            future.set_exception(AuthError(
+                "Invalid OpenID response: %s" % response.body))
             return
 
         # Make sure we got back at least an email from attribute exchange
@@ -374,17 +379,17 @@ class OAuthMixin(object):
         if http_client is None:
             http_client = self.get_auth_http_client()
         if getattr(self, "_OAUTH_VERSION", "1.0a") == "1.0a":
-            http_client.fetch(
+            fut = http_client.fetch(
                 self._oauth_request_token_url(callback_uri=callback_uri,
-                                              extra_params=extra_params),
-                functools.partial(
-                    self._on_request_token,
-                    self._OAUTH_AUTHORIZE_URL,
-                    callback_uri,
-                    callback))
+                                              extra_params=extra_params))
+            fut.add_done_callback(functools.partial(
+                self._on_request_token,
+                self._OAUTH_AUTHORIZE_URL,
+                callback_uri,
+                callback))
         else:
-            http_client.fetch(
-                self._oauth_request_token_url(),
+            fut = http_client.fetch(self._oauth_request_token_url())
+            fut.add_done_callback(
                 functools.partial(
                     self._on_request_token, self._OAUTH_AUTHORIZE_URL,
                     callback_uri,
@@ -427,8 +432,8 @@ class OAuthMixin(object):
             token["verifier"] = oauth_verifier
         if http_client is None:
             http_client = self.get_auth_http_client()
-        http_client.fetch(self._oauth_access_token_url(token),
-                          functools.partial(self._on_access_token, callback))
+        fut = http_client.fetch(self._oauth_access_token_url(token))
+        fut.add_done_callback(functools.partial(self._on_access_token, callback))
 
     def _oauth_request_token_url(self, callback_uri=None, extra_params=None):
         consumer_token = self._oauth_consumer_token()
@@ -456,9 +461,11 @@ class OAuthMixin(object):
         return url + "?" + urllib_parse.urlencode(args)
 
     def _on_request_token(self, authorize_url, callback_uri, callback,
-                          response):
-        if response.error:
-            raise Exception("Could not get request token: %s" % response.error)
+                          response_fut):
+        try:
+            response = response_fut.result()
+        except Exception as e:
+            raise Exception("Could not get request token: %s" % e)
         request_token = _oauth_parse_response(response.body)
         data = (base64.b64encode(escape.utf8(request_token["key"])) + b"|" +
                 base64.b64encode(escape.utf8(request_token["secret"])))
@@ -498,8 +505,10 @@ class OAuthMixin(object):
         args["oauth_signature"] = signature
         return url + "?" + urllib_parse.urlencode(args)
 
-    def _on_access_token(self, future, response):
-        if response.error:
+    def _on_access_token(self, future, response_fut):
+        try:
+            response = response_fut.result()
+        except Exception:
             future.set_exception(AuthError("Could not fetch access token"))
             return
 
@@ -707,15 +716,16 @@ class OAuth2Mixin(object):
         callback = functools.partial(self._on_oauth2_request, callback)
         http = self.get_auth_http_client()
         if post_args is not None:
-            http.fetch(url, method="POST", body=urllib_parse.urlencode(post_args),
-                       callback=callback)
+            fut = http.fetch(url, method="POST", body=urllib_parse.urlencode(post_args))
         else:
-            http.fetch(url, callback=callback)
-
-    def _on_oauth2_request(self, future, response):
-        if response.error:
-            future.set_exception(AuthError("Error response %s fetching %s" %
-                                           (response.error, response.request.url)))
+            fut = http.fetch(url)
+        fut.add_done_callback(callback)
+
+    def _on_oauth2_request(self, future, response_fut):
+        try:
+            response = response_fut.result()
+        except Exception as e:
+            future.set_exception(AuthError("Error response %s" % e))
             return
 
         future_set_result_unless_cancelled(future, escape.json_decode(response.body))
@@ -857,18 +867,19 @@ class TwitterMixin(OAuthMixin):
         if args:
             url += "?" + urllib_parse.urlencode(args)
         http = self.get_auth_http_client()
-        http_callback = functools.partial(self._on_twitter_request, callback)
+        http_callback = functools.partial(self._on_twitter_request, callback, url)
         if post_args is not None:
-            http.fetch(url, method="POST", body=urllib_parse.urlencode(post_args),
-                       callback=http_callback)
+            fut = http.fetch(url, method="POST", body=urllib_parse.urlencode(post_args))
         else:
-            http.fetch(url, callback=http_callback)
+            fut = http.fetch(url)
+        fut.add_done_callback(http_callback)
 
-    def _on_twitter_request(self, future, response):
-        if response.error:
+    def _on_twitter_request(self, future, url, response_fut):
+        try:
+            response = response_fut.result()
+        except Exception as e:
             future.set_exception(AuthError(
-                "Error response %s fetching %s" % (response.error,
-                                                   response.request.url)))
+                "Error response %s fetching %s" % (e, url)))
             return
         future_set_result_unless_cancelled(future, escape.json_decode(response.body))
 
@@ -967,16 +978,18 @@ class GoogleOAuth2Mixin(OAuth2Mixin):
             "grant_type": "authorization_code",
         })
 
-        http.fetch(self._OAUTH_ACCESS_TOKEN_URL,
-                   functools.partial(self._on_access_token, callback),
-                   method="POST",
-                   headers={'Content-Type': 'application/x-www-form-urlencoded'},
-                   body=body)
+        fut = http.fetch(self._OAUTH_ACCESS_TOKEN_URL,
+                         method="POST",
+                         headers={'Content-Type': 'application/x-www-form-urlencoded'},
+                         body=body)
+        fut.add_done_callback(functools.partial(self._on_access_token, callback))
 
-    def _on_access_token(self, future, response):
+    def _on_access_token(self, future, response_fut):
         """Callback function for the exchange to the access token."""
-        if response.error:
-            future.set_exception(AuthError('Google auth error: %s' % str(response)))
+        try:
+            response = response_fut.result()
+        except Exception as e:
+            future.set_exception(AuthError('Google auth error: %s' % str(e)))
             return
 
         args = escape.json_decode(response.body)
@@ -1053,15 +1066,17 @@ class FacebookGraphMixin(OAuth2Mixin):
         if extra_fields:
             fields.update(extra_fields)
 
-        http.fetch(self._oauth_request_token_url(**args),
-                   functools.partial(self._on_access_token, redirect_uri, client_id,
-                                     client_secret, callback, fields))
+        fut = http.fetch(self._oauth_request_token_url(**args))
+        fut.add_done_callback(functools.partial(self._on_access_token, redirect_uri, client_id,
+                                                client_secret, callback, fields))
 
     @gen.coroutine
     def _on_access_token(self, redirect_uri, client_id, client_secret,
-                         future, fields, response):
-        if response.error:
-            future.set_exception(AuthError('Facebook auth error: %s' % str(response)))
+                         future, fields, response_fut):
+        try:
+            response = response_fut.result()
+        except Exception as e:
+            future.set_exception(AuthError('Facebook auth error: %s' % str(e)))
             return
 
         args = escape.json_decode(response.body)
diff --git a/tornado/httpclient.py b/tornado/httpclient.py
index 9c438d15..f0a2df88 100644
--- a/tornado/httpclient.py
+++ b/tornado/httpclient.py
@@ -42,6 +42,7 @@ from __future__ import absolute_import, division, print_function
 
 import functools
 import time
+import warnings
 import weakref
 
 from tornado.concurrent import Future, future_set_result_unless_cancelled
@@ -238,6 +239,18 @@ class AsyncHTTPClient(Configurable):
         In the callback interface, `HTTPError` is not automatically raised.
         Instead, you must check the response's ``error`` attribute or
         call its `~HTTPResponse.rethrow` method.
+
+        .. deprecated:: 5.1
+
+           The ``callback`` argument is deprecated and will be removed
+           in 6.0. Use the returned `.Future` instead.
+
+           The ``raise_error=False`` argument currently suppresses
+           *all* errors, encapsulating them in `HTTPResponse` objects
+           with a 599 response code. This will change in Tornado 6.0:
+           ``raise_error=False`` will only affect the `HTTPError`
+           raised when a non-200 response code is used.
+
         """
         if self._closed:
             raise RuntimeError("fetch() called on closed AsyncHTTPClient")
@@ -253,6 +266,8 @@ class AsyncHTTPClient(Configurable):
         request = _RequestProxy(request, self.defaults)
         future = Future()
         if callback is not None:
+            warnings.warn("callback arguments are deprecated, use the returned Future instead",
+                          DeprecationWarning)
             callback = stack_context.wrap(callback)
 
             def handle_future(future):
@@ -270,8 +285,13 @@ class AsyncHTTPClient(Configurable):
 
         def handle_response(response):
             if raise_error and response.error:
+                if isinstance(response.error, HTTPError):
+                    response.error.response = response
                 future.set_exception(response.error)
             else:
+                if response.error and not response._error_is_response_code:
+                    warnings.warn("raise_error=False will allow '%s' to be raised in the future" %
+                                  response.error, DeprecationWarning)
                 future_set_result_unless_cancelled(future, response)
         self.fetch_impl(request, handle_response)
         return future
@@ -585,8 +605,10 @@ class HTTPResponse(object):
             self.effective_url = request.url
         else:
             self.effective_url = effective_url
+        self._error_is_response_code = False
         if error is None:
             if self.code < 200 or self.code >= 300:
+                self._error_is_response_code = True
                 self.error = HTTPError(self.code, message=self.reason,
                                        response=self)
             else:
@@ -615,7 +637,7 @@ class HTTPResponse(object):
         return "%s(%s)" % (self.__class__.__name__, args)
 
 
-class HTTPError(Exception):
+class HTTPClientError(Exception):
     """Exception thrown for an unsuccessful HTTP request.
 
     Attributes:
@@ -628,12 +650,18 @@ class HTTPError(Exception):
     Note that if ``follow_redirects`` is False, redirects become HTTPErrors,
     and you can look at ``error.response.headers['Location']`` to see the
     destination of the redirect.
+
+    .. versionchanged:: 5.1
+
+       Renamed from ``HTTPError`` to ``HTTPClientError`` to avoid collisions with
+       `tornado.web.HTTPError`. The name ``tornado.httpclient.HTTPError`` remains
+       as an alias.
     """
     def __init__(self, code, message=None, response=None):
         self.code = code
         self.message = message or httputil.responses.get(code, "Unknown")
         self.response = response
-        super(HTTPError, self).__init__(code, message, response)
+        super(HTTPClientError, self).__init__(code, message, response)
 
     def __str__(self):
         return "HTTP %d: %s" % (self.code, self.message)
@@ -645,6 +673,9 @@ class HTTPError(Exception):
     __repr__ = __str__
 
 
+HTTPError = HTTPClientError
+
+
 class _RequestProxy(object):
     """Combines an object with a dictionary of defaults.
 
diff --git a/tornado/ioloop.py b/tornado/ioloop.py
index 48700139..f6ec177b 100644
--- a/tornado/ioloop.py
+++ b/tornado/ioloop.py
@@ -1213,31 +1213,11 @@ class PeriodicCallback(object):
 
     def _schedule_next(self):
         if self._running:
-            self._update_next(self.io_loop.time())
-            self._timeout = self.io_loop.add_timeout(self._next_timeout, self._run)
+            current_time = self.io_loop.time()
 
-    def _update_next(self, current_time):
-        callback_time_sec = self.callback_time / 1000.0
-        if self._next_timeout <= current_time:
-            # The period should be measured from the start of one call
-            # to the start of the next. If one call takes too long,
-            # skip cycles to get back to a multiple of the original
-            # schedule.
-            self._next_timeout += (math.floor((current_time - self._next_timeout) /
-                                              callback_time_sec) + 1) * callback_time_sec
-        else:
-            # If the clock moved backwards, ensure we advance the next
-            # timeout instead of recomputing the same value again.
-            # This may result in long gaps between callbacks if the
-            # clock jumps backwards by a lot, but the far more common
-            # scenario is a small NTP adjustment that should just be
-            # ignored.
-            #
-            # Note that on some systems if time.time() runs slower
-            # than time.monotonic() (most common on windows), we
-            # effectively experience a small backwards time jump on
-            # every iteration because PeriodicCallback uses
-            # time.time() while asyncio schedules callbacks using
-            # time.monotonic().
-            # https://github.com/tornadoweb/tornado/issues/2333
-            self._next_timeout += callback_time_sec
+            if self._next_timeout <= current_time:
+                callback_time_sec = self.callback_time / 1000.0
+                self._next_timeout += (math.floor((current_time - self._next_timeout) /
+                                                  callback_time_sec) + 1) * callback_time_sec
+
+            self._timeout = self.io_loop.add_timeout(self._next_timeout, self._run)
diff --git a/tornado/platform/caresresolver.py b/tornado/platform/caresresolver.py
index 38247f7e..768cb624 100644
--- a/tornado/platform/caresresolver.py
+++ b/tornado/platform/caresresolver.py
@@ -2,6 +2,7 @@ from __future__ import absolute_import, division, print_function
 import pycares  # type: ignore
 import socket
 
+from tornado.concurrent import Future
 from tornado import gen
 from tornado.ioloop import IOLoop
 from tornado.netutil import Resolver, is_valid_ip
@@ -55,11 +56,10 @@ class CaresResolver(Resolver):
             addresses = [host]
         else:
             # gethostbyname doesn't take callback as a kwarg
-            self.channel.gethostbyname(host, family, (yield gen.Callback(1)))
-            callback_args = yield gen.Wait(1)
-            assert isinstance(callback_args, gen.Arguments)
-            assert not callback_args.kwargs
-            result, error = callback_args.args
+            fut = Future()
+            self.channel.gethostbyname(host, family,
+                                       lambda result, error: fut.set_result((result, error)))
+            result, error = yield fut
             if error:
                 raise IOError('C-Ares returned error %s: %s while resolving %s' %
                               (error, pycares.errno.strerror(error), host))
diff --git a/tornado/simple_httpclient.py b/tornado/simple_httpclient.py
index 1fc3e707..7696dd18 100644
--- a/tornado/simple_httpclient.py
+++ b/tornado/simple_httpclient.py
@@ -35,6 +35,39 @@ except ImportError:
     ssl = None
 
 
+class HTTPTimeoutError(HTTPError):
+    """Error raised by SimpleAsyncHTTPClient on timeout.
+
+    For historical reasons, this is a subclass of `.HTTPClientError`
+    which simulates a response code of 599.
+
+    .. versionadded:: 5.1
+    """
+    def __init__(self, message):
+        super(HTTPTimeoutError, self).__init__(599, message=message)
+
+    def __str__(self):
+        return self.message
+
+
+class HTTPStreamClosedError(HTTPError):
+    """Error raised by SimpleAsyncHTTPClient when the underlying stream is closed.
+
+    When a more specific exception is available (such as `ConnectionResetError`),
+    it may be raised instead of this one.
+
+    For historical reasons, this is a subclass of `.HTTPClientError`
+    which simulates a response code of 599.
+
+    .. versionadded:: 5.1
+    """
+    def __init__(self, message):
+        super(HTTPStreamClosedError, self).__init__(599, message=message)
+
+    def __str__(self):
+        return self.message
+
+
 class SimpleAsyncHTTPClient(AsyncHTTPClient):
     """Non-blocking HTTP client with no external dependencies.
 
@@ -168,7 +201,7 @@ class SimpleAsyncHTTPClient(AsyncHTTPClient):
 
         error_message = "Timeout {0}".format(info) if info else "Timeout"
         timeout_response = HTTPResponse(
-            request, 599, error=HTTPError(599, error_message),
+            request, 599, error=HTTPTimeoutError(error_message),
             request_time=self.io_loop.time() - request.start_time)
         self.io_loop.add_callback(callback, timeout_response)
         del self.waiting[key]
@@ -261,14 +294,14 @@ class _HTTPConnection(httputil.HTTPMessageDelegate):
     def _on_timeout(self, info=None):
         """Timeout callback of _HTTPConnection instance.
 
-        Raise a timeout HTTPError when a timeout occurs.
+        Raise a `HTTPTimeoutError` when a timeout occurs.
 
         :info string key: More detailed timeout information.
         """
         self._timeout = None
         error_message = "Timeout {0}".format(info) if info else "Timeout"
         if self.final_callback is not None:
-            raise HTTPError(599, error_message)
+            raise HTTPTimeoutError(error_message)
 
     def _remove_timeout(self):
         if self._timeout is not None:
@@ -413,7 +446,7 @@ class _HTTPConnection(httputil.HTTPMessageDelegate):
             self._remove_timeout()
             if isinstance(value, StreamClosedError):
                 if value.real_error is None:
-                    value = HTTPError(599, "Stream closed")
+                    value = HTTPStreamClosedError("Stream closed")
                 else:
                     value = value.real_error
             self._run_callback(HTTPResponse(self.request, 599, error=value,
@@ -439,8 +472,8 @@ class _HTTPConnection(httputil.HTTPMessageDelegate):
             if self.stream.error:
                 raise self.stream.error
             try:
-                raise HTTPError(599, message)
-            except HTTPError:
+                raise HTTPStreamClosedError(message)
+            except HTTPStreamClosedError:
                 self._handle_exception(*sys.exc_info())
 
     def headers_received(self, first_line, headers):
@@ -498,7 +531,8 @@ class _HTTPConnection(httputil.HTTPMessageDelegate):
             final_callback = self.final_callback
             self.final_callback = None
             self._release()
-            self.client.fetch(new_request, final_callback)
+            fut = self.client.fetch(new_request, raise_error=False)
+            fut.add_done_callback(lambda f: final_callback(f.result()))
             self._on_end_request()
             return
         if self.request.streaming_callback:
diff --git a/tox.ini b/tox.ini
index c8c4b67f..ddc6af28 100644
--- a/tox.ini
+++ b/tox.ini
@@ -35,7 +35,8 @@ envlist =
         {py2,py3}-full-caresresolver,
 
         # Other configurations; see comments below.
-        {py2,py3}-{monotonic,opt},
+        py2-monotonic,
+        {py2,py3}-opt,
         py3-{lang_c,lang_utf8},
         py2-locale,
         {py27,py3}-unittest2,


Test patch:
diff --git a/tornado/test/curl_httpclient_test.py b/tornado/test/curl_httpclient_test.py
index cc29ee10..d0cfa979 100644
--- a/tornado/test/curl_httpclient_test.py
+++ b/tornado/test/curl_httpclient_test.py
@@ -4,12 +4,12 @@ from __future__ import absolute_import, division, print_function
 from hashlib import md5
 
 from tornado.escape import utf8
-from tornado.httpclient import HTTPRequest
+from tornado.httpclient import HTTPRequest, HTTPClientError
 from tornado.locks import Event
 from tornado.stack_context import ExceptionStackContext
 from tornado.testing import AsyncHTTPTestCase, gen_test
 from tornado.test import httpclient_test
-from tornado.test.util import unittest
+from tornado.test.util import unittest, ignore_deprecation
 from tornado.web import Application, RequestHandler
 
 
@@ -131,5 +131,14 @@ class CurlHTTPClientTestCase(AsyncHTTPTestCase):
     def test_failed_setup(self):
         self.http_client = self.create_client(max_clients=1)
         for i in range(5):
-            response = self.fetch(u'/ユニコード')
+            with ignore_deprecation():
+                response = self.fetch(u'/ユニコード')
             self.assertIsNot(response.error, None)
+
+            with self.assertRaises((UnicodeEncodeError, HTTPClientError)):
+                # This raises UnicodeDecodeError on py3 and
+                # HTTPClientError(404) on py2. The main motivation of
+                # this test is to ensure that the UnicodeEncodeError
+                # during the setup phase doesn't lead the request to
+                # be dropped on the floor.
+                response = self.fetch(u'/ユニコード', raise_error=True)
diff --git a/tornado/test/httpclient_test.py b/tornado/test/httpclient_test.py
index a76d9eae..f980b48c 100644
--- a/tornado/test/httpclient_test.py
+++ b/tornado/test/httpclient_test.py
@@ -20,7 +20,7 @@ from tornado.log import gen_log
 from tornado import netutil
 from tornado.stack_context import ExceptionStackContext, NullContext
 from tornado.testing import AsyncHTTPTestCase, bind_unused_port, gen_test, ExpectLog
-from tornado.test.util import unittest, skipOnTravis
+from tornado.test.util import unittest, skipOnTravis, ignore_deprecation
 from tornado.web import Application, RequestHandler, url
 from tornado.httputil import format_timestamp, HTTPHeaders
 
@@ -212,8 +212,7 @@ Transfer-Encoding: chunked
                 stream.read_until(b"\r\n\r\n",
                                   functools.partial(write_response, stream))
             netutil.add_accept_handler(sock, accept_callback)
-            self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop)
-            resp = self.wait()
+            resp = self.fetch("http://127.0.0.1:%d/" % port)
             resp.rethrow()
             self.assertEqual(resp.body, b"12")
             self.io_loop.remove_handler(sock.fileno())
@@ -255,10 +254,10 @@ Transfer-Encoding: chunked
         # on an unknown mode.
         with ExpectLog(gen_log, "uncaught exception", required=False):
             with self.assertRaises((ValueError, HTTPError)):
-                response = self.fetch("/auth", auth_username="Aladdin",
-                                      auth_password="open sesame",
-                                      auth_mode="asdf")
-                response.rethrow()
+                self.fetch("/auth", auth_username="Aladdin",
+                           auth_password="open sesame",
+                           auth_mode="asdf",
+                           raise_error=True)
 
     def test_follow_redirect(self):
         response = self.fetch("/countdown/2", follow_redirects=False)
@@ -272,8 +271,7 @@ Transfer-Encoding: chunked
 
     def test_credentials_in_url(self):
         url = self.get_url("/auth").replace("http://", "http://me:secret@")
-        self.http_client.fetch(url, self.stop)
-        response = self.wait()
+        response = self.fetch(url)
         self.assertEqual(b"Basic " + base64.b64encode(b"me:secret"),
                          response.body)
 
@@ -353,14 +351,14 @@ Transfer-Encoding: chunked
         self.assertEqual(len(exc_info), 1)
         self.assertIs(exc_info[0][0], ZeroDivisionError)
 
+    @gen_test
     def test_configure_defaults(self):
         defaults = dict(user_agent='TestDefaultUserAgent', allow_ipv6=False)
         # Construct a new instance of the configured client class
         client = self.http_client.__class__(force_instance=True,
                                             defaults=defaults)
         try:
-            client.fetch(self.get_url('/user_agent'), callback=self.stop)
-            response = self.wait()
+            response = yield client.fetch(self.get_url('/user_agent'))
             self.assertEqual(response.body, b'TestDefaultUserAgent')
         finally:
             client.close()
@@ -400,8 +398,7 @@ X-XSS-Protection: 1;
                 stream.read_until(b"\r\n\r\n",
                                   functools.partial(write_response, stream))
             netutil.add_accept_handler(sock, accept_callback)
-            self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop)
-            resp = self.wait()
+            resp = self.fetch("http://127.0.0.1:%d/" % port)
             resp.rethrow()
             self.assertEqual(resp.headers['X-XSS-Protection'], "1; mode=block")
             self.io_loop.remove_handler(sock.fileno())
@@ -431,8 +428,9 @@ X-XSS-Protection: 1;
             self.stop()
         self.io_loop.handle_callback_exception = handle_callback_exception
         with NullContext():
-            self.http_client.fetch(self.get_url('/hello'),
-                                   lambda response: 1 / 0)
+            with ignore_deprecation():
+                self.http_client.fetch(self.get_url('/hello'),
+                                       lambda response: 1 / 0)
         self.wait()
         self.assertEqual(exc_info[0][0], ZeroDivisionError)
 
@@ -483,8 +481,7 @@ X-XSS-Protection: 1;
         # These methods require a body.
         for method in ('POST', 'PUT', 'PATCH'):
             with self.assertRaises(ValueError) as context:
-                resp = self.fetch('/all_methods', method=method)
-                resp.rethrow()
+                self.fetch('/all_methods', method=method, raise_error=True)
             self.assertIn('must not be None', str(context.exception))
 
             resp = self.fetch('/all_methods', method=method,
@@ -494,16 +491,14 @@ X-XSS-Protection: 1;
         # These methods don't allow a body.
         for method in ('GET', 'DELETE', 'OPTIONS'):
             with self.assertRaises(ValueError) as context:
-                resp = self.fetch('/all_methods', method=method, body=b'asdf')
-                resp.rethrow()
+                self.fetch('/all_methods', method=method, body=b'asdf', raise_error=True)
             self.assertIn('must be None', str(context.exception))
 
             # In most cases this can be overridden, but curl_httpclient
             # does not allow body with a GET at all.
             if method != 'GET':
-                resp = self.fetch('/all_methods', method=method, body=b'asdf',
-                                  allow_nonstandard_methods=True)
-                resp.rethrow()
+                self.fetch('/all_methods', method=method, body=b'asdf',
+                           allow_nonstandard_methods=True, raise_error=True)
                 self.assertEqual(resp.code, 200)
 
     # This test causes odd failures with the combination of
diff --git a/tornado/test/httpserver_test.py b/tornado/test/httpserver_test.py
index bfac9ab5..b53dd8bf 100644
--- a/tornado/test/httpserver_test.py
+++ b/tornado/test/httpserver_test.py
@@ -4,6 +4,7 @@ from tornado import netutil
 from tornado.concurrent import Future
 from tornado.escape import json_decode, json_encode, utf8, _unicode, recursive_unicode, native_str
 from tornado.http1connection import HTTP1Connection
+from tornado.httpclient import HTTPError
 from tornado.httpserver import HTTPServer
 from tornado.httputil import HTTPHeaders, HTTPMessageDelegate, HTTPServerConnectionDelegate, ResponseStartLine  # noqa: E501
 from tornado.iostream import IOStream
@@ -109,22 +110,19 @@ class SSLTestMixin(object):
         # misbehaving.
         with ExpectLog(gen_log, '(SSL Error|uncaught exception)'):
             with ExpectLog(gen_log, 'Uncaught exception', required=False):
-                self.http_client.fetch(
-                    self.get_url("/").replace('https:', 'http:'),
-                    self.stop,
-                    request_timeout=3600,
-                    connect_timeout=3600)
-                response = self.wait()
-        self.assertEqual(response.code, 599)
+                with self.assertRaises((IOError, HTTPError)):
+                    self.fetch(
+                        self.get_url("/").replace('https:', 'http:'),
+                        request_timeout=3600,
+                        connect_timeout=3600,
+                        raise_error=True)
 
     def test_error_logging(self):
         # No stack traces are logged for SSL errors.
         with ExpectLog(gen_log, 'SSL Error') as expect_log:
-            self.http_client.fetch(
-                self.get_url("/").replace("https:", "http:"),
-                self.stop)
-            response = self.wait()
-            self.assertEqual(response.code, 599)
+            with self.assertRaises((IOError, HTTPError)):
+                self.fetch(self.get_url("/").replace("https:", "http:"),
+                           raise_error=True)
         self.assertFalse(expect_log.logged_stack)
 
 # Python's SSL implementation differs significantly between versions.
@@ -965,11 +963,14 @@ class MaxHeaderSizeTest(AsyncHTTPTestCase):
 
     def test_large_headers(self):
         with ExpectLog(gen_log, "Unsatisfiable read", required=False):
-            response = self.fetch("/", headers={'X-Filler': 'a' * 1000})
-        # 431 is "Request Header Fields Too Large", defined in RFC
-        # 6585. However, many implementations just close the
-        # connection in this case, resulting in a 599.
-        self.assertIn(response.code, (431, 599))
+            try:
+                self.fetch("/", headers={'X-Filler': 'a' * 1000}, raise_error=True)
+                self.fail("did not raise expected exception")
+            except HTTPError as e:
+                # 431 is "Request Header Fields Too Large", defined in RFC
+                # 6585. However, many implementations just close the
+                # connection in this case, resulting in a 599.
+                self.assertIn(e.response.code, (431, 599))
 
 
 @skipOnTravis
@@ -1066,24 +1067,25 @@ class BodyLimitsTest(AsyncHTTPTestCase):
             response = self.fetch('/buffered', method='PUT', body=b'a' * 10240)
         self.assertEqual(response.code, 400)
 
+    @unittest.skipIf(os.name == 'nt', 'flaky on windows')
     def test_large_body_buffered_chunked(self):
+        # This test is flaky on windows for unknown reasons.
         with ExpectLog(gen_log, '.*chunked body too large'):
             response = self.fetch('/buffered', method='PUT',
                                   body_producer=lambda write: write(b'a' * 10240))
-        # this test is flaky on windows; accept 400 (expected) or 599
-        self.assertIn(response.code, [400, 599])
+        self.assertEqual(response.code, 400)
 
     def test_large_body_streaming(self):
         with ExpectLog(gen_log, '.*Content-Length too long'):
             response = self.fetch('/streaming', method='PUT', body=b'a' * 10240)
         self.assertEqual(response.code, 400)
 
+    @unittest.skipIf(os.name == 'nt', 'flaky on windows')
     def test_large_body_streaming_chunked(self):
         with ExpectLog(gen_log, '.*chunked body too large'):
             response = self.fetch('/streaming', method='PUT',
                                   body_producer=lambda write: write(b'a' * 10240))
-        # this test is flaky on windows; accept 400 (expected) or 599
-        self.assertIn(response.code, [400, 599])
+        self.assertEqual(response.code, 400)
 
     def test_large_body_streaming_override(self):
         response = self.fetch('/streaming?expected_size=10240', method='PUT',
diff --git a/tornado/test/ioloop_test.py b/tornado/test/ioloop_test.py
index 9f7c1847..09f71c5d 100644
--- a/tornado/test/ioloop_test.py
+++ b/tornado/test/ioloop_test.py
@@ -789,62 +789,6 @@ class TestPeriodicCallback(unittest.TestCase):
         io_loop.close()
 
 
-class TestPeriodicCallbackMath(unittest.TestCase):
-    def simulate_calls(self, pc, durations):
-        """Simulate a series of calls to the PeriodicCallback.
-
-        Pass a list of call durations in seconds (negative values
-        work to simulate clock adjustments during the call, or more or
-        less equivalently, between calls). This method returns the
-        times at which each call would be made.
-        """
-        calls = []
-        now = 1000
-        pc._next_timeout = now
-        for d in durations:
-            pc._update_next(now)
-            calls.append(pc._next_timeout)
-            now = pc._next_timeout + d
-        return calls
-
-    def test_basic(self):
-        pc = PeriodicCallback(None, 10000)
-        self.assertEqual(self.simulate_calls(pc, [0] * 5),
-                         [1010, 1020, 1030, 1040, 1050])
-
-    def test_overrun(self):
-        # If a call runs for too long, we skip entire cycles to get
-        # back on schedule.
-        call_durations = [9, 9, 10, 11, 20, 20, 35, 35, 0, 0, 0]
-        expected = [
-            1010, 1020, 1030,  # first 3 calls on schedule
-            1050, 1070,  # next 2 delayed one cycle
-            1100, 1130,  # next 2 delayed 2 cycles
-            1170, 1210,  # next 2 delayed 3 cycles
-            1220, 1230,  # then back on schedule.
-        ]
-
-        pc = PeriodicCallback(None, 10000)
-        self.assertEqual(self.simulate_calls(pc, call_durations),
-                         expected)
-
-    def test_clock_backwards(self):
-        pc = PeriodicCallback(None, 10000)
-        # Backwards jumps are ignored, potentially resulting in a
-        # slightly slow schedule (although we assume that when
-        # time.time() and time.monotonic() are different, time.time()
-        # is getting adjusted by NTP and is therefore more accurate)
-        self.assertEqual(self.simulate_calls(pc, [-2, -1, -3, -2, 0]),
-                         [1010, 1020, 1030, 1040, 1050])
-
-        # For big jumps, we should perhaps alter the schedule, but we
-        # don't currently. This trace shows that we run callbacks
-        # every 10s of time.time(), but the first and second calls are
-        # 110s of real time apart because the backwards jump is
-        # ignored.
-        self.assertEqual(self.simulate_calls(pc, [-100, 0, 0]),
-                         [1010, 1020, 1030])
-
 class TestIOLoopConfiguration(unittest.TestCase):
     def run_python(self, *statements):
         statements = [
diff --git a/tornado/test/runtests.py b/tornado/test/runtests.py
index d5bd769c..e4f938d3 100644
--- a/tornado/test/runtests.py
+++ b/tornado/test/runtests.py
@@ -137,6 +137,8 @@ def main():
     # 2.7 and 3.2
     warnings.filterwarnings("ignore", category=DeprecationWarning,
                             message="Please use assert.* instead")
+    warnings.filterwarnings("ignore", category=PendingDeprecationWarning,
+                            message="Please use assert.* instead")
     # Twisted 15.0.0 triggers some warnings on py3 with -bb.
     warnings.filterwarnings("ignore", category=BytesWarning,
                             module=r"twisted\..*")
diff --git a/tornado/test/simple_httpclient_test.py b/tornado/test/simple_httpclient_test.py
index 973353e0..b7daeda9 100644
--- a/tornado/test/simple_httpclient_test.py
+++ b/tornado/test/simple_httpclient_test.py
@@ -16,14 +16,17 @@ from tornado import gen
 from tornado.httpclient import AsyncHTTPClient
 from tornado.httputil import HTTPHeaders, ResponseStartLine
 from tornado.ioloop import IOLoop
+from tornado.iostream import UnsatisfiableReadError
 from tornado.log import gen_log
 from tornado.concurrent import Future
 from tornado.netutil import Resolver, bind_sockets
-from tornado.simple_httpclient import SimpleAsyncHTTPClient
+from tornado.simple_httpclient import SimpleAsyncHTTPClient, HTTPStreamClosedError
 from tornado.test.httpclient_test import ChunkHandler, CountdownHandler, HelloWorldHandler, RedirectHandler  # noqa: E501
 from tornado.test import httpclient_test
-from tornado.testing import AsyncHTTPTestCase, AsyncHTTPSTestCase, AsyncTestCase, ExpectLog
-from tornado.test.util import skipOnTravis, skipIfNoIPv6, refusing_port, skipBefore35, exec_test
+from tornado.testing import (AsyncHTTPTestCase, AsyncHTTPSTestCase, AsyncTestCase,
+                             ExpectLog, gen_test)
+from tornado.test.util import (skipOnTravis, skipIfNoIPv6, refusing_port, skipBefore35,
+                               exec_test, ignore_deprecation)
 from tornado.web import RequestHandler, Application, asynchronous, url, stream_request_body
 
 
@@ -167,8 +170,8 @@ class SimpleHTTPClientTestMixin(object):
             # Send 4 requests.  Two can be sent immediately, while the others
             # will be queued
             for i in range(4):
-                client.fetch(self.get_url("/trigger"),
-                             lambda response, i=i: (seen.append(i), self.stop()))
+                client.fetch(self.get_url("/trigger")).add_done_callback(
+                    lambda fut, i=i: (seen.append(i), self.stop()))
             self.wait(condition=lambda: len(self.triggers) == 2)
             self.assertEqual(len(client.queue), 2)
 
@@ -187,12 +190,12 @@ class SimpleHTTPClientTestMixin(object):
             self.assertEqual(set(seen), set([0, 1, 2, 3]))
             self.assertEqual(len(self.triggers), 0)
 
+    @gen_test
     def test_redirect_connection_limit(self):
         # following redirects should not consume additional connections
         with closing(self.create_client(max_clients=1)) as client:
-            client.fetch(self.get_url('/countdown/3'), self.stop,
-                         max_redirects=3)
-            response = self.wait()
+            response = yield client.fetch(self.get_url('/countdown/3'),
+                                          max_redirects=3)
             response.rethrow()
 
     def test_gzip(self):
@@ -236,6 +239,7 @@ class SimpleHTTPClientTestMixin(object):
             self.assertEqual("POST", response.request.method)
 
     @skipOnTravis
+    @gen_test
     def test_connect_timeout(self):
         timeout = 0.1
         timeout_min, timeout_max = 0.099, 1.0
@@ -245,13 +249,14 @@ class SimpleHTTPClientTestMixin(object):
                 return Future()  # never completes
 
         with closing(self.create_client(resolver=TimeoutResolver())) as client:
-            client.fetch(self.get_url('/hello'), self.stop,
-                         connect_timeout=timeout)
-            response = self.wait()
-            self.assertEqual(response.code, 599)
-            self.assertTrue(timeout_min < response.request_time < timeout_max,
-                            response.request_time)
-            self.assertEqual(str(response.error), "HTTP 599: Timeout while connecting")
+            with ignore_deprecation():
+                response = yield client.fetch(self.get_url('/hello'),
+                                              connect_timeout=timeout,
+                                              raise_error=False)
+                self.assertEqual(response.code, 599)
+                self.assertTrue(timeout_min < response.request_time < timeout_max,
+                                response.request_time)
+                self.assertEqual(str(response.error), "Timeout while connecting")
 
     @skipOnTravis
     def test_request_timeout(self):
@@ -261,11 +266,12 @@ class SimpleHTTPClientTestMixin(object):
             timeout = 0.5
             timeout_min, timeout_max = 0.4, 0.6
 
-        response = self.fetch('/trigger?wake=false', request_timeout=timeout)
-        self.assertEqual(response.code, 599)
+        with ignore_deprecation():
+            response = self.fetch('/trigger?wake=false', request_timeout=timeout)
+            self.assertEqual(response.code, 599)
         self.assertTrue(timeout_min < response.request_time < timeout_max,
                         response.request_time)
-        self.assertEqual(str(response.error), "HTTP 599: Timeout during request")
+        self.assertEqual(str(response.error), "Timeout during request")
         # trigger the hanging request to let it clean up after itself
         self.triggers.popleft()()
 
@@ -277,12 +283,10 @@ class SimpleHTTPClientTestMixin(object):
         url = '%s://[::1]:%d/hello' % (self.get_protocol(), port)
 
         # ipv6 is currently enabled by default but can be disabled
-        self.http_client.fetch(url, self.stop, allow_ipv6=False)
-        response = self.wait()
-        self.assertEqual(response.code, 599)
+        with self.assertRaises(Exception):
+            self.fetch(url, allow_ipv6=False, raise_error=True)
 
-        self.http_client.fetch(url, self.stop)
-        response = self.wait()
+        response = self.fetch(url)
         self.assertEqual(response.body, b"Hello world!")
 
     def xtest_multiple_content_length_accepted(self):
@@ -324,16 +328,15 @@ class SimpleHTTPClientTestMixin(object):
         self.assertTrue(host_re.match(response.body))
 
         url = self.get_url("/host_echo").replace("http://", "http://me:secret@")
-        self.http_client.fetch(url, self.stop)
-        response = self.wait()
+        response = self.fetch(url)
         self.assertTrue(host_re.match(response.body), response.body)
 
     def test_connection_refused(self):
         cleanup_func, port = refusing_port()
         self.addCleanup(cleanup_func)
         with ExpectLog(gen_log, ".*", required=False):
-            self.http_client.fetch("http://127.0.0.1:%d/" % port, self.stop)
-            response = self.wait()
+            with ignore_deprecation():
+                response = self.fetch("http://127.0.0.1:%d/" % port)
         self.assertEqual(599, response.code)
 
         if sys.platform != 'cygwin':
@@ -349,20 +352,24 @@ class SimpleHTTPClientTestMixin(object):
                             response.error)
 
     def test_queue_timeout(self):
-        with closing(self.create_client(max_clients=1)) as client:
-            client.fetch(self.get_url('/trigger'), self.stop,
-                         request_timeout=10)
-            # Wait for the trigger request to block, not complete.
-            self.wait()
-            client.fetch(self.get_url('/hello'), self.stop,
-                         connect_timeout=0.1)
-            response = self.wait()
-
-            self.assertEqual(response.code, 599)
-            self.assertTrue(response.request_time < 1, response.request_time)
-            self.assertEqual(str(response.error), "HTTP 599: Timeout in request queue")
-            self.triggers.popleft()()
-            self.wait()
+        with ignore_deprecation():
+            with closing(self.create_client(max_clients=1)) as client:
+                # Wait for the trigger request to block, not complete.
+                fut1 = client.fetch(self.get_url('/trigger'),
+                                    request_timeout=10, raise_error=False)
+                self.wait()
+                fut2 = client.fetch(self.get_url('/hello'),
+                                    connect_timeout=0.1, raise_error=False)
+                fut2.add_done_callback(self.stop)
+                response = self.wait().result()
+
+                self.assertEqual(response.code, 599)
+                self.assertTrue(response.request_time < 1, response.request_time)
+                self.assertEqual(str(response.error), "Timeout in request queue")
+                self.triggers.popleft()()
+                fut1.add_done_callback(self.stop)
+                self.wait()
+                fut1.result()
 
     def test_no_content_length(self):
         response = self.fetch("/no_content_length")
@@ -500,25 +507,25 @@ class SimpleHTTPSClientTestCase(SimpleHTTPClientTestMixin, AsyncHTTPSTestCase):
     def test_ssl_options_handshake_fail(self):
         with ExpectLog(gen_log, "SSL Error|Uncaught exception",
                        required=False):
-            resp = self.fetch(
-                "/hello", ssl_options=dict(cert_reqs=ssl.CERT_REQUIRED))
-        self.assertRaises(ssl.SSLError, resp.rethrow)
+            with self.assertRaises(ssl.SSLError):
+                self.fetch(
+                    "/hello", ssl_options=dict(cert_reqs=ssl.CERT_REQUIRED),
+                    raise_error=True)
 
     def test_ssl_context_handshake_fail(self):
         with ExpectLog(gen_log, "SSL Error|Uncaught exception"):
             ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
             ctx.verify_mode = ssl.CERT_REQUIRED
-            resp = self.fetch("/hello", ssl_options=ctx)
-        self.assertRaises(ssl.SSLError, resp.rethrow)
+            with self.assertRaises(ssl.SSLError):
+                self.fetch("/hello", ssl_options=ctx, raise_error=True)
 
     def test_error_logging(self):
         # No stack traces are logged for SSL errors (in this case,
         # failure to validate the testing self-signed cert).
         # The SSLError is exposed through ssl.SSLError.
         with ExpectLog(gen_log, '.*') as expect_log:
-            response = self.fetch("/", validate_cert=True)
-            self.assertEqual(response.code, 599)
-            self.assertIsInstance(response.error, ssl.SSLError)
+            with self.assertRaises(ssl.SSLError):
+                self.fetch("/", validate_cert=True, raise_error=True)
         self.assertFalse(expect_log.logged_stack)
 
 
@@ -619,7 +626,8 @@ class HTTP204NoContentTestCase(AsyncHTTPTestCase):
     def test_204_invalid_content_length(self):
         # 204 status with non-zero content length is malformed
         with ExpectLog(gen_log, ".*Response with code 204 should not have body"):
-            response = self.fetch("/?error=1")
+            with ignore_deprecation():
+                response = self.fetch("/?error=1")
             if not self.http1:
                 self.skipTest("requires HTTP/1.x")
             if self.http_client.configured_class != SimpleAsyncHTTPClient:
@@ -640,15 +648,13 @@ class HostnameMappingTestCase(AsyncHTTPTestCase):
         return Application([url("/hello", HelloWorldHandler), ])
 
     def test_hostname_mapping(self):
-        self.http_client.fetch(
-            'http://www.example.com:%d/hello' % self.get_http_port(), self.stop)
-        response = self.wait()
+        response = self.fetch(
+            'http://www.example.com:%d/hello' % self.get_http_port())
         response.rethrow()
         self.assertEqual(response.body, b'Hello world!')
 
     def test_port_mapping(self):
-        self.http_client.fetch('http://foo.example.com:8000/hello', self.stop)
-        response = self.wait()
+        response = self.fetch('http://foo.example.com:8000/hello')
         response.rethrow()
         self.assertEqual(response.body, b'Hello world!')
 
@@ -668,8 +674,8 @@ class ResolveTimeoutTestCase(AsyncHTTPTestCase):
         return Application([url("/hello", HelloWorldHandler), ])
 
     def test_resolve_timeout(self):
-        response = self.fetch('/hello', connect_timeout=0.1)
-        self.assertEqual(response.code, 599)
+        with self.assertRaises(TypeError):
+            self.fetch('/hello', connect_timeout=0.1, raise_error=True)
 
 
 class MaxHeaderSizeTest(AsyncHTTPTestCase):
@@ -697,8 +703,8 @@ class MaxHeaderSizeTest(AsyncHTTPTestCase):
 
     def test_large_headers(self):
         with ExpectLog(gen_log, "Unsatisfiable read"):
-            response = self.fetch('/large')
-        self.assertEqual(response.code, 599)
+            with self.assertRaises(UnsatisfiableReadError):
+                self.fetch('/large', raise_error=True)
 
 
 class MaxBodySizeTest(AsyncHTTPTestCase):
@@ -724,8 +730,8 @@ class MaxBodySizeTest(AsyncHTTPTestCase):
 
     def test_large_body(self):
         with ExpectLog(gen_log, "Malformed HTTP message from None: Content-Length too long"):
-            response = self.fetch('/large')
-        self.assertEqual(response.code, 599)
+            with self.assertRaises(HTTPStreamClosedError):
+                self.fetch('/large', raise_error=True)
 
 
 class MaxBufferSizeTest(AsyncHTTPTestCase):
@@ -765,5 +771,5 @@ class ChunkedWithContentLengthTest(AsyncHTTPTestCase):
         # Make sure the invalid headers are detected
         with ExpectLog(gen_log, ("Malformed HTTP message from None: Response "
                                  "with both Transfer-Encoding and Content-Length")):
-            response = self.fetch('/chunkwithcl')
-        self.assertEqual(response.code, 599)
+            with self.assertRaises(HTTPStreamClosedError):
+                self.fetch('/chunkwithcl', raise_error=True)
diff --git a/tornado/test/stack_context_test.py b/tornado/test/stack_context_test.py
index 4afe2a86..243ea47f 100644
--- a/tornado/test/stack_context_test.py
+++ b/tornado/test/stack_context_test.py
@@ -6,7 +6,7 @@ from tornado.log import app_log
 from tornado.stack_context import (StackContext, wrap, NullContext, StackContextInconsistentError,
                                    ExceptionStackContext, run_with_stack_context, _state)
 from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, ExpectLog, gen_test
-from tornado.test.util import unittest
+from tornado.test.util import unittest, ignore_deprecation
 from tornado.web import asynchronous, Application, RequestHandler
 import contextlib
 import functools
@@ -48,8 +48,9 @@ class HTTPStackContextTest(AsyncHTTPTestCase):
 
     def test_stack_context(self):
         with ExpectLog(app_log, "Uncaught exception GET /"):
-            self.http_client.fetch(self.get_url('/'), self.handle_response)
-            self.wait()
+            with ignore_deprecation():
+                self.http_client.fetch(self.get_url('/'), self.handle_response)
+                self.wait()
         self.assertEqual(self.response.code, 500)
         self.assertTrue(b'got expected exception' in self.response.body)
 
diff --git a/tornado/test/tcpclient_test.py b/tornado/test/tcpclient_test.py
index c5a83342..d7de4223 100644
--- a/tornado/test/tcpclient_test.py
+++ b/tornado/test/tcpclient_test.py
@@ -77,7 +77,7 @@ class TCPClientTest(AsyncTestCase):
     def skipIfLocalhostV4(self):
         # The port used here doesn't matter, but some systems require it
         # to be non-zero if we do not also pass AI_PASSIVE.
-        addrinfo = yield Resolver().resolve('localhost', 80)
+        addrinfo = self.io_loop.run_sync(lambda: Resolver().resolve('localhost', 80))
         families = set(addr[0] for addr in addrinfo)
         if socket.AF_INET6 not in families:
             self.skipTest("localhost does not resolve to ipv6")
diff --git a/tornado/test/testing_test.py b/tornado/test/testing_test.py
index 37449466..eb3445e9 100644
--- a/tornado/test/testing_test.py
+++ b/tornado/test/testing_test.py
@@ -2,7 +2,7 @@ from __future__ import absolute_import, division, print_function
 
 from tornado import gen, ioloop
 from tornado.log import app_log
-from tornado.test.util import unittest, skipBefore35, exec_test
+from tornado.test.util import unittest, skipBefore35, exec_test, ignore_deprecation
 from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, bind_unused_port, gen_test, ExpectLog
 from tornado.web import Application
 import contextlib
@@ -99,13 +99,15 @@ class AsyncHTTPTestCaseTest(AsyncHTTPTestCase):
     def test_fetch_full_http_url(self):
         path = 'http://localhost:%d/path' % self.external_port
 
-        response = self.fetch(path, request_timeout=0.1)
+        with ignore_deprecation():
+            response = self.fetch(path, request_timeout=0.1, raise_error=False)
         self.assertEqual(response.request.url, path)
 
     def test_fetch_full_https_url(self):
         path = 'https://localhost:%d/path' % self.external_port
 
-        response = self.fetch(path, request_timeout=0.1)
+        with ignore_deprecation():
+            response = self.fetch(path, request_timeout=0.1)
         self.assertEqual(response.request.url, path)
 
     @classmethod
diff --git a/tornado/test/web_test.py b/tornado/test/web_test.py
index 6673a4f2..7f053ef0 100644
--- a/tornado/test/web_test.py
+++ b/tornado/test/web_test.py
@@ -3,6 +3,7 @@ from __future__ import absolute_import, division, print_function
 from tornado.concurrent import Future
 from tornado import gen
 from tornado.escape import json_decode, utf8, to_unicode, recursive_unicode, native_str, to_basestring  # noqa: E501
+from tornado.httpclient import HTTPClientError
 from tornado.httputil import format_timestamp
 from tornado.ioloop import IOLoop
 from tornado.iostream import IOStream
@@ -350,16 +351,14 @@ class AuthRedirectTest(WebTestCase):
                  dict(login_url='http://example.com/login'))]
 
     def test_relative_auth_redirect(self):
-        self.http_client.fetch(self.get_url('/relative'), self.stop,
-                               follow_redirects=False)
-        response = self.wait()
+        response = self.fetch(self.get_url('/relative'),
+                              follow_redirects=False)
         self.assertEqual(response.code, 302)
         self.assertEqual(response.headers['Location'], '/login?next=%2Frelative')
 
     def test_absolute_auth_redirect(self):
-        self.http_client.fetch(self.get_url('/absolute'), self.stop,
-                               follow_redirects=False)
-        response = self.wait()
+        response = self.fetch(self.get_url('/absolute'),
+                              follow_redirects=False)
         self.assertEqual(response.code, 302)
         self.assertTrue(re.match(
             'http://example.com/login\?next=http%3A%2F%2F127.0.0.1%3A[0-9]+%2Fabsolute',
@@ -2334,8 +2333,8 @@ class IncorrectContentLengthTest(SimpleHandlerTestCase):
             with ExpectLog(gen_log,
                            "(Cannot send error response after headers written"
                            "|Failed to flush partial response)"):
-                response = self.fetch("/high")
-        self.assertEqual(response.code, 599)
+                with self.assertRaises(HTTPClientError):
+                    self.fetch("/high", raise_error=True)
         self.assertEqual(str(self.server_error),
                          "Tried to write 40 bytes less than Content-Length")
 
@@ -2347,8 +2346,8 @@ class IncorrectContentLengthTest(SimpleHandlerTestCase):
             with ExpectLog(gen_log,
                            "(Cannot send error response after headers written"
                            "|Failed to flush partial response)"):
-                response = self.fetch("/low")
-        self.assertEqual(response.code, 599)
+                with self.assertRaises(HTTPClientError):
+                    self.fetch("/low", raise_error=True)
         self.assertEqual(str(self.server_error),
                          "Tried to write more data than Content-Length")
 
@@ -2369,7 +2368,8 @@ class ClientCloseTest(SimpleHandlerTestCase):
                 self.write('requires HTTP/1.x')
 
     def test_client_close(self):
-        response = self.fetch('/')
+        with ignore_deprecation():
+            response = self.fetch('/')
         if response.body == b'requires HTTP/1.x':
             self.skipTest('requires HTTP/1.x')
         self.assertEqual(response.code, 599)
diff --git a/tornado/testing.py b/tornado/testing.py
index 19bc5a94..04ea3816 100644
--- a/tornado/testing.py
+++ b/tornado/testing.py
@@ -80,6 +80,7 @@ else:
     import tornado.platform.asyncio
     _NON_OWNED_IOLOOPS = tornado.platform.asyncio.AsyncIOMainLoop
 
+
 def bind_unused_port(reuse_port=False):
     """Binds a server socket to an available port on localhost.
 
@@ -386,7 +387,7 @@ class AsyncHTTPTestCase(AsyncTestCase):
         """
         raise NotImplementedError()
 
-    def fetch(self, path, **kwargs):
+    def fetch(self, path, raise_error=False, **kwargs):
         """Convenience method to synchronously fetch a URL.
 
         The given path will be appended to the local server's host and
@@ -397,14 +398,36 @@ class AsyncHTTPTestCase(AsyncTestCase):
         If the path begins with http:// or https://, it will be treated as a
         full URL and will be fetched as-is.
 
+        If ``raise_error`` is True, a `tornado.httpclient.HTTPError` will
+        be raised if the response code is not 200. This is the same behavior
+        as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but
+        the default is False here (it's True in `.AsyncHTTPClient`) because
+        tests often need to deal with non-200 response codes.
+
         .. versionchanged:: 5.0
            Added support for absolute URLs.
+
+        .. versionchanged:: 5.1
+
+           Added the ``raise_error`` argument.
+
+        .. deprecated:: 5.1
+
+           This method currently turns any exception into an
+           `.HTTPResponse` with status code 599. In Tornado 6.0,
+           errors other than `tornado.httpclient.HTTPError` will be
+           passed through, and ``raise_error=False`` will only
+           suppress errors that would be raised due to non-200
+           response codes.
+
         """
         if path.lower().startswith(('http://', 'https://')):
-            self.http_client.fetch(path, self.stop, **kwargs)
+            url = path
         else:
-            self.http_client.fetch(self.get_url(path), self.stop, **kwargs)
-        return self.wait()
+            url = self.get_url(path)
+        return self.io_loop.run_sync(
+            lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),
+            timeout=get_async_test_timeout())
 
     def get_httpserver_options(self):
         """May be overridden by subclasses to return additional