class CaresResolver(Resolver):
"""Name resolver based on the c-ares library.
- This is a non-blocking and non-threaded resolver. It may not produce
- the same results as the system resolver, but can be used for non-blocking
+ This is a non-blocking and non-threaded resolver. It may not produce the
+ same results as the system resolver, but can be used for non-blocking
resolution when threads cannot be used.
- c-ares fails to resolve some names when ``family`` is ``AF_UNSPEC``,
- so it is only recommended for use in ``AF_INET`` (i.e. IPv4). This is
- the default for ``tornado.simple_httpclient``, but other libraries
- may default to ``AF_UNSPEC``.
+ ``pycares`` will not return a mix of ``AF_INET`` and ``AF_INET6`` when
+ ``family`` is ``AF_UNSPEC``, so it is only recommended for use in
+ ``AF_INET`` (i.e. IPv4). This is the default for
+ ``tornado.simple_httpclient``, but other libraries may default to
+ ``AF_UNSPEC``.
.. versionchanged:: 5.0
The ``io_loop`` argument (deprecated since version 4.1) has been removed.
@gen_test
def test_localhost(self: typing.Any):
addrinfo = yield self.resolver.resolve("localhost", 80, socket.AF_UNSPEC)
- self.assertIn((socket.AF_INET, ("127.0.0.1", 80)), addrinfo)
+ # Most of the time localhost resovles to either the ipv4 loopback
+ # address alone, or ipv4+ipv6. But some versions of pycares will only
+ # return the ipv6 version, so we have to check for either one alone.
+ self.assertTrue(
+ ((socket.AF_INET, ("127.0.0.1", 80)) in addrinfo)
+ or ((socket.AF_INET6, ("::1", 80)) in addrinfo),
+ f"loopback address not found in {addrinfo}",
+ )
# It is impossible to quickly and consistently generate an error in name
def do_test_connect(self, family, host, source_ip=None, source_port=None):
port = self.start_server(family)
stream = yield self.client.connect(
- host, port, source_ip=source_ip, source_port=source_port
+ host,
+ port,
+ source_ip=source_ip,
+ source_port=source_port,
+ af=family,
)
assert self.server is not None
server_stream = yield self.server.queue.get()
def test_fetch_full_http_url(self):
# Ensure that self.fetch() recognizes absolute urls and does
# not transform them into references to our main test server.
- path = "http://localhost:%d/path" % self.second_port
+ path = "http://127.0.0.1:%d/path" % self.second_port
response = self.fetch(path)
self.assertEqual(response.request.url, path)
import asyncio
import functools
+import socket
import traceback
import typing
import unittest
from tornado.httpclient import HTTPError, HTTPRequest
from tornado.locks import Event
from tornado.log import gen_log, app_log
+from tornado.netutil import Resolver
from tornado.simple_httpclient import SimpleAsyncHTTPClient
from tornado.template import DictLoader
from tornado.testing import AsyncHTTPTestCase, gen_test, bind_unused_port, ExpectLog
def test_check_origin_invalid_subdomains(self):
port = self.get_http_port()
+ # CaresResolver may return ipv6-only results for localhost, but our
+ # server is only running on ipv4. Test for this edge case and skip
+ # the test if it happens.
+ addrinfo = yield Resolver().resolve("localhost", port)
+ families = set(addr[0] for addr in addrinfo)
+ if socket.AF_INET not in families:
+ self.skipTest("localhost does not resolve to ipv4")
+ return
+
url = "ws://localhost:%d/echo" % port
# Subdomains should be disallowed by default. If we could pass a
# resolver to websocket_connect we could test sibling domains as well.