With this change, all unit tests on volume drivers pass on Python 3.
Changes:
* Replace a/b with a//b to get an integer on Python 3.
* Replace map(...) and filter(...) with a list-comprehension to get a
list on Python 3.
* Replace dict.keys() with list(dict.keys()) to get a list
on Python 3.
* Replace (str, int, float, long) with
six.integer_types + (str, float): long type was removed
from Python 3.
* decode_base32_to_hex(): on Python 3, decode encode_hex_to_base32()
to return a Unicode string.
* convert_es_fmt_to_uuid(): encode/decode on Python 3 to pass the
right types to base64.b32decode() (bytes) and uuid.UUID()
(Unicode).
* fakes.py: Change type of XML document from Unicode to bytes: add
b prefix to literal strings (b'...').
* Replace range(a, b) with list(range(a, b)) to get a list
on Python 3.
* tests-py3.txt: add cinder.tests.unit.volume.drivers, all volume
drivers tests now pass on Python 3.
Partial-Implements: blueprint cinder-python3
Change-Id: Iee609f72bbbef3789fa5e970d209047a113d005c
import cinder.volume.drivers.netapp.dataontap.client.api as netapp_api
-FAKE_VOL_XML = """<volume-info xmlns='http://www.netapp.com/filer/admin'>
+FAKE_VOL_XML = b"""<volume-info xmlns='http://www.netapp.com/filer/admin'>
<name>open123</name>
<state>online</state>
<size-total>0</size-total>
<is-invalid>false</is-invalid>
</volume-info>"""
-FAKE_XML1 = """<options>\
+FAKE_XML1 = b"""<options>\
<test1>abc</test1>\
<test2>abc</test2>\
</options>"""
-FAKE_XML2 = """<root><options>somecontent</options></root>"""
+FAKE_XML2 = b"""<root><options>somecontent</options></root>"""
FAKE_NA_ELEMENT = netapp_api.NaElement(etree.XML(FAKE_VOL_XML))
api_name = zapi_fakes.FAKE_API_NAME
invoke_generator = netapp_api.invoke_api(na_server, api_name)
- self.assertRaises(exception.InvalidInput, invoke_generator.next)
+ self.assertRaises(exception.InvalidInput, next, invoke_generator)
@ddt.data({'params': {'na_server': zapi_fakes.FAKE_NA_SERVER,
'api_name': zapi_fakes.FAKE_API_NAME}},
def test_update_performance_cache(self):
- self.perf_library.performance_counters = range(11, 21)
+ self.perf_library.performance_counters = list(range(11, 21))
mock_get_node_utilization_counters = self.mock_object(
self.perf_library, '_get_node_utilization_counters',
self.perf_library.update_performance_cache()
- self.assertEqual(range(12, 22), self.perf_library.performance_counters)
+ self.assertEqual(list(range(12, 22)),
+ self.perf_library.performance_counters)
self.assertEqual(25, self.perf_library.utilization)
mock_get_node_utilization_counters.assert_called_once_with()
mock_get_node_utilization.assert_called_once_with(12, 21, fake.NODE)
def test_update_performance_cache_counters_unavailable(self):
- self.perf_library.performance_counters = range(11, 21)
+ self.perf_library.performance_counters = list(range(11, 21))
self.perf_library.utilization = 55.0
mock_get_node_utilization_counters = self.mock_object(
self.perf_library.update_performance_cache()
- self.assertEqual(range(11, 21),
+ self.assertEqual(list(range(11, 21)),
self.perf_library.performance_counters)
self.assertEqual(55.0, self.perf_library.utilization)
mock_get_node_utilization_counters.assert_called_once_with()
def test_update_performance_cache(self):
self.perf_library.performance_counters = {
- 'node1': range(11, 21),
- 'node2': range(21, 31),
+ 'node1': list(range(11, 21)),
+ 'node2': list(range(21, 31)),
}
mock_get_aggregates_for_pools = self.mock_object(
self.perf_library, '_get_aggregates_for_pools',
self.perf_library.update_performance_cache(self.fake_volumes)
expected_performance_counters = {
- 'node1': range(12, 22),
- 'node2': range(22, 32),
+ 'node1': list(range(12, 22)),
+ 'node2': list(range(22, 32)),
}
self.assertEqual(expected_performance_counters,
self.perf_library.performance_counters)
thick = not nfs_sparsed_volumes
total_capacity_gb = na_utils.round_down(
- fake.TOTAL_BYTES / units.Gi, '0.01')
+ fake.TOTAL_BYTES // units.Gi, '0.01')
free_capacity_gb = na_utils.round_down(
- fake.AVAILABLE_BYTES / units.Gi, '0.01')
+ fake.AVAILABLE_BYTES // units.Gi, '0.01')
provisioned_capacity_gb = total_capacity_gb - free_capacity_gb
capacity = {
'reserved_percentage': fake.RESERVED_PERCENTAGE,
thick = not thin and not nfs_sparsed_volumes
total_capacity_gb = na_utils.round_down(
- fake.TOTAL_BYTES / units.Gi, '0.01')
+ fake.TOTAL_BYTES // units.Gi, '0.01')
free_capacity_gb = na_utils.round_down(
- fake.AVAILABLE_BYTES / units.Gi, '0.01')
+ fake.AVAILABLE_BYTES // units.Gi, '0.01')
provisioned_capacity_gb = total_capacity_gb - free_capacity_gb
capacity = {
'reserved_percentage': fake.RESERVED_PERCENTAGE,
FAKE_MAPPINGS = [{u'lun': 1}]
-FAKE_USED_UP_MAPPINGS = map(lambda n: {u'lun': n}, range(256))
+FAKE_USED_UP_MAPPINGS = [{u'lun': n} for n in range(256)]
FAKE_USED_UP_LUN_ID_DICT = {n: 1 for n in range(256)}
def get_attr_names(self):
"""Returns the list of attribute names."""
attributes = self._element.attrib or {}
- return attributes.keys()
+ return list(attributes.keys())
def add_new_child(self, name, content, convert=False):
"""Add child with tag name and context.
child = NaElement(key)
child.add_child_elem(value)
self.add_child_elem(child)
- elif isinstance(value, (str, int, float, long)):
+ elif isinstance(value, six.integer_types + (str, float)):
self.add_new_child(key, six.text_type(value))
elif isinstance(value, (list, tuple, dict)):
child = NaElement(key)
def get_volume_mappings_for_volume(self, volume):
"""Gets all host mappings for given volume from array."""
mappings = self.get_volume_mappings() or []
- host_maps = filter(lambda x: x.get('volumeRef') == volume['volumeRef'],
- mappings)
- return host_maps
+ return [x for x in mappings
+ if x.get('volumeRef') == volume['volumeRef']]
def get_volume_mappings_for_host(self, host_ref):
"""Gets all volume mappings for given host from array."""
mappings = self.get_volume_mappings() or []
- host_maps = filter(lambda x: x.get('mapRef') == host_ref, mappings)
- return host_maps
+ return [x for x in mappings if x.get('mapRef') == host_ref]
def get_volume_mappings_for_host_group(self, hg_ref):
"""Gets all volume mappings for given host group from array."""
mappings = self.get_volume_mappings() or []
- hg_maps = filter(lambda x: x.get('mapRef') == hg_ref, mappings)
- return hg_maps
+ return [x for x in mappings if x.get('mapRef') == hg_ref]
def create_volume_mapping(self, object_id, target_id, lun):
"""Creates volume mapping on array."""
def convert_uuid_to_es_fmt(uuid_str):
"""Converts uuid to e-series compatible name format."""
uuid_base32 = encode_hex_to_base32(uuid.UUID(six.text_type(uuid_str)).hex)
- return str(uuid_base32.strip(b'='))
+ es_label = uuid_base32.strip(b'=')
+ if six.PY3:
+ es_label = es_label.decode('ascii')
+ return es_label
def convert_es_fmt_to_uuid(es_label):
"""Converts e-series name format to uuid."""
- if es_label.startswith('tmp-'):
+ if isinstance(es_label, six.text_type):
+ es_label = es_label.encode('utf-8')
+ if es_label.startswith(b'tmp-'):
es_label = es_label[4:]
- es_label_b32 = es_label.ljust(32, '=')
- return uuid.UUID(binascii.hexlify(base64.b32decode(es_label_b32)))
+ es_label = es_label.ljust(32, b'=')
+ es_label = binascii.hexlify(base64.b32decode(es_label))
+ if six.PY3:
+ es_label = es_label.decode('ascii')
+ return uuid.UUID(es_label)
cinder.tests.unit.test_vzstorage
cinder.tests.unit.test_xio
cinder.tests.unit.test_zfssa
-cinder.tests.unit.volume.drivers.emc.scaleio
-cinder.tests.unit.volume.drivers.netapp.eseries.test_library
-cinder.tests.unit.volume.drivers.test_fujitsu
-cinder.tests.unit.volume.drivers.test_hgst
+cinder.tests.unit.volume.drivers
cinder.tests.unit.volume.flows.test_create_volume_flow
cinder.tests.unit.windows.test_smbfs
cinder.tests.unit.windows.test_vhdutils
Code Review / openstack-build / cinder-build.git / commitdiff