mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-02 08:02:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Merge 3.5 (asyncio)

Browse source
This commit is contained in:
Yury Selivanov 2016-06-02 16:51:27 -04:00
commit 1f56e5f6af
2 changed files with 39 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

22
Lib/asyncio/base_events.py
View file

@ -102,10 +102,26 @@ def _ipaddr_info(host, port, family, type, proto):
else: else:
return None return None
if port in {None, '', b''}: if port is None:
port = 0 port = 0
elif isinstance(port, (bytes, str)): elif isinstance(port, bytes):
port = int(port) if port == b'':
port = 0
else:
try:
port = int(port)
except ValueError:
# Might be a service name like b"http".
port = socket.getservbyname(port.decode('ascii'))
elif isinstance(port, str):
if port == '':
port = 0
else:
try:
port = int(port)
except ValueError:
# Might be a service name like "http".
port = socket.getservbyname(port)
if hasattr(socket, 'inet_pton'): if hasattr(socket, 'inet_pton'):
if family == socket.AF_UNSPEC: if family == socket.AF_UNSPEC:

20
Lib/test/test_asyncio/test_base_events.py
View file

@ -146,6 +146,26 @@ class BaseEventTests(test_utils.TestCase):
(INET, STREAM, TCP, '', ('1.2.3.4', 1)), (INET, STREAM, TCP, '', ('1.2.3.4', 1)),
base_events._ipaddr_info('1.2.3.4', b'1', INET, STREAM, TCP)) base_events._ipaddr_info('1.2.3.4', b'1', INET, STREAM, TCP))
def test_getaddrinfo_servname(self):
INET = socket.AF_INET
STREAM = socket.SOCK_STREAM
TCP = socket.IPPROTO_TCP
self.assertEqual(
(INET, STREAM, TCP, '', ('1.2.3.4', 80)),
base_events._ipaddr_info('1.2.3.4', 'http', INET, STREAM, TCP))
self.assertEqual(
(INET, STREAM, TCP, '', ('1.2.3.4', 80)),
base_events._ipaddr_info('1.2.3.4', b'http', INET, STREAM, TCP))
# Raises "service/proto not found".
with self.assertRaises(OSError):
base_events._ipaddr_info('1.2.3.4', 'nonsense', INET, STREAM, TCP)
with self.assertRaises(OSError):
base_events._ipaddr_info('1.2.3.4', 'nonsense', INET, STREAM, TCP)
@patch_socket @patch_socket
def test_ipaddr_info_no_inet_pton(self, m_socket): def test_ipaddr_info_no_inet_pton(self, m_socket):
del m_socket.inet_pton del m_socket.inet_pton