summaryrefslogtreecommitdiff
path: root/Lib/test/test_asyncio
diff options
context:
space:
mode:
authorErwan Le Pape <lepaperwan@users.noreply.github.com>2019-05-17 10:28:39 +0200
committerMiss Islington (bot) <31488909+miss-islington@users.noreply.github.com>2019-05-17 01:28:39 -0700
commitac8eb8f36bf7ca636f8d886eb65a3b532f4725d5 (patch)
treeee22645473384845c6a8911ef3061cdc1fca3766 /Lib/test/test_asyncio
parent14514d9084a40f599c57da853a305aa264562a43 (diff)
downloadcpython-git-ac8eb8f36bf7ca636f8d886eb65a3b532f4725d5.tar.gz
bpo-35545: Fix asyncio discarding IPv6 scopes (GH-11271)
This PR proposes a solution to [bpo-35545](https://bugs.python.org/issue35545) by adding an optional `flowinfo` and `scopeid` to `asyncio.base_events._ipaddr_info` to carry the full address information into `_ipaddr_info` and avoid discarding IPv6 specific information. Changelog entry & regression tests to come. https://bugs.python.org/issue35545
Diffstat (limited to 'Lib/test/test_asyncio')
-rw-r--r--Lib/test/test_asyncio/test_base_events.py22
1 files changed, 22 insertions, 0 deletions
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py
index 27e21b33d3..f068fc781f 100644
--- a/Lib/test/test_asyncio/test_base_events.py
+++ b/Lib/test/test_asyncio/test_base_events.py
@@ -1300,6 +1300,28 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase):
test_utils.run_briefly(self.loop) # allow transport to close
@patch_socket
+ def test_create_connection_ipv6_scope(self, m_socket):
+ m_socket.getaddrinfo = socket.getaddrinfo
+ sock = m_socket.socket.return_value
+ sock.family = socket.AF_INET6
+
+ 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(asyncio.Protocol, 'fe80::1%1', 80)
+ t, p = self.loop.run_until_complete(coro)
+ try:
+ sock.connect.assert_called_with(('fe80::1', 80, 0, 1))
+ _, kwargs = m_socket.socket.call_args
+ self.assertEqual(kwargs['family'], m_socket.AF_INET6)
+ self.assertEqual(kwargs['type'], m_socket.SOCK_STREAM)
+ finally:
+ t.close()
+ test_utils.run_briefly(self.loop) # allow transport to close
+
+ @patch_socket
def test_create_connection_ip_addr(self, m_socket):
self._test_create_connection_ip_addr(m_socket, True)