openstack
/
openstacksdk
Code Issues Proposed changes

Remove use of six 

We're python3 now, we don't need to use six.

Change-Id: I7c1f0bc89838432aa0ce916beb462916c8763cd1
This commit is contained in:
Monty Taylor 2020-03-29 10:43:43 -05:00
parent fe00460b99
commit 2ddbf57ee6
38 changed files with 78 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
lower-constraints.txt
View File

@ -28,7 +28,6 @@ PyYAML==3.12
requests==2.18.0
requests-mock==1.2.0
requestsexceptions==1.2.0
six==1.10.0
statsd==3.3.0
stestr==1.0.0
testscenarios==0.4

6
openstack/baremetal/configdrive.py
View File

@ -21,8 +21,6 @@ import shutil
import subprocess
import tempfile
import six
@contextlib.contextmanager
def populate_directory(metadata, user_data=None, versions=None,
@ -60,7 +58,7 @@ def populate_directory(metadata, user_data=None, versions=None,
if user_data:
# Strictly speaking, user data is binary, but in many cases
# it's actually a text (cloud-init, ignition, etc).
flag = 't' if isinstance(user_data, six.text_type) else 'b'
flag = 't' if isinstance(user_data, str) else 'b'
with open(os.path.join(subdir, 'user_data'),
'w%s' % flag) as fp:
fp.write(user_data)
@ -142,7 +140,7 @@ def pack(path):
# NOTE(dtantsur): Ironic expects configdrive to be a string, but base64
# returns bytes on Python 3.
if not isinstance(cd, six.string_types):
if not isinstance(cd, str):
cd = cd.decode('utf-8')
return cd

4
openstack/block_storage/_base_proxy.py
View File

@ -11,13 +11,11 @@
# under the License.
import abc
import six
from openstack import exceptions
from openstack import proxy
class BaseBlockStorageProxy(six.with_metaclass(abc.ABCMeta, proxy.Proxy)):
class BaseBlockStorageProxy(proxy.Proxy, metaclass=abc.ABCMeta):
def create_image(
self, name, volume, allow_duplicates,

7
openstack/cloud/_compute.py
View File

@ -18,7 +18,6 @@ import datetime
import functools
import iso8601
import operator
import six
import threading
import time
import types # noqa
@ -1634,7 +1633,7 @@ class ComputeCloudMixin(_normalize.Normalizer):
:raises: OpenStackCloudException on operation error.
"""
if (
isinstance(name_or_id, six.string_types + (six.binary_type,))
isinstance(name_or_id, (str, bytes))
and not name_or_id.isdigit()
):
aggregate = self.get_aggregate(name_or_id)
@ -1828,9 +1827,9 @@ class ComputeCloudMixin(_normalize.Normalizer):
if hasattr(userdata, 'read'):
userdata = userdata.read()
if not isinstance(userdata, six.binary_type):
if not isinstance(userdata, bytes):
# If the userdata passed in is bytes, just send it unmodified
if not isinstance(userdata, six.string_types):
if not isinstance(userdata, str):
raise TypeError("%s can't be encoded" % type(userdata))
# If it's not bytes, make it bytes
userdata = userdata.encode('utf-8', 'strict')

3
openstack/cloud/_floating_ip.py
View File

@ -16,7 +16,6 @@
import ipaddress
# import jsonpatch
import threading
import six
import time
import types # noqa
@ -262,7 +261,7 @@ class FloatingIPCloudMixin(_normalize.Normalizer):
project_id = self.current_project_id
if network:
if isinstance(network, six.string_types):
if isinstance(network, str):
network = [network]
# Use given list to get first matching external network

3
openstack/cloud/_network.py
View File

@ -13,7 +13,6 @@
# import types so that we can reference ListType in sphinx param declarations.
# We can't just use list, because sphinx gets confused by
# openstack.resource.Resource.list and openstack.resource2.Resource.list
import six
import time
import threading
import types # noqa
@ -2205,7 +2204,7 @@ class NetworkCloudMixin(_normalize.Normalizer):
'True')
# Be friendly on ip_version and allow strings
if isinstance(ip_version, six.string_types):
if isinstance(ip_version, str):
try:
ip_version = int(ip_version)
except ValueError:

3
openstack/cloud/_normalize.py
View File

@ -18,7 +18,6 @@
import datetime
import munch
import six
from openstack import resource
@ -117,7 +116,7 @@ def _split_filters(obj_name='', filters=None, **kwargs):
def _to_bool(value):
if isinstance(value, six.string_types):
if isinstance(value, str):
if not value:
return False
prospective = value.lower().capitalize()

8
openstack/cloud/_object_store.py
View File

@ -18,7 +18,6 @@ import concurrent.futures
import hashlib
import json
import os
import six
import types # noqa
import urllib.parse
@ -256,9 +255,6 @@ class ObjectStoreCloudMixin(_normalize.Normalizer):
caps = self.get_object_capabilities()
except exc.OpenStackCloudHTTPError as e:
if e.response.status_code in (404, 412):
# Clear the exception so that it doesn't linger
# and get reported as an Inner Exception later
_utils._exc_clear()
server_max_file_size = DEFAULT_MAX_FILE_SIZE
self.log.info(
"Swift capabilities not supported. "
@ -819,14 +815,14 @@ class ObjectStoreCloudMixin(_normalize.Normalizer):
response_headers = {
k.lower(): v for k, v in response.headers.items()}
if outfile:
if isinstance(outfile, six.string_types):
if isinstance(outfile, str):
outfile_handle = open(outfile, 'wb')
else:
outfile_handle = outfile
for chunk in response.iter_content(
resp_chunk_size, decode_unicode=False):
outfile_handle.write(chunk)
if isinstance(outfile, six.string_types):
if isinstance(outfile, str):
outfile_handle.close()
else:
outfile_handle.flush()

10
openstack/cloud/_utils.py
View File

@ -20,9 +20,7 @@ import jmespath
import munch
import netifaces
import re
import six
import sre_constants
import sys
import time
import uuid
@ -35,12 +33,6 @@ from openstack.cloud import meta
_decorated_methods = []
def _exc_clear():
"""Because sys.exc_clear is gone in py3 and is not in six."""
if sys.version_info[0] == 2:
sys.exc_clear()
def _make_unicode(input):
"""Turn an input into unicode unconditionally
@ -134,7 +126,7 @@ def _filter_list(data, name_or_id, filters):
if not filters:
return data
if isinstance(filters, six.string_types):
if isinstance(filters, str):
return jmespath.search(filters, data)
def _dict_filter(f, d):

3
openstack/cloud/meta.py
View File

@ -15,7 +15,6 @@
import munch
import ipaddress
import six
import socket
from openstack import _log
@ -23,7 +22,7 @@ from openstack import utils
from openstack.cloud import exc
NON_CALLABLES = (six.string_types, bool, dict, int, float, list, type(None))
NON_CALLABLES = (str, bool, dict, int, float, list, type(None))
def find_nova_interfaces(addresses, ext_tag=None, key_name=None, version=4,

3
openstack/cloud/openstackcloud.py
View File

@ -12,7 +12,6 @@
import copy
import functools
import queue
import six
# import types so that we can reference ListType in sphinx param declarations.
# We can't just use list, because sphinx gets confused by
# openstack.resource.Resource.list and openstack.resource2.Resource.list
@ -351,7 +350,7 @@ class _OpenStackCloudMixin(object):
def _get_major_version_id(self, version):
if isinstance(version, int):
return version
elif isinstance(version, six.string_types + (tuple,)):
elif isinstance(version, (str, tuple)):
return int(version[0])
return version

3
openstack/compute/v2/metadata.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack import exceptions
from openstack import utils
@ -22,7 +21,7 @@ class MetadataMixin(object):
metadata=None):
metadata = metadata or {}
for k, v in metadata.items():
if not isinstance(v, six.string_types):
if not isinstance(v, str):
raise ValueError("The value for %s (%s) must be "
"a text string" % (k, v))

3
openstack/connection.py
View File

@ -182,7 +182,6 @@ import weakref
import concurrent.futures
import keystoneauth1.exceptions
import requestsexceptions
import six
from openstack import _log
from openstack import _services_mixin
@ -465,7 +464,7 @@ class Connection(
"""
# If we don't have a proxy, just instantiate Proxy so that
# we get an adapter.
if isinstance(service, six.string_types):
if isinstance(service, str):
service = service_description.ServiceDescription(service)
# Directly invoke descriptor of the ServiceDescription

9
openstack/dns/v2/_base.py
View File

@ -9,7 +9,7 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import six
import urllib.parse
from openstack import exceptions
from openstack import resource
@ -89,11 +89,10 @@ class Resource(resource.Resource):
# This prevents duplication of query parameters that with large
# number of pages result in HTTP 414 error eventually.
if next_link:
parts = six.moves.urllib.parse.urlparse(next_link)
query_params = six.moves.urllib.parse.parse_qs(parts.query)
parts = urllib.parse.urlparse(next_link)
query_params = urllib.parse.parse_qs(parts.query)
params.update(query_params)
next_link = six.moves.urllib.parse.urljoin(next_link,
parts.path)
next_link = urllib.parse.urljoin(next_link, parts.path)
# If we still have no link, and limit was given and is non-zero,
# and the number of records yielded equals the limit, then the user

5
openstack/exceptions.py
View File

@ -20,7 +20,6 @@ import json
import re
from requests import exceptions as _rex
import six
class SDKException(Exception):
@ -101,7 +100,7 @@ class HttpException(SDKException, _rex.HTTPError):
if self.details:
remote_error += ', '
if self.details:
remote_error += six.text_type(self.details)
remote_error += str(self.details)
return "{message}: {remote_error}".format(
message=super(HttpException, self).__str__(),
@ -176,7 +175,7 @@ def _extract_message(obj):
# Ironic starting with Stein
elif obj.get('faultstring'):
return obj['faultstring']
elif isinstance(obj, six.string_types):
elif isinstance(obj, str):
# Ironic before Stein has double JSON encoding, nobody remembers why.
try:
obj = json.loads(obj)

3
openstack/image/_base_proxy.py
View File

@ -12,13 +12,12 @@
import abc
import os
import six
from openstack import exceptions
from openstack import proxy
class BaseImageProxy(six.with_metaclass(abc.ABCMeta, proxy.Proxy)):
class BaseImageProxy(proxy.Proxy, metaclass=abc.ABCMeta):
retriable_status_codes = [503]

6
openstack/image/_download.py
View File

@ -11,7 +11,6 @@
# under the License.
import io
import hashlib
import six
from openstack import exceptions
from openstack import utils
@ -49,10 +48,7 @@ class DownloadMixin(object):
md5 = hashlib.md5()
if output:
try:
# In python 2 we might get StringIO - delete it as soon as
# py2 support is dropped
if isinstance(output, io.IOBase) \
or isinstance(output, six.StringIO):
if isinstance(output, io.IOBase):
for chunk in resp.iter_content(chunk_size=chunk_size):
output.write(chunk)
md5.update(chunk)

18
openstack/object_store/v1/_proxy.py
View File

@ -19,8 +19,6 @@ import os
import time
from urllib import parse
import six
from openstack.object_store.v1 import account as _account
from openstack.object_store.v1 import container as _container
from openstack.object_store.v1 import obj as _obj
@ -637,9 +635,6 @@ class Proxy(proxy.Proxy):
caps = self.get_info()
except exceptions.SDKException as e:
if e.response.status_code in (404, 412):
# Clear the exception so that it doesn't linger
# and get reported as an Inner Exception later
_utils._exc_clear()
server_max_file_size = DEFAULT_MAX_FILE_SIZE
self._connection.log.info(
"Swift capabilities not supported. "
@ -730,13 +725,13 @@ class Proxy(proxy.Proxy):
account_meta = self.get_account_metadata()
temp_url_key = (account_meta.meta_temp_url_key_2
or account_meta.meta_temp_url_key)
if temp_url_key and not isinstance(temp_url_key, six.binary_type):
if temp_url_key and not isinstance(temp_url_key, bytes):
temp_url_key = temp_url_key.encode('utf8')
return temp_url_key
def _check_temp_url_key(self, container=None, temp_url_key=None):
if temp_url_key:
if not isinstance(temp_url_key, six.binary_type):
if not isinstance(temp_url_key, bytes):
temp_url_key = temp_url_key.encode('utf8')
else:
temp_url_key = self.get_temp_url_key(container)
@ -792,8 +787,7 @@ class Proxy(proxy.Proxy):
data = '%s\n%s\n%s\n%s\n%s' % (path, redirect_url, max_file_size,
max_upload_count, expires)
if six.PY3:
data = data.encode('utf8')
data = data.encode('utf8')
sig = hmac.new(temp_url_key, data, sha1).hexdigest()
return (expires, sig)
@ -864,7 +858,7 @@ class Proxy(proxy.Proxy):
raise ValueError('time must either be a whole number '
'or in specific ISO 8601 format.')
if isinstance(path, six.binary_type):
if isinstance(path, bytes):
try:
path_for_body = path.decode('utf-8')
except UnicodeDecodeError:
@ -895,7 +889,7 @@ class Proxy(proxy.Proxy):
('prefix:' if prefix else '') + path_for_body]
if ip_range:
if isinstance(ip_range, six.binary_type):
if isinstance(ip_range, bytes):
try:
ip_range = ip_range.decode('utf-8')
except UnicodeDecodeError:
@ -924,7 +918,7 @@ class Proxy(proxy.Proxy):
if prefix:
temp_url += u'&temp_url_prefix={}'.format(parts[4])
# Have return type match path from caller
if isinstance(path, six.binary_type):
if isinstance(path, bytes):
return temp_url.encode('utf-8')
else:
return temp_url

7
openstack/orchestration/util/template_utils.py
View File

@ -14,7 +14,6 @@
import collections
import json
import six
from urllib import parse
from urllib import request
@ -55,7 +54,7 @@ def get_template_contents(template_file=None, template_url=None,
'Could not fetch template from %s' % template_url)
try:
if isinstance(tpl, six.binary_type):
if isinstance(tpl, bytes):
tpl = tpl.decode('utf-8')
template = template_format.parse(tpl)
except ValueError as e:
@ -77,7 +76,7 @@ def resolve_template_get_files(template, files, template_base_url,
def ignore_if(key, value):
if key != 'get_file' and key != 'type':
return True
if not isinstance(value, six.string_types):
if not isinstance(value, str):
return True
if (key == 'type'
and not value.endswith(('.yaml', '.template'))):
@ -93,7 +92,7 @@ def resolve_template_get_files(template, files, template_base_url,
def is_template(file_content):
try:
if isinstance(file_content, six.binary_type):
if isinstance(file_content, bytes):
file_content = file_content.decode('utf-8')
template_format.parse(file_content)
except (ValueError, TypeError):

11
openstack/resource.py
View File

@ -34,6 +34,7 @@ and then returned to the caller.
import collections
import inspect
import itertools
import urllib.parse
import jsonpatch
import operator
@ -41,7 +42,6 @@ from keystoneauth1 import adapter
from keystoneauth1 import discover
import munch
from requests import structures
import six
from openstack import _log
from openstack import exceptions
@ -1094,7 +1094,7 @@ class Resource(dict):
uri = utils.urljoin(uri, self.id)
if params:
query_params = six.moves.urllib.parse.urlencode(params)
query_params = urllib.parse.urlencode(params)
uri += '?' + query_params
return _Request(uri, body, headers)
@ -1776,11 +1776,10 @@ class Resource(dict):
# This prevents duplication of query parameters that with large
# number of pages result in HTTP 414 error eventually.
if next_link:
parts = six.moves.urllib.parse.urlparse(next_link)
query_params = six.moves.urllib.parse.parse_qs(parts.query)
parts = urllib.parse.urlparse(next_link)
query_params = urllib.parse.parse_qs(parts.query)
params.update(query_params)
next_link = six.moves.urllib.parse.urljoin(next_link,
parts.path)
next_link = urllib.parse.urljoin(next_link, parts.path)
# If we still have no link, and limit was given and is non-zero,
# and the number of records yielded equals the limit, then the user

5
openstack/tests/functional/cloud/test_compute.py
View File

@ -20,7 +20,6 @@ Functional tests for `shade` compute methods.
import datetime
from fixtures import TimeoutException
import six
from openstack.cloud import exc
from openstack.tests.functional import base
@ -200,7 +199,7 @@ class TestCompute(base.BaseFunctionalTest):
# returning a string tests that the call is correct. Testing that
# the cloud returns actual data in the output is out of scope.
log = self.user_cloud._get_server_console_output(server_id=server.id)
self.assertTrue(isinstance(log, six.string_types))
self.assertTrue(isinstance(log, str))
def test_get_server_console_name_or_id(self):
self.addCleanup(self._cleanup_servers_and_volumes, self.server_name)
@ -210,7 +209,7 @@ class TestCompute(base.BaseFunctionalTest):
flavor=self.flavor,
wait=True)
log = self.user_cloud.get_server_console(server=self.server_name)
self.assertTrue(isinstance(log, six.string_types))
self.assertTrue(isinstance(log, str))
def test_list_availability_zone_names(self):
self.assertEqual(

4
openstack/tests/functional/cloud/test_floating_ip.py
View File

@ -20,7 +20,6 @@ Functional tests for floating IP resource.
"""
import pprint
import six
import sys
from testtools import content
@ -102,8 +101,7 @@ class TestFloatingIP(base.BaseFunctionalTest):
content.text_content(
'\n'.join([str(ex) for ex in exception_list])))
exc = exception_list[0]
tb = tb_list[0]
six.reraise(type(exc), exc, tb)
raise exc
def _cleanup_servers(self):
exception_list = list()

7
openstack/tests/functional/compute/v2/test_extension.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack.tests.functional import base
@ -22,6 +21,6 @@ class TestExtension(base.BaseFunctionalTest):
self.assertGreater(len(extensions), 0)
for ext in extensions:
self.assertIsInstance(ext.name, six.string_types)
self.assertIsInstance(ext.namespace, six.string_types)
self.assertIsInstance(ext.alias, six.string_types)
self.assertIsInstance(ext.name, str)
self.assertIsInstance(ext.namespace, str)
self.assertIsInstance(ext.alias, str)

5
openstack/tests/functional/compute/v2/test_flavor.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack import exceptions
from openstack.tests.functional import base
@ -28,8 +27,8 @@ class TestFlavor(base.BaseFunctionalTest):
self.assertGreater(len(flavors), 0)
for flavor in flavors:
self.assertIsInstance(flavor.id, six.string_types)
self.assertIsInstance(flavor.name, six.string_types)
self.assertIsInstance(flavor.id, str)
self.assertIsInstance(flavor.name, str)
self.assertIsInstance(flavor.disk, int)
self.assertIsInstance(flavor.ram, int)
self.assertIsInstance(flavor.vcpus, int)

3
openstack/tests/functional/compute/v2/test_image.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack.tests.functional import base
from openstack.tests.functional.image.v2.test_image import TEST_IMAGE_NAME
@ -22,7 +21,7 @@ class TestImage(base.BaseFunctionalTest):
images = list(self.conn.compute.images())
self.assertGreater(len(images), 0)
for image in images:
self.assertIsInstance(image.id, six.string_types)
self.assertIsInstance(image.id, str)
def _get_non_test_image(self):
images = self.conn.compute.images()

7
openstack/tests/functional/network/v2/test_availability_zone.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack.tests.functional import base
@ -22,6 +21,6 @@ class TestAvailabilityZone(base.BaseFunctionalTest):
self.assertGreater(len(availability_zones), 0)
for az in availability_zones:
self.assertIsInstance(az.name, six.string_types)
self.assertIsInstance(az.resource, six.string_types)
self.assertIsInstance(az.state, six.string_types)
self.assertIsInstance(az.name, str)
self.assertIsInstance(az.resource, str)
self.assertIsInstance(az.state, str)

5
openstack/tests/functional/network/v2/test_extension.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack.tests.functional import base
@ -22,8 +21,8 @@ class TestExtension(base.BaseFunctionalTest):
self.assertGreater(len(extensions), 0)
for ext in extensions:
self.assertIsInstance(ext.name, six.string_types)
self.assertIsInstance(ext.alias, six.string_types)
self.assertIsInstance(ext.name, str)
self.assertIsInstance(ext.alias, str)
def test_find(self):
extension = self.conn.network.find_extension('external-net')

3
openstack/tests/functional/network/v2/test_qos_rule_type.py
View File

@ -10,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import six
from openstack.tests.functional import base
@ -34,4 +33,4 @@ class TestQoSRuleType(base.BaseFunctionalTest):
self.assertGreater(len(rule_types), 0)
for rule_type in rule_types:
self.assertIsInstance(rule_type.type, six.string_types)
self.assertIsInstance(rule_type.type, str)

5
openstack/tests/unit/baremetal/test_configdrive.py
View File

@ -17,7 +17,6 @@ import json
import os
import mock
import six
import testtools
from openstack.baremetal import configdrive
@ -58,7 +57,7 @@ class TestPopulateDirectory(testtools.TestCase):
if user_data is None:
self.assertFalse(os.path.exists(user_data_file))
else:
if isinstance(user_data, six.text_type):
if isinstance(user_data, str):
user_data = user_data.encode()
with open(user_data_file, 'rb') as fp:
self.assertEqual(user_data, fp.read())
@ -101,4 +100,4 @@ class TestPack(testtools.TestCase):
mock_popen.return_value.returncode = 0
result = configdrive.pack("/fake")
# Make sure the result is string on all python versions
self.assertIsInstance(result, six.string_types)
self.assertIsInstance(result, str)

7
openstack/tests/unit/cloud/test_image.py
View File

@ -11,12 +11,11 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import io
import operator
import tempfile
import uuid
import six
from openstack import exceptions
from openstack.cloud import exc
from openstack.cloud import meta
@ -75,7 +74,7 @@ class TestImage(BaseTestImage):
self.cloud.download_image, self.image_name)
def test_download_image_two_outputs(self):
fake_fd = six.BytesIO()
fake_fd = io.BytesIO()
self.assertRaises(exc.OpenStackCloudException,
self.cloud.download_image, self.image_name,
output_path='fake_path', output_file=fake_fd)
@ -120,7 +119,7 @@ class TestImage(BaseTestImage):
def test_download_image_with_fd(self):
self._register_image_mocks()
output_file = six.BytesIO()
output_file = io.BytesIO()
self.cloud.download_image(self.image_name, output_file=output_file)
output_file.seek(0)
self.assertEqual(output_file.read(), self.output)

3
openstack/tests/unit/compute/v2/test_server.py
View File

@ -11,7 +11,6 @@
# under the License.
import mock
import six
from openstack.tests.unit import base
from openstack.image.v2 import image
@ -818,7 +817,7 @@ class TestServer(base.TestCase):
self.sess, host='HOST2', force=False, block_migration=False)
self.assertIn(
"Live migration on this cloud implies 'force'",
six.text_type(ex))
str(ex))
def test_live_migrate_no_microversion_force_true(self):
sot = server.Server(**EXAMPLE)

4
openstack/tests/unit/image/v2/test_image.py
View File

@ -10,8 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import hashlib
import io
import operator
import six
import tempfile
from keystoneauth1 import adapter
@ -403,7 +403,7 @@ class TestImage(base.TestCase):
self.assertEqual(rv, resp)
def test_image_download_output_fd(self):
output_file = six.BytesIO()
output_file = io.BytesIO()
sot = image.Image(**EXAMPLE)
response = mock.Mock()
response.status_code = 200

19
openstack/tests/unit/object_store/v1/test_proxy.py
View File

@ -14,7 +14,6 @@ from testscenarios import load_tests_apply_scenarios as load_tests # noqa
from hashlib import sha1
import mock
import random
import six
import string
import tempfile
import time
@ -281,7 +280,7 @@ class TestTempURL(TestObjectStoreProxy):
url = self.proxy.generate_temp_url(
self.url, self.seconds, self.method, temp_url_key=self.key)
key = self.key
if not isinstance(key, six.binary_type):
if not isinstance(key, bytes):
key = key.encode('utf-8')
self.assertEqual(url, self.expected_url)
self.assertEqual(hmac_mock.mock_calls, [
@ -309,10 +308,10 @@ class TestTempURL(TestObjectStoreProxy):
path, self.seconds, self.method,
temp_url_key=self.key, ip_range=ip_range)
key = self.key
if not isinstance(key, six.binary_type):
if not isinstance(key, bytes):
key = key.encode('utf-8')
if isinstance(ip_range, six.binary_type):
if isinstance(ip_range, bytes):
ip_range_expected_url = (
expected_url + ip_range.decode('utf-8')
)
@ -357,7 +356,7 @@ class TestTempURL(TestObjectStoreProxy):
lt = time.localtime()
expires = time.strftime(self.expires_iso8601_format[:-1], lt)
if not isinstance(self.expected_url, six.string_types):
if not isinstance(self.expected_url, str):
expected_url = self.expected_url.replace(
b'1400003600', bytes(str(int(time.mktime(lt))),
encoding='ascii'))
@ -372,7 +371,7 @@ class TestTempURL(TestObjectStoreProxy):
expires = time.strftime(self.short_expires_iso8601_format, lt)
lt = time.strptime(expires, self.short_expires_iso8601_format)
if not isinstance(self.expected_url, six.string_types):
if not isinstance(self.expected_url, str):
expected_url = self.expected_url.replace(
b'1400003600', bytes(str(int(time.mktime(lt))),
encoding='ascii'))
@ -393,12 +392,12 @@ class TestTempURL(TestObjectStoreProxy):
temp_url_key=self.key,
iso8601=True)
key = self.key
if not isinstance(key, six.binary_type):
if not isinstance(key, bytes):
key = key.encode('utf-8')
expires = time.strftime(self.expires_iso8601_format,
time.gmtime(1400003600))
if not isinstance(self.url, six.string_types):
if not isinstance(self.url, str):
self.assertTrue(url.endswith(bytes(expires, 'utf-8')))
else:
self.assertTrue(url.endswith(expires))
@ -429,7 +428,7 @@ class TestTempURL(TestObjectStoreProxy):
path, self.seconds, self.method, prefix=True,
temp_url_key=self.key)
key = self.key
if not isinstance(key, six.binary_type):
if not isinstance(key, bytes):
key = key.encode('utf-8')
self.assertEqual(url, expected_url)
self.assertEqual(hmac_mock.mock_calls, [
@ -448,7 +447,7 @@ class TestTempURL(TestObjectStoreProxy):
@mock.patch('hmac.HMAC.hexdigest', return_value="temp_url_signature")
def test_generate_absolute_expiry_temp_url(self, hmac_mock):
if isinstance(self.expected_url, six.binary_type):
if isinstance(self.expected_url, bytes):
expected_url = self.expected_url.replace(
b'1400003600', b'2146636800')
else:

5
openstack/tests/unit/orchestration/v1/test_proxy.py
View File

@ -12,7 +12,6 @@
from testscenarios import load_tests_apply_scenarios as load_tests # noqa
import mock
import six
from openstack import exceptions
from openstack.orchestration.v1 import _proxy
@ -253,7 +252,7 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
ex = self.assertRaises(exceptions.ResourceNotFound,
self.proxy.resources, stack_name)
self.assertEqual('No stack found for test_stack', six.text_type(ex))
self.assertEqual('No stack found for test_stack', str(ex))
def test_create_software_config(self):
self.verify_create(self.proxy.create_software_config,
@ -320,7 +319,7 @@ class TestOrchestrationProxy(test_proxy_base.TestProxyBase):
self.proxy.validate_template,
None, template_url=None)
self.assertEqual("'template_url' must be specified when template is "
"None", six.text_type(err))
"None", str(err))
class TestExtractName(TestOrchestrationProxy):

6
openstack/tests/unit/orchestration/v1/test_stack.py
View File

@ -11,7 +11,7 @@
# under the License.
import mock
import six
from openstack.tests.unit import base
from openstack.tests.unit import test_resource
@ -223,10 +223,10 @@ class TestStack(base.TestCase):
'stacks/{id}?resolve_outputs=False'.format(id=sot.id),
microversion=None)
ex = self.assertRaises(exceptions.ResourceNotFound, sot.fetch, sess)
self.assertEqual('oops', six.text_type(ex))
self.assertEqual('oops', str(ex))
ex = self.assertRaises(exceptions.ResourceNotFound, sot.fetch, sess)
self.assertEqual('No stack found for %s' % FAKE_ID,
six.text_type(ex))
str(ex))
def test_abandon(self):
sess = mock.Mock()

3
openstack/tests/unit/test_resource.py
View File

@ -17,7 +17,6 @@ from keystoneauth1 import adapter
import mock
import munch
import requests
import six
from openstack import exceptions
from openstack import format
@ -960,7 +959,7 @@ class TestResource(base.TestCase):
body=False, headers=False, computed=False)
self.assertEqual(
'At least one of `body`, `headers` or `computed` must be True',
six.text_type(err))
str(err))
def test_to_dict_with_mro_no_override(self):

6
openstack/utils.py
View File

@ -15,8 +15,6 @@ import string
import threading
import time
import six
import keystoneauth1
from keystoneauth1 import discover
@ -31,7 +29,7 @@ def urljoin(*args):
like /path this should be joined to http://host/path as it is an anchored
link. We generally won't care about that in client.
"""
return '/'.join(six.text_type(a or '').strip('/') for a in args)
return '/'.join(str(a or '').strip('/') for a in args)
def iterate_timeout(timeout, message, wait=2):
@ -194,7 +192,7 @@ def maximum_supported_microversion(adapter, client_maximum):
return discover.version_to_string(result)
class TinyDAG(six.Iterator):
class TinyDAG(object):
"""Tiny DAG
Bases on the Kahn's algorithm, and enables parallel visiting of the nodes

1
requirements.txt
View File

@ -6,7 +6,6 @@ PyYAML>=3.12 # MIT
appdirs>=1.3.0 # MIT License
requestsexceptions>=1.2.0 # Apache-2.0
jsonpatch!=1.20,>=1.16 # BSD
six>=1.10.0 # MIT
os-service-types>=1.7.0 # Apache-2.0
keystoneauth1>=3.18.0 # Apache-2.0