mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-07-24 11:44:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

GH-86508: skip binding to local addresses of different family in asyncio.open_connection (#100615)

Browse source
This commit is contained in:
Kumar Aditya 2023-01-04 13:30:26 +05:30 • committed by GitHub
parent a286caa937
commit ba8dcdbcab
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 50 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

41
Lib/test/test_asyncio/test_events.py
View file

@ -670,6 +670,47 @@ class EventLoopTestsMixin:
self.assertEqual(port, expected)
tr.close()
def test_create_connection_local_addr_skip_different_family(self):
# See https://github.com/python/cpython/issues/86508
port1 = socket_helper.find_unused_port()
port2 = socket_helper.find_unused_port()
getaddrinfo_orig = self.loop.getaddrinfo
async def getaddrinfo(host, port, *args, **kwargs):
if port == port2:
return [(socket.AF_INET6, socket.SOCK_STREAM, 0, '', ('::1', 0, 0, 0)),
(socket.AF_INET, socket.SOCK_STREAM, 0, '', ('127.0.0.1', 0))]
return await getaddrinfo_orig(host, port, *args, **kwargs)
self.loop.getaddrinfo = getaddrinfo
f = self.loop.create_connection(
lambda: MyProto(loop=self.loop),
'localhost', port1, local_addr=('localhost', port2))
with self.assertRaises(OSError):
self.loop.run_until_complete(f)
def test_create_connection_local_addr_nomatch_family(self):
# See https://github.com/python/cpython/issues/86508
port1 = socket_helper.find_unused_port()
port2 = socket_helper.find_unused_port()
getaddrinfo_orig = self.loop.getaddrinfo
async def getaddrinfo(host, port, *args, **kwargs):
if port == port2:
return [(socket.AF_INET6, socket.SOCK_STREAM, 0, '', ('::1', 0, 0, 0))]
return await getaddrinfo_orig(host, port, *args, **kwargs)
self.loop.getaddrinfo = getaddrinfo
f = self.loop.create_connection(
lambda: MyProto(loop=self.loop),
'localhost', port1, local_addr=('localhost', port2))
with self.assertRaises(OSError):
self.loop.run_until_complete(f)
def test_create_connection_local_addr_in_use(self):
with test_utils.run_test_server() as httpd:
f = self.loop.create_connection(