import collections
import concurrent.futures
+import functools
import heapq
import inspect
+import ipaddress
import itertools
import logging
import os
return repr(fd)
+# Linux's sock.type is a bitmask that can include extra info about socket.
+_SOCKET_TYPE_MASK = 0
+if hasattr(socket, 'SOCK_NONBLOCK'):
+ _SOCKET_TYPE_MASK |= socket.SOCK_NONBLOCK
+if hasattr(socket, 'SOCK_CLOEXEC'):
+ _SOCKET_TYPE_MASK |= socket.SOCK_CLOEXEC
+
+
+@functools.lru_cache(maxsize=1024)
+def _ipaddr_info(host, port, family, type, proto):
+ # Try to skip getaddrinfo if "host" is already an IP. Since getaddrinfo
+ # blocks on an exclusive lock on some platforms, users might handle name
+ # resolution in their own code and pass in resolved IPs.
+ if proto not in {0, socket.IPPROTO_TCP, socket.IPPROTO_UDP} or host is None:
+ return None
+
+ type &= ~_SOCKET_TYPE_MASK
+ if type == socket.SOCK_STREAM:
+ proto = socket.IPPROTO_TCP
+ elif type == socket.SOCK_DGRAM:
+ proto = socket.IPPROTO_UDP
+ else:
+ return None
+
+ if hasattr(socket, 'inet_pton'):
+ if family == socket.AF_UNSPEC:
+ afs = [socket.AF_INET, socket.AF_INET6]
+ else:
+ afs = [family]
+
+ for af in afs:
+ # Linux's inet_pton doesn't accept an IPv6 zone index after host,
+ # like '::1%lo0', so strip it. If we happen to make an invalid
+ # address look valid, we fail later in sock.connect or sock.bind.
+ try:
+ if af == socket.AF_INET6:
+ socket.inet_pton(af, host.partition('%')[0])
+ else:
+ socket.inet_pton(af, host)
+ return af, type, proto, '', (host, port)
+ except OSError:
+ pass
+
+ # "host" is not an IP address.
+ return None
+
+ # No inet_pton. (On Windows it's only available since Python 3.4.)
+ # Even though getaddrinfo with AI_NUMERICHOST would be non-blocking, it
+ # still requires a lock on some platforms, and waiting for that lock could
+ # block the event loop. Use ipaddress instead, it's just text parsing.
+ try:
+ addr = ipaddress.IPv4Address(host)
+ except ValueError:
+ try:
+ addr = ipaddress.IPv6Address(host.partition('%')[0])
+ except ValueError:
+ return None
+
+ af = socket.AF_INET if addr.version == 4 else socket.AF_INET6
+ if family not in (socket.AF_UNSPEC, af):
+ # "host" is wrong IP version for "family".
+ return None
+
+ return af, type, proto, '', (host, port)
+
+
def _check_resolved_address(sock, address):
# Ensure that the address is already resolved to avoid the trap of hanging
# the entire event loop when the address requires doing a DNS lookup.
- #
- # getaddrinfo() is slow (around 10 us per call): this function should only
- # be called in debug mode
- family = sock.family
-
- if family == socket.AF_INET:
- host, port = address
- elif family == socket.AF_INET6:
- host, port = address[:2]
- else:
+
+ if hasattr(socket, 'AF_UNIX') and sock.family == socket.AF_UNIX:
return
- # On Windows, socket.inet_pton() is only available since Python 3.4
- if hasattr(socket, 'inet_pton'):
- # getaddrinfo() is slow and has known issue: prefer inet_pton()
- # if available
- try:
- socket.inet_pton(family, host)
- except OSError as exc:
- raise ValueError("address must be resolved (IP address), "
- "got host %r: %s"
- % (host, exc))
- else:
- # Use getaddrinfo(flags=AI_NUMERICHOST) to ensure that the address is
- # already resolved.
- type_mask = 0
- if hasattr(socket, 'SOCK_NONBLOCK'):
- type_mask |= socket.SOCK_NONBLOCK
- if hasattr(socket, 'SOCK_CLOEXEC'):
- type_mask |= socket.SOCK_CLOEXEC
- try:
- socket.getaddrinfo(host, port,
- family=family,
- type=(sock.type & ~type_mask),
- proto=sock.proto,
- flags=socket.AI_NUMERICHOST)
- except socket.gaierror as err:
- raise ValueError("address must be resolved (IP address), "
- "got host %r: %s"
- % (host, err))
+ host, port = address[:2]
+ if _ipaddr_info(host, port, sock.family, sock.type, sock.proto) is None:
+ raise ValueError("address must be resolved (IP address),"
+ " got host %r" % host)
def _run_until_complete_cb(fut):
def getaddrinfo(self, host, port, *,
family=0, type=0, proto=0, flags=0):
- if self._debug:
+ info = _ipaddr_info(host, port, family, type, proto)
+ if info is not None:
+ fut = futures.Future(loop=self)
+ fut.set_result([info])
+ return fut
+ elif self._debug:
return self.run_in_executor(None, self._getaddrinfo_debug,
host, port, family, type, proto, flags)
else:
PY34 = sys.version_info >= (3, 4)
+def mock_socket_module():
+ m_socket = mock.MagicMock(spec=socket)
+ for name in (
+ 'AF_INET', 'AF_INET6', 'AF_UNSPEC', 'IPPROTO_TCP', 'IPPROTO_UDP',
+ 'SOCK_STREAM', 'SOCK_DGRAM', 'SOL_SOCKET', 'SO_REUSEADDR', 'inet_pton'
+ ):
+ if hasattr(socket, name):
+ setattr(m_socket, name, getattr(socket, name))
+ else:
+ delattr(m_socket, name)
+
+ m_socket.socket = mock.MagicMock()
+ m_socket.socket.return_value = test_utils.mock_nonblocking_socket()
+
+ return m_socket
+
+
+def patch_socket(f):
+ return mock.patch('asyncio.base_events.socket',
+ new_callable=mock_socket_module)(f)
+
+
+class BaseEventTests(test_utils.TestCase):
+
+ def setUp(self):
+ super().setUp()
+ base_events._ipaddr_info.cache_clear()
+
+ def tearDown(self):
+ base_events._ipaddr_info.cache_clear()
+ super().tearDown()
+
+ def test_ipaddr_info(self):
+ UNSPEC = socket.AF_UNSPEC
+ INET = socket.AF_INET
+ INET6 = socket.AF_INET6
+ STREAM = socket.SOCK_STREAM
+ DGRAM = socket.SOCK_DGRAM
+ TCP = socket.IPPROTO_TCP
+ UDP = socket.IPPROTO_UDP
+
+ self.assertEqual(
+ (INET, STREAM, TCP, '', ('1.2.3.4', 1)),
+ base_events._ipaddr_info('1.2.3.4', 1, INET, STREAM, TCP))
+
+ self.assertEqual(
+ (INET, STREAM, TCP, '', ('1.2.3.4', 1)),
+ base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, STREAM, TCP))
+
+ self.assertEqual(
+ (INET, DGRAM, UDP, '', ('1.2.3.4', 1)),
+ base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, DGRAM, UDP))
+
+ # Socket type STREAM implies TCP protocol.
+ self.assertEqual(
+ (INET, STREAM, TCP, '', ('1.2.3.4', 1)),
+ base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, STREAM, 0))
+
+ # Socket type DGRAM implies UDP protocol.
+ self.assertEqual(
+ (INET, DGRAM, UDP, '', ('1.2.3.4', 1)),
+ base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, DGRAM, 0))
+
+ # No socket type.
+ self.assertIsNone(
+ base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, 0, 0))
+
+ # IPv4 address with family IPv6.
+ self.assertIsNone(
+ base_events._ipaddr_info('1.2.3.4', 1, INET6, STREAM, TCP))
+
+ self.assertEqual(
+ (INET6, STREAM, TCP, '', ('::3', 1)),
+ base_events._ipaddr_info('::3', 1, INET6, STREAM, TCP))
+
+ self.assertEqual(
+ (INET6, STREAM, TCP, '', ('::3', 1)),
+ base_events._ipaddr_info('::3', 1, UNSPEC, STREAM, TCP))
+
+ # IPv6 address with family IPv4.
+ self.assertIsNone(
+ base_events._ipaddr_info('::3', 1, INET, STREAM, TCP))
+
+ # IPv6 address with zone index.
+ self.assertEqual(
+ (INET6, STREAM, TCP, '', ('::3%lo0', 1)),
+ base_events._ipaddr_info('::3%lo0', 1, INET6, STREAM, TCP))
+
+ @patch_socket
+ def test_ipaddr_info_no_inet_pton(self, m_socket):
+ del m_socket.inet_pton
+ self.test_ipaddr_info()
+
+ def test_check_resolved_address(self):
+ sock = socket.socket(socket.AF_INET)
+ base_events._check_resolved_address(sock, ('1.2.3.4', 1))
+
+ sock = socket.socket(socket.AF_INET6)
+ base_events._check_resolved_address(sock, ('::3', 1))
+ base_events._check_resolved_address(sock, ('::3%lo0', 1))
+ self.assertRaises(ValueError,
+ base_events._check_resolved_address, sock, ('foo', 1))
+
+ def test_check_resolved_sock_type(self):
+ # Ensure we ignore extra flags in sock.type.
+ if hasattr(socket, 'SOCK_NONBLOCK'):
+ sock = socket.socket(type=socket.SOCK_STREAM | socket.SOCK_NONBLOCK)
+ base_events._check_resolved_address(sock, ('1.2.3.4', 1))
+
+ if hasattr(socket, 'SOCK_CLOEXEC'):
+ sock = socket.socket(type=socket.SOCK_STREAM | socket.SOCK_CLOEXEC)
+ base_events._check_resolved_address(sock, ('1.2.3.4', 1))
+
+
class BaseEventLoopTests(test_utils.TestCase):
def setUp(self):
self.loop = asyncio.new_event_loop()
self.set_event_loop(self.loop)
- @mock.patch('asyncio.base_events.socket')
+ def tearDown(self):
+ # Clear mocked constants like AF_INET from the cache.
+ base_events._ipaddr_info.cache_clear()
+ super().tearDown()
+
+ @patch_socket
def test_create_connection_multiple_errors(self, m_socket):
class MyProto(asyncio.Protocol):
self.assertEqual(str(cm.exception), 'Multiple exceptions: err1, err2')
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_connection_timeout(self, m_socket):
# Ensure that the socket is closed on timeout
sock = mock.Mock()
with self.assertRaises(OSError):
self.loop.run_until_complete(coro)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_connection_multiple_errors_local_addr(self, m_socket):
def bind(addr):
self.assertTrue(str(cm.exception).startswith('Multiple exceptions: '))
self.assertTrue(m_socket.socket.return_value.close.called)
+ def _test_create_connection_ip_addr(self, m_socket, allow_inet_pton):
+ # Test the fallback code, even if this system has inet_pton.
+ if not allow_inet_pton:
+ del m_socket.inet_pton
+
+ def getaddrinfo(*args, **kw):
+ self.fail('should not have called getaddrinfo')
+
+ m_socket.getaddrinfo = getaddrinfo
+ sock = m_socket.socket.return_value
+
+ self.loop.add_reader = mock.Mock()
+ self.loop.add_reader._is_coroutine = False
+ self.loop.add_writer = mock.Mock()
+ self.loop.add_writer._is_coroutine = False
+
+ coro = self.loop.create_connection(MyProto, '1.2.3.4', 80)
+ self.loop.run_until_complete(coro)
+ sock.connect.assert_called_with(('1.2.3.4', 80))
+ m_socket.socket.assert_called_with(family=m_socket.AF_INET,
+ proto=m_socket.IPPROTO_TCP,
+ type=m_socket.SOCK_STREAM)
+
+ sock.family = socket.AF_INET6
+ coro = self.loop.create_connection(MyProto, '::2', 80)
+
+ self.loop.run_until_complete(coro)
+ sock.connect.assert_called_with(('::2', 80))
+ m_socket.socket.assert_called_with(family=m_socket.AF_INET6,
+ proto=m_socket.IPPROTO_TCP,
+ type=m_socket.SOCK_STREAM)
+
+ @patch_socket
+ def test_create_connection_ip_addr(self, m_socket):
+ self._test_create_connection_ip_addr(m_socket, True)
+
+ @patch_socket
+ def test_create_connection_no_inet_pton(self, m_socket):
+ self._test_create_connection_ip_addr(m_socket, False)
+
def test_create_connection_no_local_addr(self):
@asyncio.coroutine
def getaddrinfo(host, *args, **kw):
f = self.loop.create_server(MyProto, '0.0.0.0', 0)
self.assertRaises(OSError, self.loop.run_until_complete, f)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_server_nosoreuseport(self, m_socket):
m_socket.getaddrinfo = socket.getaddrinfo
- m_socket.SOCK_STREAM = socket.SOCK_STREAM
- m_socket.SOL_SOCKET = socket.SOL_SOCKET
del m_socket.SO_REUSEPORT
m_socket.socket.return_value = mock.Mock()
self.assertRaises(ValueError, self.loop.run_until_complete, f)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_server_cant_bind(self, m_socket):
class Err(OSError):
self.assertRaises(OSError, self.loop.run_until_complete, fut)
self.assertTrue(m_sock.close.called)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_datagram_endpoint_no_addrinfo(self, m_socket):
m_socket.getaddrinfo.return_value = []
m_socket.getaddrinfo._is_coroutine = False
self.assertRaises(
OSError, self.loop.run_until_complete, coro)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_datagram_endpoint_socket_err(self, m_socket):
m_socket.getaddrinfo = socket.getaddrinfo
m_socket.socket.side_effect = OSError
self.assertRaises(
ValueError, self.loop.run_until_complete, coro)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_datagram_endpoint_setblk_err(self, m_socket):
m_socket.socket.return_value.setblocking.side_effect = OSError
asyncio.DatagramProtocol)
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_datagram_endpoint_cant_bind(self, m_socket):
class Err(OSError):
pass
- m_socket.AF_INET6 = socket.AF_INET6
m_socket.getaddrinfo = socket.getaddrinfo
m_sock = m_socket.socket.return_value = mock.Mock()
m_sock.bind.side_effect = Err
self.loop.run_until_complete(protocol.done)
self.assertEqual('CLOSED', protocol.state)
- @mock.patch('asyncio.base_events.socket')
+ @patch_socket
def test_create_datagram_endpoint_nosoreuseport(self, m_socket):
- m_socket.getaddrinfo = socket.getaddrinfo
- m_socket.SOCK_DGRAM = socket.SOCK_DGRAM
- m_socket.SOL_SOCKET = socket.SOL_SOCKET
del m_socket.SO_REUSEPORT
m_socket.socket.return_value = mock.Mock()
self.assertRaises(ValueError, self.loop.run_until_complete, coro)
+ @patch_socket
+ def test_create_datagram_endpoint_ip_addr(self, m_socket):
+ def getaddrinfo(*args, **kw):
+ self.fail('should not have called getaddrinfo')
+
+ m_socket.getaddrinfo = getaddrinfo
+ m_socket.socket.return_value.bind = bind = mock.Mock()
+ self.loop.add_reader = mock.Mock()
+ self.loop.add_reader._is_coroutine = False
+
+ reuseport_supported = hasattr(socket, 'SO_REUSEPORT')
+ coro = self.loop.create_datagram_endpoint(
+ lambda: MyDatagramProto(loop=self.loop),
+ local_addr=('1.2.3.4', 0),
+ reuse_address=False,
+ reuse_port=reuseport_supported)
+
+ self.loop.run_until_complete(coro)
+ bind.assert_called_with(('1.2.3.4', 0))
+ m_socket.socket.assert_called_with(family=m_socket.AF_INET,
+ proto=m_socket.IPPROTO_UDP,
+ type=m_socket.SOCK_DGRAM)
+
def test_accept_connection_retry(self):
sock = mock.Mock()
sock.accept.side_effect = BlockingIOError()
projects / python / commitdiff