Better BasicAuth tuple

aiohttp/client.py

@@ -1,9 +1,8 @@
 """HTTP Client for asyncio."""
 
-__all__ = ['request', 'HttpClient', 'BasicAuth', 'BasicAuthEx']
+__all__ = ['request', 'HttpClient']
 
 import asyncio
-import base64
 import collections
 import http.cookies
 import json
@@ -29,10 +28,6 @@
 HTTP_PORT = 80
 HTTPS_PORT = 443
 
-BasicAuth = collections.namedtuple('BasicAuth', ['login', 'password'])
-BasicAuthEx = collections.namedtuple('BasicAuthEx',
-                                     ['login', 'password', 'encoding'])
-
 
 @asyncio.coroutine
 def request(method, url, *,
@@ -227,11 +222,7 @@ def update_host(self, url):
         # basic auth info
         if '@' in netloc:
             authinfo, netloc = netloc.split('@', 1)
-            creds = authinfo.split(':', 1)
-            if len(creds) > 1:
-                self.auth = BasicAuth(creds[0], creds[1])
-            else:
-                self.auth = BasicAuth(creds[0], '')
+            self.auth = helpers.BasicAuth(*authinfo.split(':', 1))
 
         # Record entire netloc for usage in host header
         self.netloc = netloc
@@ -356,24 +347,15 @@ def update_auth(self, auth):
         if auth is None:
             return
 
-        if not isinstance(auth, (BasicAuth, BasicAuthEx)):
+        if not isinstance(auth, helpers.BasicAuth):
             warnings.warn(
-                'BasicAuth() or BasicAuthEx() tuple is required instead ',
-                DeprecationWarning)
+                'BasicAuth() tuple is required instead ', DeprecationWarning)
 
-        if isinstance(auth, BasicAuthEx):
-            basic_login, basic_passwd, encoding = auth
-        else:
-            basic_login, basic_passwd = auth
-            encoding = 'latin1'
-
-        if basic_login is not None and basic_passwd is not None:
-            self.headers['AUTHORIZATION'] = 'Basic %s' % (
-                base64.b64encode(
-                    ('%s:%s' % (basic_login, basic_passwd)).encode(encoding))
-                .strip().decode(encoding))
-        elif basic_login is not None or basic_passwd is not None:
-            raise ValueError("HTTP Auth login or password is missing")
+        if not isinstance(auth, helpers.BasicAuth):
+            auth = helpers.BasicAuth(*auth)
+
+        if auth.login:
+            self.headers['AUTHORIZATION'] = auth.encode()
 
     def update_body_from_data(self, data):
         if isinstance(data, str):

aiohttp/connector.py

@@ -10,9 +10,10 @@
 import socket
 import weakref
 
+from .client import ClientRequest
 from .errors import HttpProxyError
 from .errors import ProxyConnectionError
-from .client import ClientRequest, BasicAuth, BasicAuthEx
+from .helpers import BasicAuth
 
 
 class Connection(object):
@@ -285,10 +286,8 @@ def __init__(self, proxy, *args, proxy_auth=None, **kwargs):
         self._proxy_auth = proxy_auth
         assert proxy.startswith('http://'), (
             "Only http proxy supported", proxy)
-        assert (proxy_auth is None
-                or isinstance(proxy_auth, (BasicAuth, BasicAuthEx))), \
-            ("proxy_auth must be None, BasicAuth() or BasicAuthEx()",
-             proxy_auth)
+        assert proxy_auth is None or isinstance(proxy_auth, BasicAuth), (
+            "proxy_auth must be None or BasicAuth() tuple", proxy_auth)
 
     @property
     def proxy(self):

aiohttp/helpers.py

@@ -1,8 +1,24 @@
 """Various helper functions"""
+import base64
 import io
 import os
 import uuid
 import urllib.parse
+from collections import namedtuple
+
+
+class BasicAuth(namedtuple('BasicAuth', ['login', 'password', 'encoding'])):
+    def __new__(cls, login, password=None, encoding='latin1'):
+        return super().__new__(cls, login, password, encoding)
+
+    def __bool__(self):
+        return self.login is not None and self.password is not None
+
+    def encode(self):
+        if not self:
+            raise ValueError("HTTP Auth login or password is missing")
+        creds = ('%s:%s' % (self.login, self.password)).encode(self.encoding)
+        return 'Basic %s' % base64.b64encode(creds).decode(self.encoding)
 
 
 class FormData:

tests/test_client.py

@@ -391,13 +391,14 @@ def test_no_path(self):
 
     def test_basic_auth(self):
         req = ClientRequest('get', 'http://python.org',
-                            auth=aiohttp.BasicAuth('nkim', '1234'))
+                            auth=aiohttp.helpers.BasicAuth('nkim', '1234'))
         self.assertIn('AUTHORIZATION', req.headers)
         self.assertEqual('Basic bmtpbToxMjM0', req.headers['AUTHORIZATION'])
 
     def test_basic_auth_utf8(self):
         req = ClientRequest('get', 'http://python.org',
-                            auth=aiohttp.BasicAuthEx('nkim', 'секрет', 'utf-8'))
+                            auth=aiohttp.helpers.BasicAuth('nkim', 'секрет',
+                                                           'utf-8'))
         self.assertIn('AUTHORIZATION', req.headers)
         self.assertEqual('Basic bmtpbTrRgdC10LrRgNC10YI=',
                          req.headers['AUTHORIZATION'])
@@ -414,15 +415,16 @@ def test_basic_auth_from_url(self):
 
         req = ClientRequest(
             'get', 'http://[email protected]',
-            auth=aiohttp.BasicAuth('nkim', '1234'))
+            auth=aiohttp.helpers.BasicAuth('nkim', '1234'))
         self.assertIn('AUTHORIZATION', req.headers)
         self.assertEqual('Basic bmtpbToxMjM0', req.headers['AUTHORIZATION'])
 
     def test_basic_auth_err(self):
         # missing password here
         self.assertRaises(
             ValueError, ClientRequest,
-            'get', 'http://python.org', auth=aiohttp.BasicAuth('nkim', None))
+            'get', 'http://python.org',
+            auth=aiohttp.helpers.BasicAuth('nkim', None))
 
     def test_no_content_length(self):
         req = ClientRequest('get', 'http://python.org', loop=self.loop)

tests/test_connector.py

@@ -357,8 +357,8 @@ def test_proxy_auth(self):
                                    proxy_auth=('user', 'pass'),
                                    loop=unittest.mock.Mock())
         self.assertEqual(ctx.exception.args[0],
-                         ("proxy_auth must be None, BasicAuth() "
-                          "or BasicAuthEx()", ('user', 'pass')))
+                         ("proxy_auth must be None or BasicAuth() tuple",
+                          ('user', 'pass')))
 
 
     @unittest.mock.patch('aiohttp.connector.ClientRequest')
@@ -371,7 +371,7 @@ def test_proxy_override_auth(self, ClientRequestMock):
 
         connector = aiohttp.ProxyConnector(
             'http://user:[email protected]',
-            proxy_auth=aiohttp.BasicAuth(None, None),
+            proxy_auth=aiohttp.helpers.BasicAuth(None, None),
             loop=loop_mock)
 
         def check_proxy_req(*args, **kw):
@@ -398,15 +398,16 @@ def test_proxy_connection_error(self):
     @unittest.mock.patch('aiohttp.connector.ClientRequest')
     def test_auth(self, ClientRequestMock):
         proxy_req = ClientRequest('GET', 'http://proxy.example.com',
-                                  auth=aiohttp.BasicAuth('user', 'pass'))
+                                  auth=aiohttp.helpers.BasicAuth('user',
+                                                                 'pass'))
         ClientRequestMock.return_value = proxy_req
         self.assertIn('AUTHORIZATION', proxy_req.headers)
         self.assertNotIn('PROXY-AUTHORIZATION', proxy_req.headers)
 
         loop_mock = unittest.mock.Mock()
         connector = aiohttp.ProxyConnector(
             'http://proxy.example.com', loop=loop_mock,
-            proxy_auth=aiohttp.BasicAuth('user', 'pass'))
+            proxy_auth=aiohttp.helpers.BasicAuth('user', 'pass'))
         connector._resolve_host = resolve_mock = unittest.mock.Mock()
         self._fake_coroutine(resolve_mock, [unittest.mock.MagicMock()])
 
@@ -426,15 +427,13 @@ def test_auth(self, ClientRequestMock):
 
         ClientRequestMock.assert_called_with(
             'GET', 'http://proxy.example.com',
-            auth=aiohttp.BasicAuth('user', 'pass'),
+            auth=aiohttp.helpers.BasicAuth('user', 'pass'),
             loop=unittest.mock.ANY, headers=unittest.mock.ANY)
 
-    @unittest.mock.patch('aiohttp.connector.ClientRequest')
-    def test_auth_utf8(self, ClientRequestMock):
+    def test_auth_utf8(self):
         proxy_req = ClientRequest(
             'GET', 'http://proxy.example.com',
-            auth=aiohttp.BasicAuthEx('юзер', 'пасс', 'utf-8'))
-        ClientRequestMock.return_value = proxy_req
+            auth=aiohttp.helpers.BasicAuth('юзер', 'пасс', 'utf-8'))
         self.assertIn('AUTHORIZATION', proxy_req.headers)
 
     @unittest.mock.patch('aiohttp.connector.ClientRequest')