openstack/keystonemiddleware
Code Issues Proposed changes

Remove six

Browse Source 
Change-Id: Ib3edfdd087ed1d954f1ecf72a191138f8f1c46a1
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2023-06-27 17:40:30 +01:00
parent 0f48ff3e8d
commit 22408f8da0
17 changed files with 53 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
keystonemiddleware/audit/_api.py
View File

@ -11,6 +11,7 @@
# under the License.
import collections
import configparser
import re
from oslo_log import log as logging
@ -24,18 +25,7 @@ from pycadf import host
from pycadf import identifier
from pycadf import resource
from pycadf import tag
import six
from six.moves import configparser
from six.moves.urllib import parse as urlparse
# NOTE(blk-u): Compatibility for Python 2. SafeConfigParser and
# SafeConfigParser.readfp are deprecated in Python 3. Remove this when we drop
# support for Python 2.
if six.PY2:
class _ConfigParser(configparser.SafeConfigParser):
read_file = configparser.SafeConfigParser.readfp
else:
_ConfigParser = configparser.ConfigParser
from urllib import parse as urlparse
Service = collections.namedtuple('Service',
@ -86,7 +76,7 @@ class OpenStackAuditApi(object):
if cfg_file:
try:
map_conf = _ConfigParser()
map_conf = configparser.ConfigParser()
with open(cfg_file) as fh:
map_conf.read_file(fh)

13
keystonemiddleware/auth_token/_cache.py
View File

@ -15,7 +15,6 @@ import hashlib
from oslo_serialization import jsonutils
from oslo_utils import timeutils
import six
from keystonemiddleware.auth_token import _exceptions as exc
from keystonemiddleware.auth_token import _memcache_crypt as memcache_crypt
@ -28,7 +27,7 @@ def _hash_key(key):
Using a known-length cache key is important to ensure that memcache
maximum key length is not exceeded causing failures to validate.
"""
if isinstance(key, six.text_type):
if isinstance(key, str):
# NOTE(morganfainberg): Ensure we are always working with a bytes
# type required for the hasher. In python 2.7 it is possible to
# get a text_type (unicode). In python 3.4 all strings are
@ -236,11 +235,11 @@ class TokenCache(object):
if serialized is None:
return None
if isinstance(serialized, six.text_type):
if isinstance(serialized, str):
serialized = serialized.encode('utf8')
data = self._deserialize(serialized, context)
if not isinstance(data, six.string_types):
if not isinstance(data, str):
data = data.decode('utf-8')
return jsonutils.loads(data)
@ -248,7 +247,7 @@ class TokenCache(object):
def set(self, token_id, data):
"""Store value into memcache."""
data = jsonutils.dumps(data)
if isinstance(data, six.text_type):
if isinstance(data, str):
data = data.encode('utf-8')
cache_key, context = self._get_cache_key(token_id)
@ -273,9 +272,9 @@ class SecureTokenCache(TokenCache):
'memcache_security_strategy is defined')
raise exc.ConfigurationError(msg)
if isinstance(security_strategy, six.string_types):
if isinstance(security_strategy, str):
security_strategy = security_strategy.encode('utf-8')
if isinstance(secret_key, six.string_types):
if isinstance(secret_key, str):
secret_key = secret_key.encode('utf-8')
self._security_strategy = security_strategy

3
keystonemiddleware/auth_token/_identity.py
View File

@ -10,11 +10,12 @@
# License for the specific language governing permissions and limitations
# under the License.
import urllib.parse
from keystoneauth1 import discover
from keystoneauth1 import exceptions as ksa_exceptions
from keystoneauth1 import plugin
from keystoneclient.v3 import client as v3_client
from six.moves import urllib
from keystonemiddleware.auth_token import _auth
from keystonemiddleware.auth_token import _exceptions as ksm_exceptions

11
keystonemiddleware/auth_token/_memcache_crypt.py
View File

@ -33,7 +33,6 @@ import hashlib
import hmac
import math
import os
import six
from keystonemiddleware.i18n import _
from oslo_utils import secretutils
@ -99,13 +98,13 @@ def derive_keys(token, secret, strategy):
This approach is faster than computing a separate hmac as the KDF
for each desired key.
"""
if not isinstance(secret, six.binary_type):
if not isinstance(secret, bytes):
secret = secret.encode()
if not isinstance(token, six.binary_type):
if not isinstance(token, bytes):
token = token.encode()
if not isinstance(strategy, six.binary_type):
if not isinstance(strategy, bytes):
strategy = strategy.encode()
digest = hmac.new(secret, token + strategy, HASH_FUNCTION).digest()
@ -117,10 +116,10 @@ def derive_keys(token, secret, strategy):
def sign_data(key, data):
"""Sign the data using the defined function and the derived key."""
if not isinstance(key, six.binary_type):
if not isinstance(key, bytes):
key = key.encode()
if not isinstance(data, six.binary_type):
if not isinstance(data, bytes):
data = data.encode()
mac = hmac.new(key, data, HASH_FUNCTION).digest()

11
keystonemiddleware/ec2_token.py
View File

@ -23,7 +23,6 @@ from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
import requests
import six
import webob.dec
from keystonemiddleware.i18n import _
@ -75,8 +74,7 @@ class EC2Token(object):
'<Response><Errors><Error><Code>%s</Code>'
'<Message>%s</Message></Error></Errors></Response>' %
(code, message))
if six.PY3:
error_msg = error_msg.encode()
error_msg = error_msg.encode()
resp.body = error_msg
return resp
@ -141,10 +139,9 @@ class EC2Token(object):
auth_params.pop('Signature', None)
headers = req.headers
if six.PY3:
# NOTE(andrey-mp): jsonutils dumps it as list of keys without
# conversion instead real dict
headers = {k: headers[k] for k in headers}
# NOTE(andrey-mp): jsonutils dumps it as list of keys without
# conversion instead real dict
headers = {k: headers[k] for k in headers}
cred_dict = {
'access': access,
'signature': signature,

6
keystonemiddleware/s3_token.py
View File

@ -38,7 +38,6 @@ from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import strutils
import requests
import six
s3_opts = [
cfg.IntOpt('timeout', default=60,
@ -112,8 +111,7 @@ class S3Token(object):
'<Error>\r\n <Code>%s</Code>\r\n '
'<Message>%s</Message>\r\n</Error>\r\n' %
(code, error_table[code][1]))
if six.PY3:
error_msg = error_msg.encode()
error_msg = error_msg.encode()
resp.body = error_msg
return resp
@ -223,8 +221,6 @@ class S3Token(object):
req.headers['X-Auth-Token'] = token_id
tenant_to_connect = force_tenant or tenant['id']
if six.PY2 and isinstance(tenant_to_connect, six.text_type):
tenant_to_connect = tenant_to_connect.encode('utf-8')
self._logger.debug('Connecting with tenant: %s', tenant_to_connect)
new_tenant_name = '%s%s' % (self._reseller_prefix, tenant_to_connect)
environ['PATH_INFO'] = environ['PATH_INFO'].replace(account,

6
keystonemiddleware/tests/unit/auth_token/base.py
View File

@ -10,13 +10,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import http.client as http_client
import fixtures
from oslo_config import cfg
from oslo_config import fixture as cfg_fixture
from oslo_log import log as logging
from requests_mock.contrib import fixture as rm_fixture
import six
from six.moves import http_client
import webob.dec
from keystonemiddleware import auth_token
@ -60,6 +60,6 @@ class BaseAuthTokenTestCase(utils.MiddlewareTestCase):
resp = req.get_response(middleware)
self.assertEqual(expected_status, resp.status_int)
if expected_body_string:
self.assertIn(expected_body_string, six.text_type(resp.body))
self.assertIn(expected_body_string, str(resp.body))
resp.request = req
return resp

4
keystonemiddleware/tests/unit/auth_token/test_auth.py
View File

@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import io
import uuid
from keystoneauth1 import fixture
@ -17,7 +18,6 @@ from keystoneauth1 import plugin as ksa_plugin
from keystoneauth1 import session
from oslo_log import log as logging
from requests_mock.contrib import fixture as rm_fixture
import six
from keystonemiddleware.auth_token import _auth
from keystonemiddleware.tests.unit import utils
@ -47,7 +47,7 @@ class DefaultAuthPluginTests(utils.BaseTestCase):
def setUp(self):
super(DefaultAuthPluginTests, self).setUp()
self.stream = six.StringIO()
self.stream = io.StringIO()
self.logger = logging.getLogger(__name__)
self.session = session.Session()
self.requests_mock = self.useFixture(rm_fixture.Fixture())

25
keystonemiddleware/tests/unit/auth_token/test_auth_token_middleware.py
View File

@ -28,7 +28,6 @@ from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import timeutils
import pbr.version
import six
import testresources
from testtools import matchers
import webob
@ -129,7 +128,7 @@ class TimeFixture(fixtures.Fixture):
def __init__(self, new_time, normalize=True):
super(TimeFixture, self).__init__()
if isinstance(new_time, six.string_types):
if isinstance(new_time, str):
new_time = timeutils.parse_isotime(new_time)
if normalize:
new_time = timeutils.normalize_time(new_time)
@ -313,11 +312,11 @@ class BaseAuthTokenMiddlewareTest(base.BaseAuthTokenTestCase):
self.middleware._app.expected_env.update(expected_env)
def purge_token_expected_env(self):
for key in six.iterkeys(self.token_expected_env):
for key in self.token_expected_env.keys():
del self.middleware._app.expected_env[key]
def purge_service_token_expected_env(self):
for key in six.iterkeys(self.service_token_expected_env):
for key in self.service_token_expected_env.keys():
del self.middleware._app.expected_env[key]
def assertLastPath(self, path):
@ -1349,7 +1348,7 @@ class DelayedAuthTests(BaseAuthTokenMiddlewareTest):
body=body,
conf=conf)
resp = self.call(middleware, expected_status=401)
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
self.assertEqual('Keystone uri="%s"' % www_authenticate_uri,
resp.headers['WWW-Authenticate'])
@ -1387,7 +1386,7 @@ class DelayedAuthTests(BaseAuthTokenMiddlewareTest):
middleware = self.create_simple_middleware(body=body, conf=conf)
resp = self.call(middleware)
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
token_auth = resp.request.environ['keystone.token_auth']
@ -1414,7 +1413,7 @@ class DelayedAuthTests(BaseAuthTokenMiddlewareTest):
middleware = self.create_simple_middleware(body=body, conf=conf)
resp = self.call(middleware, headers={
'X-Auth-Token': 'non-keystone-token'})
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
token_auth = resp.request.environ['keystone.token_auth']
@ -1441,7 +1440,7 @@ class DelayedAuthTests(BaseAuthTokenMiddlewareTest):
middleware = self.create_simple_middleware(body=body, conf=conf)
resp = self.call(middleware, headers={'X-Auth-Token': ERROR_TOKEN})
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
token_auth = resp.request.environ['keystone.token_auth']
@ -1504,7 +1503,7 @@ class CommonCompositeAuthTests(object):
# Check arbitrary headers not removed
req.headers['X-Foo'] = 'Bar'
resp = req.get_response(self.middleware)
for key in six.iterkeys(self.service_token_expected_env):
for key in self.service_token_expected_env.keys():
header_key = key[len('HTTP_'):].replace('_', '-')
self.assertFalse(req.headers.get(header_key))
self.assertEqual('Bar', req.headers.get('X-Foo'))
@ -1581,7 +1580,7 @@ class CommonCompositeAuthTests(object):
# Check arbitrary headers not removed
req.headers['X-Foo'] = 'Bar'
resp = req.get_response(self.middleware)
for key in six.iterkeys(self.service_token_expected_env):
for key in self.service_token_expected_env.keys():
header_key = key[len('HTTP_'):].replace('_', '-')
self.assertFalse(req.headers.get(header_key))
self.assertEqual('Bar', req.headers.get('X-Foo'))
@ -1836,7 +1835,7 @@ class AuthProtocolLoadingTests(BaseAuthTokenMiddlewareTest):
app = self.create_simple_middleware(body=body)
resp = self.good_request(app)
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
@staticmethod
def get_plugin(app):
@ -1878,7 +1877,7 @@ class AuthProtocolLoadingTests(BaseAuthTokenMiddlewareTest):
app = self.create_simple_middleware(body=body, conf=conf)
resp = self.good_request(app)
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
plugin = self.get_plugin(app)
@ -1916,7 +1915,7 @@ class AuthProtocolLoadingTests(BaseAuthTokenMiddlewareTest):
app = self.create_simple_middleware(body=body, conf=conf)
resp = self.good_request(app)
self.assertEqual(six.b(body), resp.body)
self.assertEqual(body.encode(), resp.body)
plugin = self.get_plugin(app)

9
keystonemiddleware/tests/unit/auth_token/test_cache.py
View File

@ -13,7 +13,6 @@
import uuid
import fixtures
import six
from keystonemiddleware.auth_token import _cache
from keystonemiddleware.auth_token import _exceptions as exc
@ -114,7 +113,7 @@ class TestLiveMemcache(base.BaseAuthTokenTestCase):
'memcache_secret_key': 'mysecret'
}
token = six.b(uuid.uuid4().hex)
token = uuid.uuid4().hex.encode()
data = uuid.uuid4().hex
token_cache = self.create_simple_middleware(conf=conf)._token_cache
@ -130,7 +129,7 @@ class TestLiveMemcache(base.BaseAuthTokenTestCase):
'memcache_secret_key': 'mysecret'
}
token = six.b(uuid.uuid4().hex)
token = uuid.uuid4().hex.encode()
data = uuid.uuid4().hex
token_cache = self.create_simple_middleware(conf=conf)._token_cache
@ -145,7 +144,7 @@ class TestLiveMemcache(base.BaseAuthTokenTestCase):
'memcache_secret_key': 'mysecret'
}
token = six.b(uuid.uuid4().hex)
token = uuid.uuid4().hex.encode()
data = uuid.uuid4().hex
token_cache = self.create_simple_middleware(conf=conf)._token_cache
@ -159,7 +158,7 @@ class TestLiveMemcache(base.BaseAuthTokenTestCase):
'memcache_use_advanced_pool': True
}
token = six.b(uuid.uuid4().hex)
token = uuid.uuid4().hex.encode()
data = uuid.uuid4().hex
token_cache = self.create_simple_middleware(conf=conf)._token_cache

2
keystonemiddleware/tests/unit/auth_token/test_connection_pool.py
View File

@ -10,11 +10,11 @@
# License for the specific language governing permissions and limitations
# under the License.
import queue
import time
from unittest import mock
from oslo_cache import _memcache_pool
from six.moves import queue
import testtools
from testtools import matchers

5
keystonemiddleware/tests/unit/auth_token/test_memcache_crypt.py
View File

@ -10,7 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
import struct
from keystonemiddleware.auth_token import _memcache_crypt as memcache_crypt
from keystonemiddleware.tests.unit import utils
@ -41,10 +41,11 @@ class MemcacheCryptPositiveTests(utils.BaseTestCase):
self.assertEqual(len(sig), memcache_crypt.DIGEST_LENGTH_B64)
def test_encryption(self):
int2byte = struct.Struct(">B").pack
keys = self._setup_keys(b'ENCRYPT')
# what you put in is what you get out
for data in [b'data', b'1234567890123456', b'\x00\xFF' * 13
] + [six.int2byte(x % 256) * x for x in range(768)]:
] + [int2byte(x % 256) * x for x in range(768)]:
crypt = memcache_crypt.encrypt_data(keys['ENCRYPTION'], data)
decrypt = memcache_crypt.decrypt_data(keys['ENCRYPTION'], crypt)
self.assertEqual(data, decrypt)

5
keystonemiddleware/tests/unit/test_ec2_token_middleware.py
View File

@ -16,7 +16,6 @@ from unittest import mock
from oslo_serialization import jsonutils
import requests
import six
import webob
from keystonemiddleware import ec2_token
@ -65,9 +64,7 @@ class EC2TokenMiddlewareTestBase(utils.TestCase):
self.assertEqual(http_status, response.status_code,
'Expected HTTP status %s' % http_status)
error_msg = '<Code>%s</Code>' % ec2_code
if six.PY3:
# encode error message like main code
error_msg = error_msg.encode()
error_msg = error_msg.encode()
self.assertIn(error_msg, response.body)

5
keystonemiddleware/tests/unit/test_oauth2_mtls_token_middleware.py
View File

@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import http.client as http_client
import json
import logging
import ssl
@ -21,8 +22,6 @@ import webob.dec
import fixtures
from oslo_config import cfg
import six
from six.moves import http_client
import testresources
from keystoneauth1 import access
@ -132,7 +131,7 @@ class BaseOauth2mTlsTokenMiddlewareTest(base.BaseAuthTokenTestCase):
resp = req.get_response(middleware)
self.assertEqual(expected_status, resp.status_int)
if expected_body_string:
self.assertIn(expected_body_string, six.text_type(resp.body))
self.assertIn(expected_body_string, str(resp.body))
resp.request = req
return resp

5
keystonemiddleware/tests/unit/test_oauth2_token_middleware.py
View File

@ -13,14 +13,13 @@
# under the License.
import fixtures
import http.client as http_client
import logging
import six
import testresources
import uuid
import webob.dec
from oslo_config import cfg
from six.moves import http_client
from keystoneauth1 import exceptions as ksa_exceptions
@ -118,7 +117,7 @@ class BaseOauth2TokenMiddlewareTest(base.BaseAuthTokenTestCase):
resp = req.get_response(middleware)
self.assertEqual(expected_status, resp.status_int)
if expected_body_string:
self.assertIn(expected_body_string, six.text_type(resp.body))
self.assertIn(expected_body_string, str(resp.body))
resp.request = req
return resp

7
keystonemiddleware/tests/unit/test_s3_token_middleware.py
View File

@ -13,13 +13,12 @@
# under the License.
from unittest import mock
import urllib.parse
import fixtures
from oslo_serialization import jsonutils
import requests
from requests_mock.contrib import fixture as rm_fixture
import six
from six.moves import urllib
from testtools import matchers
import webob
@ -129,9 +128,7 @@ class S3TokenMiddlewareTestGood(S3TokenMiddlewareTestBase):
self.middleware = (
s3_token.filter_factory({'insecure': 'True'})(FakeApp()))
text_return_value = jsonutils.dumps(GOOD_RESPONSE)
if six.PY3:
text_return_value = text_return_value.encode()
text_return_value = jsonutils.dumps(GOOD_RESPONSE).encode()
MOCK_REQUEST.return_value = utils.TestResponse({
'status_code': 201,
'text': text_return_value})

1
requirements.txt
View File

@ -14,5 +14,4 @@ pbr!=2.1.0,>=2.0.0 # Apache-2.0
pycadf!=2.0.0,>=1.1.0 # Apache-2.0
python-keystoneclient>=3.20.0 # Apache-2.0
requests>=2.14.2 # Apache-2.0
six>=1.10.0 # MIT
WebOb>=1.7.1 # MIT