Merge pull request #425 from ffix/master

Added IPv6 address support in ClientRequest.update_host

Author: asvetlov

aiohttp/client_reqrep.py

@@ -97,41 +97,48 @@ def __init__(self, method, url, *,
 
     def update_host(self, url):
         """Update destination host, port and connection type (ssl)."""
-        scheme, netloc, path, query, fragment = urllib.parse.urlsplit(url)
+        url_parsed = urllib.parse.urlsplit(url)
+
+        # check for network location part
+        netloc = url_parsed.netloc
         if not netloc:
             raise ValueError('Host could not be detected.')
 
+        # get host/port
+        host = url_parsed.hostname
+        try:
+            port = url_parsed.port
+        except ValueError:
+            raise ValueError(
+                'Port number could not be converted.') from None
+
         # check domain idna encoding
         try:
             netloc = netloc.encode('idna').decode('utf-8')
+            host = host.encode('idna').decode('utf-8')
         except UnicodeError:
             raise ValueError('URL has an invalid label.')
 
         # basic auth info
-        if '@' in netloc:
-            authinfo, netloc = netloc.split('@', 1)
-            self.auth = helpers.BasicAuth(*authinfo.split(':', 1))
+        username, password = url_parsed.username, url_parsed.password
+        if username:
+            self.auth = helpers.BasicAuth(username, password or '')
+            netloc = netloc.split('@', 1)[1]
 
         # Record entire netloc for usage in host header
         self.netloc = netloc
 
-        # extract host and port
+        scheme = url_parsed.scheme
         self.ssl = scheme == 'https'
-        if ':' in netloc:
-            netloc, port_s = netloc.split(':', 1)
-            try:
-                self.port = int(port_s)
-            except ValueError:
-                raise ValueError(
-                    'Port number could not be converted.') from None
-        else:
+
+        # set port number if it isn't already set
+        if not port:
             if self.ssl:
-                self.port = HTTPS_PORT
+                port = HTTPS_PORT
             else:
-                self.port = HTTP_PORT
+                port = HTTP_PORT
 
-        self.scheme = scheme
-        self.host = netloc
+        self.host, self.port, self.scheme = host, port, scheme
 
     def update_version(self, version):
         """Convert request version to two elements tuple.

tests/test_client_request.py

@@ -135,6 +135,25 @@ def test_invalid_idna(self):
             ValueError, ClientRequest, 'get', 'http://\u2061owhefopw.com',
             loop=self.loop)
 
+    def test_ipv6_host_port(self):
+        req = ClientRequest('get', 'http://[2001:db8::1]/', loop=self.loop)
+        self.assertEqual(req.host, '2001:db8::1')
+        self.assertEqual(req.port, 80)
+        self.assertFalse(req.ssl)
+        self.loop.run_until_complete(req.close())
+
+        req = ClientRequest('get', 'https://[2001:db8::1]', loop=self.loop)
+        self.assertEqual(req.host, '2001:db8::1')
+        self.assertEqual(req.port, 443)
+        self.assertTrue(req.ssl)
+        self.loop.run_until_complete(req.close())
+
+        req = ClientRequest('get', 'https://[2001:db8::1]:960', loop=self.loop)
+        self.assertEqual(req.host, '2001:db8::1')
+        self.assertEqual(req.port, 960)
+        self.assertTrue(req.ssl)
+        self.loop.run_until_complete(req.close())
+
     def test_no_path(self):
         req = ClientRequest('get', 'http://python.org', loop=self.loop)
         self.assertEqual('/', req.path)
@@ -172,6 +191,7 @@ def test_basic_auth_from_url(self):
                             loop=self.loop)
         self.assertIn('AUTHORIZATION', req.headers)
         self.assertEqual('Basic bmtpbToxMjM0', req.headers['AUTHORIZATION'])
+        self.assertEqual('python.org', req.netloc)
         self.loop.run_until_complete(req.close())
 
         req = ClientRequest(
@@ -180,6 +200,7 @@ def test_basic_auth_from_url(self):
             loop=self.loop)
         self.assertIn('AUTHORIZATION', req.headers)
         self.assertEqual('Basic bmtpbToxMjM0', req.headers['AUTHORIZATION'])
+        self.assertEqual('python.org', req.netloc)
         self.loop.run_until_complete(req.close())
 
     def test_no_content_length(self):