# Copyright (c) 2014 Clinton Knight. All rights reserved. # Copyright (c) 2015 Tom Barron. All rights reserved. # Copyright (c) 2016 Michael Price. All rights reserved. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. """ Mock unit tests for the NetApp driver utility module """ import copy import ddt import platform import mock from oslo_concurrency import processutils as putils from cinder import context from cinder import exception from cinder import test from cinder.tests.unit.volume.drivers.netapp.dataontap.client import ( fakes as zapi_fakes) import cinder.tests.unit.volume.drivers.netapp.fakes as fake from cinder import version from cinder.volume.drivers.netapp.dataontap.client import api as netapp_api from cinder.volume.drivers.netapp import utils as na_utils from cinder.volume import qos_specs from cinder.volume import volume_types @ddt.ddt class NetAppDriverUtilsTestCase(test.TestCase): @mock.patch.object(na_utils, 'LOG', mock.Mock()) def test_validate_instantiation_proxy(self): kwargs = {'netapp_mode': 'proxy'} na_utils.validate_instantiation(**kwargs) self.assertEqual(0, na_utils.LOG.warning.call_count) @mock.patch.object(na_utils, 'LOG', mock.Mock()) def test_validate_instantiation_no_proxy(self): kwargs = {'netapp_mode': 'asdf'} na_utils.validate_instantiation(**kwargs) self.assertEqual(1, na_utils.LOG.warning.call_count) def test_check_flags(self): class TestClass(object): pass required_flags = ['flag1', 'flag2'] configuration = TestClass() setattr(configuration, 'flag1', 'value1') setattr(configuration, 'flag3', 'value3') self.assertRaises(exception.InvalidInput, na_utils.check_flags, required_flags, configuration) setattr(configuration, 'flag2', 'value2') self.assertIsNone(na_utils.check_flags(required_flags, configuration)) def test_to_bool(self): self.assertTrue(na_utils.to_bool(True)) self.assertTrue(na_utils.to_bool('true')) self.assertTrue(na_utils.to_bool('yes')) self.assertTrue(na_utils.to_bool('y')) self.assertTrue(na_utils.to_bool(1)) self.assertTrue(na_utils.to_bool('1')) self.assertFalse(na_utils.to_bool(False)) self.assertFalse(na_utils.to_bool('false')) self.assertFalse(na_utils.to_bool('asdf')) self.assertFalse(na_utils.to_bool('no')) self.assertFalse(na_utils.to_bool('n')) self.assertFalse(na_utils.to_bool(0)) self.assertFalse(na_utils.to_bool('0')) self.assertFalse(na_utils.to_bool(2)) self.assertFalse(na_utils.to_bool('2')) def test_set_safe_attr(self): fake_object = mock.Mock() fake_object.fake_attr = None # test initial checks self.assertFalse(na_utils.set_safe_attr(None, fake_object, None)) self.assertFalse(na_utils.set_safe_attr(fake_object, None, None)) self.assertFalse(na_utils.set_safe_attr(fake_object, 'fake_attr', None)) # test value isn't changed if it shouldn't be and retval is False fake_object.fake_attr = 'fake_value' self.assertFalse(na_utils.set_safe_attr(fake_object, 'fake_attr', 'fake_value')) self.assertEqual('fake_value', fake_object.fake_attr) # test value is changed if it should be and retval is True self.assertTrue(na_utils.set_safe_attr(fake_object, 'fake_attr', 'new_fake_value')) self.assertEqual('new_fake_value', fake_object.fake_attr) def test_round_down(self): self.assertAlmostEqual(na_utils.round_down(5.567), 5.56) self.assertAlmostEqual(na_utils.round_down(5.567, '0.00'), 5.56) self.assertAlmostEqual(na_utils.round_down(5.567, '0.0'), 5.5) self.assertAlmostEqual(na_utils.round_down(5.567, '0'), 5) self.assertAlmostEqual(na_utils.round_down(0, '0.00'), 0) self.assertAlmostEqual(na_utils.round_down(-5.567), -5.56) self.assertAlmostEqual(na_utils.round_down(-5.567, '0.00'), -5.56) self.assertAlmostEqual(na_utils.round_down(-5.567, '0.0'), -5.5) self.assertAlmostEqual(na_utils.round_down(-5.567, '0'), -5) def test_iscsi_connection_properties(self): actual_properties = na_utils.get_iscsi_connection_properties( fake.ISCSI_FAKE_LUN_ID, fake.ISCSI_FAKE_VOLUME, [fake.ISCSI_FAKE_IQN, fake.ISCSI_FAKE_IQN2], [fake.ISCSI_FAKE_ADDRESS_IPV4, fake.ISCSI_FAKE_ADDRESS2_IPV4], [fake.ISCSI_FAKE_PORT, fake.ISCSI_FAKE_PORT]) actual_properties_mapped = actual_properties['data'] self.assertDictEqual(actual_properties_mapped, fake.ISCSI_MP_TARGET_INFO_DICT) def test_iscsi_connection_properties_single_iqn(self): actual_properties = na_utils.get_iscsi_connection_properties( fake.ISCSI_FAKE_LUN_ID, fake.ISCSI_FAKE_VOLUME, fake.ISCSI_FAKE_IQN, [fake.ISCSI_FAKE_ADDRESS_IPV4, fake.ISCSI_FAKE_ADDRESS2_IPV4], [fake.ISCSI_FAKE_PORT, fake.ISCSI_FAKE_PORT]) actual_properties_mapped = actual_properties['data'] expected = fake.ISCSI_MP_TARGET_INFO_DICT.copy() expected['target_iqns'][1] = expected['target_iqns'][0] self.assertDictEqual(actual_properties_mapped, fake.ISCSI_MP_TARGET_INFO_DICT) def test_iscsi_connection_lun_id_type_str(self): FAKE_LUN_ID = '1' actual_properties = na_utils.get_iscsi_connection_properties( FAKE_LUN_ID, fake.ISCSI_FAKE_VOLUME, fake.ISCSI_FAKE_IQN, [fake.ISCSI_FAKE_ADDRESS_IPV4], [fake.ISCSI_FAKE_PORT]) actual_properties_mapped = actual_properties['data'] self.assertIs(int, type(actual_properties_mapped['target_lun'])) self.assertDictEqual(actual_properties_mapped, fake.FC_ISCSI_TARGET_INFO_DICT) def test_iscsi_connection_lun_id_type_dict(self): FAKE_LUN_ID = {'id': 'fake_id'} self.assertRaises(TypeError, na_utils.get_iscsi_connection_properties, FAKE_LUN_ID, fake.ISCSI_FAKE_VOLUME, fake.ISCSI_FAKE_IQN, [fake.ISCSI_FAKE_ADDRESS_IPV4], [fake.ISCSI_FAKE_PORT]) def test_iscsi_connection_properties_ipv6(self): actual_properties = na_utils.get_iscsi_connection_properties( '1', fake.ISCSI_FAKE_VOLUME_NO_AUTH, fake.ISCSI_FAKE_IQN, [fake.ISCSI_FAKE_ADDRESS_IPV6], [fake.ISCSI_FAKE_PORT]) self.assertDictEqual(actual_properties['data'], fake.FC_ISCSI_TARGET_INFO_DICT_IPV6) def test_get_volume_extra_specs(self): fake_extra_specs = {'fake_key': 'fake_value'} fake_volume_type = {'extra_specs': fake_extra_specs} fake_volume = {'volume_type_id': 'fake_volume_type_id'} self.mock_object(context, 'get_admin_context') self.mock_object(volume_types, 'get_volume_type', return_value=fake_volume_type) self.mock_object(na_utils, 'log_extra_spec_warnings') result = na_utils.get_volume_extra_specs(fake_volume) self.assertEqual(fake_extra_specs, result) def test_trace_filter_func_api(self): na_utils.setup_api_trace_pattern("^(?!(perf)).*$") na_element = zapi_fakes.FAKE_NA_ELEMENT all_args = {'na_element': na_element} self.assertTrue(na_utils.trace_filter_func_api(all_args)) def test_trace_filter_func_api_invalid(self): all_args = {'fake': 'not_na_element'} self.assertTrue(na_utils.trace_filter_func_api(all_args)) def test_trace_filter_func_api_filtered(self): na_utils.setup_api_trace_pattern("^(?!(perf)).*$") na_element = netapp_api.NaElement("perf-object-counter-list-info") all_args = {'na_element': na_element} self.assertFalse(na_utils.trace_filter_func_api(all_args)) def test_get_volume_extra_specs_no_type_id(self): fake_volume = {} self.mock_object(context, 'get_admin_context') self.mock_object(volume_types, 'get_volume_type') self.mock_object(na_utils, 'log_extra_spec_warnings') result = na_utils.get_volume_extra_specs(fake_volume) self.assertEqual({}, result) def test_get_volume_extra_specs_no_volume_type(self): fake_volume = {'volume_type_id': 'fake_volume_type_id'} self.mock_object(context, 'get_admin_context') self.mock_object(volume_types, 'get_volume_type', return_value=None) self.mock_object(na_utils, 'log_extra_spec_warnings') result = na_utils.get_volume_extra_specs(fake_volume) self.assertEqual({}, result) def test_log_extra_spec_warnings_obsolete_specs(self): mock_log = self.mock_object(na_utils.LOG, 'warning') na_utils.log_extra_spec_warnings({'netapp:raid_type': 'raid4'}) self.assertEqual(1, mock_log.call_count) def test_log_extra_spec_warnings_deprecated_specs(self): mock_log = self.mock_object(na_utils.LOG, 'warning') na_utils.log_extra_spec_warnings({'netapp_thick_provisioned': 'true'}) self.assertEqual(1, mock_log.call_count) def test_validate_qos_spec_none(self): qos_spec = None # Just return without raising an exception. na_utils.validate_qos_spec(qos_spec) def test_validate_qos_spec_keys_weirdly_cased(self): qos_spec = {'mAxIopS': 33000} # Just return without raising an exception. na_utils.validate_qos_spec(qos_spec) def test_validate_qos_spec_bad_key(self): qos_spec = {'maxFlops': 33000} self.assertRaises(exception.Invalid, na_utils.validate_qos_spec, qos_spec) def test_validate_qos_spec_bad_key_combination(self): qos_spec = {'maxIOPS': 33000, 'maxBPS': 10000000} self.assertRaises(exception.Invalid, na_utils.validate_qos_spec, qos_spec) def test_map_qos_spec_none(self): qos_spec = None result = na_utils.map_qos_spec(qos_spec, fake.VOLUME) self.assertIsNone(result) def test_map_qos_spec_maxiops(self): qos_spec = {'maxIOPs': 33000} mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name') mock_get_name.return_value = 'fake_qos_policy' expected = { 'policy_name': 'fake_qos_policy', 'max_throughput': '33000iops', } result = na_utils.map_qos_spec(qos_spec, fake.VOLUME) self.assertEqual(expected, result) def test_map_qos_spec_maxiopspergib(self): qos_spec = {'maxIOPSperGiB': 1000} mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name') mock_get_name.return_value = 'fake_qos_policy' expected = { 'policy_name': 'fake_qos_policy', 'max_throughput': '42000iops', } result = na_utils.map_qos_spec(qos_spec, fake.VOLUME) self.assertEqual(expected, result) def test_map_qos_spec_maxbps(self): qos_spec = {'maxBPS': 1000000} mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name') mock_get_name.return_value = 'fake_qos_policy' expected = { 'policy_name': 'fake_qos_policy', 'max_throughput': '1000000B/s', } result = na_utils.map_qos_spec(qos_spec, fake.VOLUME) self.assertEqual(expected, result) def test_map_qos_spec_maxbpspergib(self): qos_spec = {'maxBPSperGiB': 100000} mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name') mock_get_name.return_value = 'fake_qos_policy' expected = { 'policy_name': 'fake_qos_policy', 'max_throughput': '4200000B/s', } result = na_utils.map_qos_spec(qos_spec, fake.VOLUME) self.assertEqual(expected, result) def test_map_qos_spec_no_key_present(self): qos_spec = {} mock_get_name = self.mock_object(na_utils, 'get_qos_policy_group_name') mock_get_name.return_value = 'fake_qos_policy' expected = { 'policy_name': 'fake_qos_policy', 'max_throughput': None, } result = na_utils.map_qos_spec(qos_spec, fake.VOLUME) self.assertEqual(expected, result) def test_map_dict_to_lower(self): original = {'UPperKey': 'Value'} expected = {'upperkey': 'Value'} result = na_utils.map_dict_to_lower(original) self.assertEqual(expected, result) def test_get_qos_policy_group_name(self): expected = 'openstack-%s' % fake.VOLUME_ID result = na_utils.get_qos_policy_group_name(fake.VOLUME) self.assertEqual(expected, result) def test_get_qos_policy_group_name_no_id(self): volume = copy.deepcopy(fake.VOLUME) del(volume['id']) result = na_utils.get_qos_policy_group_name(volume) self.assertIsNone(result) def test_get_qos_policy_group_name_from_info(self): expected = 'openstack-%s' % fake.VOLUME_ID result = na_utils.get_qos_policy_group_name_from_info( fake.QOS_POLICY_GROUP_INFO) self.assertEqual(expected, result) def test_get_qos_policy_group_name_from_info_no_info(self): result = na_utils.get_qos_policy_group_name_from_info(None) self.assertIsNone(result) def test_get_qos_policy_group_name_from_legacy_info(self): expected = fake.QOS_POLICY_GROUP_NAME result = na_utils.get_qos_policy_group_name_from_info( fake.LEGACY_QOS_POLICY_GROUP_INFO) self.assertEqual(expected, result) def test_get_qos_policy_group_name_from_spec_info(self): expected = 'openstack-%s' % fake.VOLUME_ID result = na_utils.get_qos_policy_group_name_from_info( fake.QOS_POLICY_GROUP_INFO) self.assertEqual(expected, result) def test_get_qos_policy_group_name_from_none_qos_info(self): expected = None result = na_utils.get_qos_policy_group_name_from_info( fake.QOS_POLICY_GROUP_INFO_NONE) self.assertEqual(expected, result) def test_get_valid_qos_policy_group_info_exception_path(self): mock_get_volume_type = self.mock_object(na_utils, 'get_volume_type_from_volume') mock_get_volume_type.side_effect = exception.VolumeTypeNotFound expected = fake.QOS_POLICY_GROUP_INFO_NONE result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME) self.assertEqual(expected, result) def test_get_valid_qos_policy_group_info_volume_type_none(self): mock_get_volume_type = self.mock_object(na_utils, 'get_volume_type_from_volume') mock_get_volume_type.return_value = None expected = fake.QOS_POLICY_GROUP_INFO_NONE result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME) self.assertEqual(expected, result) def test_get_valid_qos_policy_group_info_no_info(self): mock_get_volume_type = self.mock_object(na_utils, 'get_volume_type_from_volume') mock_get_volume_type.return_value = fake.VOLUME_TYPE mock_get_legacy_qos_policy = self.mock_object(na_utils, 'get_legacy_qos_policy') mock_get_legacy_qos_policy.return_value = None mock_get_valid_qos_spec_from_volume_type = self.mock_object( na_utils, 'get_valid_backend_qos_spec_from_volume_type') mock_get_valid_qos_spec_from_volume_type.return_value = None self.mock_object(na_utils, 'check_for_invalid_qos_spec_combination') expected = fake.QOS_POLICY_GROUP_INFO_NONE result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME) self.assertEqual(expected, result) def test_get_valid_legacy_qos_policy_group_info(self): mock_get_volume_type = self.mock_object(na_utils, 'get_volume_type_from_volume') mock_get_volume_type.return_value = fake.VOLUME_TYPE mock_get_legacy_qos_policy = self.mock_object(na_utils, 'get_legacy_qos_policy') mock_get_legacy_qos_policy.return_value = fake.LEGACY_QOS mock_get_valid_qos_spec_from_volume_type = self.mock_object( na_utils, 'get_valid_backend_qos_spec_from_volume_type') mock_get_valid_qos_spec_from_volume_type.return_value = None self.mock_object(na_utils, 'check_for_invalid_qos_spec_combination') result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME) self.assertEqual(fake.LEGACY_QOS_POLICY_GROUP_INFO, result) def test_get_valid_spec_qos_policy_group_info(self): mock_get_volume_type = self.mock_object(na_utils, 'get_volume_type_from_volume') mock_get_volume_type.return_value = fake.VOLUME_TYPE mock_get_legacy_qos_policy = self.mock_object(na_utils, 'get_legacy_qos_policy') mock_get_legacy_qos_policy.return_value = None mock_get_valid_qos_spec_from_volume_type = self.mock_object( na_utils, 'get_valid_backend_qos_spec_from_volume_type') mock_get_valid_qos_spec_from_volume_type.return_value =\ fake.QOS_POLICY_GROUP_SPEC self.mock_object(na_utils, 'check_for_invalid_qos_spec_combination') result = na_utils.get_valid_qos_policy_group_info(fake.VOLUME) self.assertEqual(fake.QOS_POLICY_GROUP_INFO, result) def test_get_valid_backend_qos_spec_from_volume_type_no_spec(self): mock_get_spec = self.mock_object( na_utils, 'get_backend_qos_spec_from_volume_type') mock_get_spec.return_value = None mock_validate = self.mock_object(na_utils, 'validate_qos_spec') result = na_utils.get_valid_backend_qos_spec_from_volume_type( fake.VOLUME, fake.VOLUME_TYPE) self.assertIsNone(result) self.assertEqual(0, mock_validate.call_count) def test_get_valid_backend_qos_spec_from_volume_type(self): mock_get_spec = self.mock_object( na_utils, 'get_backend_qos_spec_from_volume_type') mock_get_spec.return_value = fake.QOS_SPEC mock_validate = self.mock_object(na_utils, 'validate_qos_spec') result = na_utils.get_valid_backend_qos_spec_from_volume_type( fake.VOLUME, fake.VOLUME_TYPE) self.assertEqual(fake.QOS_POLICY_GROUP_SPEC, result) self.assertEqual(1, mock_validate.call_count) def test_get_backend_qos_spec_from_volume_type_no_qos_specs_id(self): volume_type = copy.deepcopy(fake.VOLUME_TYPE) del(volume_type['qos_specs_id']) mock_get_context = self.mock_object(context, 'get_admin_context') result = na_utils.get_backend_qos_spec_from_volume_type(volume_type) self.assertIsNone(result) self.assertEqual(0, mock_get_context.call_count) def test_get_backend_qos_spec_from_volume_type_no_qos_spec(self): volume_type = fake.VOLUME_TYPE self.mock_object(context, 'get_admin_context') mock_get_specs = self.mock_object(qos_specs, 'get_qos_specs') mock_get_specs.return_value = None result = na_utils.get_backend_qos_spec_from_volume_type(volume_type) self.assertIsNone(result) def test_get_backend_qos_spec_from_volume_type_with_frontend_spec(self): volume_type = fake.VOLUME_TYPE self.mock_object(context, 'get_admin_context') mock_get_specs = self.mock_object(qos_specs, 'get_qos_specs') mock_get_specs.return_value = fake.OUTER_FRONTEND_QOS_SPEC result = na_utils.get_backend_qos_spec_from_volume_type(volume_type) self.assertIsNone(result) def test_get_backend_qos_spec_from_volume_type_with_backend_spec(self): volume_type = fake.VOLUME_TYPE self.mock_object(context, 'get_admin_context') mock_get_specs = self.mock_object(qos_specs, 'get_qos_specs') mock_get_specs.return_value = fake.OUTER_BACKEND_QOS_SPEC result = na_utils.get_backend_qos_spec_from_volume_type(volume_type) self.assertEqual(fake.QOS_SPEC, result) def test_get_backend_qos_spec_from_volume_type_with_both_spec(self): volume_type = fake.VOLUME_TYPE self.mock_object(context, 'get_admin_context') mock_get_specs = self.mock_object(qos_specs, 'get_qos_specs') mock_get_specs.return_value = fake.OUTER_BOTH_QOS_SPEC result = na_utils.get_backend_qos_spec_from_volume_type(volume_type) self.assertEqual(fake.QOS_SPEC, result) def test_check_for_invalid_qos_spec_combination(self): self.assertRaises(exception.Invalid, na_utils.check_for_invalid_qos_spec_combination, fake.INVALID_QOS_POLICY_GROUP_INFO, fake.VOLUME_TYPE) def test_get_legacy_qos_policy(self): extra_specs = fake.LEGACY_EXTRA_SPECS expected = {'policy_name': fake.QOS_POLICY_GROUP_NAME} result = na_utils.get_legacy_qos_policy(extra_specs) self.assertEqual(expected, result) def test_get_legacy_qos_policy_no_policy_name(self): extra_specs = fake.EXTRA_SPECS result = na_utils.get_legacy_qos_policy(extra_specs) self.assertIsNone(result) @ddt.data(("192.168.99.24:/fake/export/path", "192.168.99.24", "/fake/export/path"), ("127.0.0.1:/", "127.0.0.1", "/"), ("[f180::30d9]:/path_to-export/3.1/this folder", "f180::30d9", "/path_to-export/3.1/this folder"), ("[::]:/", "::", "/"), ("[2001:db8::1]:/fake_export", "2001:db8::1", "/fake_export")) @ddt.unpack def test_get_export_host_junction_path(self, share, host, junction_path): result_host, result_path = na_utils.get_export_host_junction_path( share) self.assertEqual(host, result_host) self.assertEqual(junction_path, result_path) @ddt.data("192.14.21.0/wrong_export", "192.14.21.0:8080:/wrong_export" "2001:db8::1:/wrong_export", "[2001:db8::1:/wrong_export", "2001:db8::1]:/wrong_export") def test_get_export_host_junction_path_with_invalid_exports(self, share): self.assertRaises(exception.NetAppDriverException, na_utils.get_export_host_junction_path, share) class OpenStackInfoTestCase(test.TestCase): UNKNOWN_VERSION = 'unknown version' UNKNOWN_RELEASE = 'unknown release' UNKNOWN_VENDOR = 'unknown vendor' UNKNOWN_PLATFORM = 'unknown platform' VERSION_STRING_RET_VAL = 'fake_version_1' RELEASE_STRING_RET_VAL = 'fake_release_1' PLATFORM_RET_VAL = 'fake_platform_1' VERSION_INFO_VERSION = 'fake_version_2' VERSION_INFO_RELEASE = 'fake_release_2' RPM_INFO_VERSION = 'fake_version_3' RPM_INFO_RELEASE = 'fake_release_3' RPM_INFO_VENDOR = 'fake vendor 3' PUTILS_RPM_RET_VAL = ('fake_version_3 fake_release_3 fake vendor 3', '') NO_PKG_FOUND = ('', 'whatever') PUTILS_DPKG_RET_VAL = ('epoch:upstream_version-debian_revision', '') DEB_RLS = 'upstream_version-debian_revision' DEB_VENDOR = 'debian_revision' def test_openstack_info_init(self): info = na_utils.OpenStackInfo() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(version.version_info, 'version_string', mock.Mock(return_value=VERSION_STRING_RET_VAL)) def test_update_version_from_version_string(self): info = na_utils.OpenStackInfo() info._update_version_from_version_string() self.assertEqual(self.VERSION_STRING_RET_VAL, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(version.version_info, 'version_string', mock.Mock(side_effect=Exception)) def test_xcption_in_update_version_from_version_string(self): info = na_utils.OpenStackInfo() info._update_version_from_version_string() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(version.version_info, 'release_string', mock.Mock(return_value=RELEASE_STRING_RET_VAL)) def test_update_release_from_release_string(self): info = na_utils.OpenStackInfo() info._update_release_from_release_string() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.RELEASE_STRING_RET_VAL, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(version.version_info, 'release_string', mock.Mock(side_effect=Exception)) def test_xcption_in_update_release_from_release_string(self): info = na_utils.OpenStackInfo() info._update_release_from_release_string() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(platform, 'platform', mock.Mock(return_value=PLATFORM_RET_VAL)) def test_update_platform(self): info = na_utils.OpenStackInfo() info._update_platform() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.PLATFORM_RET_VAL, info._platform) @mock.patch.object(platform, 'platform', mock.Mock(side_effect=Exception)) def test_xcption_in_update_platform(self): info = na_utils.OpenStackInfo() info._update_platform() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(na_utils.OpenStackInfo, '_get_version_info_version', mock.Mock(return_value=VERSION_INFO_VERSION)) @mock.patch.object(na_utils.OpenStackInfo, '_get_version_info_release', mock.Mock(return_value=VERSION_INFO_RELEASE)) def test_update_info_from_version_info(self): info = na_utils.OpenStackInfo() info._update_info_from_version_info() self.assertEqual(self.VERSION_INFO_VERSION, info._version) self.assertEqual(self.VERSION_INFO_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(na_utils.OpenStackInfo, '_get_version_info_version', mock.Mock(return_value='')) @mock.patch.object(na_utils.OpenStackInfo, '_get_version_info_release', mock.Mock(return_value=None)) def test_no_info_from_version_info(self): info = na_utils.OpenStackInfo() info._update_info_from_version_info() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(na_utils.OpenStackInfo, '_get_version_info_version', mock.Mock(return_value=VERSION_INFO_VERSION)) @mock.patch.object(na_utils.OpenStackInfo, '_get_version_info_release', mock.Mock(side_effect=Exception)) def test_xcption_in_info_from_version_info(self): info = na_utils.OpenStackInfo() info._update_info_from_version_info() self.assertEqual(self.VERSION_INFO_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) @mock.patch.object(putils, 'execute', mock.Mock(return_value=PUTILS_RPM_RET_VAL)) def test_update_info_from_rpm(self): info = na_utils.OpenStackInfo() found_package = info._update_info_from_rpm() self.assertEqual(self.RPM_INFO_VERSION, info._version) self.assertEqual(self.RPM_INFO_RELEASE, info._release) self.assertEqual(self.RPM_INFO_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) self.assertTrue(found_package) @mock.patch.object(putils, 'execute', mock.Mock(return_value=NO_PKG_FOUND)) def test_update_info_from_rpm_no_pkg_found(self): info = na_utils.OpenStackInfo() found_package = info._update_info_from_rpm() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) self.assertFalse(found_package) @mock.patch.object(putils, 'execute', mock.Mock(side_effect=Exception)) def test_xcption_in_update_info_from_rpm(self): info = na_utils.OpenStackInfo() found_package = info._update_info_from_rpm() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) self.assertFalse(found_package) @mock.patch.object(putils, 'execute', mock.Mock(return_value=PUTILS_DPKG_RET_VAL)) def test_update_info_from_dpkg(self): info = na_utils.OpenStackInfo() found_package = info._update_info_from_dpkg() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.DEB_RLS, info._release) self.assertEqual(self.DEB_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) self.assertTrue(found_package) @mock.patch.object(putils, 'execute', mock.Mock(return_value=NO_PKG_FOUND)) def test_update_info_from_dpkg_no_pkg_found(self): info = na_utils.OpenStackInfo() found_package = info._update_info_from_dpkg() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) self.assertFalse(found_package) @mock.patch.object(putils, 'execute', mock.Mock(side_effect=Exception)) def test_xcption_in_update_info_from_dpkg(self): info = na_utils.OpenStackInfo() found_package = info._update_info_from_dpkg() self.assertEqual(self.UNKNOWN_VERSION, info._version) self.assertEqual(self.UNKNOWN_RELEASE, info._release) self.assertEqual(self.UNKNOWN_VENDOR, info._vendor) self.assertEqual(self.UNKNOWN_PLATFORM, info._platform) self.assertFalse(found_package) @mock.patch.object(na_utils.OpenStackInfo, '_update_version_from_version_string', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_release_from_release_string', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_platform', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_info_from_version_info', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_info_from_rpm', mock.Mock(return_value=True)) @mock.patch.object(na_utils.OpenStackInfo, '_update_info_from_dpkg') def test_update_openstack_info_rpm_pkg_found(self, mock_updt_from_dpkg): info = na_utils.OpenStackInfo() info._update_openstack_info() self.assertFalse(mock_updt_from_dpkg.called) @mock.patch.object(na_utils.OpenStackInfo, '_update_version_from_version_string', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_release_from_release_string', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_platform', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_info_from_version_info', mock.Mock()) @mock.patch.object(na_utils.OpenStackInfo, '_update_info_from_rpm', mock.Mock(return_value=False)) @mock.patch.object(na_utils.OpenStackInfo, '_update_info_from_dpkg') def test_update_openstack_info_rpm_pkg_not_found(self, mock_updt_from_dpkg): info = na_utils.OpenStackInfo() info._update_openstack_info() self.assertTrue(mock_updt_from_dpkg.called) @ddt.ddt class FeaturesTestCase(test.TestCase): def setUp(self): super(FeaturesTestCase, self).setUp() self.features = na_utils.Features() def test_init(self): self.assertSetEqual(set(), self.features.defined_features) def test_add_feature_default(self): self.features.add_feature('FEATURE_1') self.assertTrue(self.features.FEATURE_1.supported) self.assertIn('FEATURE_1', self.features.defined_features) @ddt.data(True, False) def test_add_feature(self, value): self.features.add_feature('FEATURE_2', value) self.assertEqual(value, bool(self.features.FEATURE_2)) self.assertEqual(value, self.features.FEATURE_2.supported) self.assertIsNone(self.features.FEATURE_2.minimum_version) self.assertIn('FEATURE_2', self.features.defined_features) @ddt.data((True, '1'), (False, 2), (False, None), (True, None)) @ddt.unpack def test_add_feature_min_version(self, enabled, min_version): self.features.add_feature('FEATURE_2', enabled, min_version=min_version) self.assertEqual(enabled, bool(self.features.FEATURE_2)) self.assertEqual(enabled, self.features.FEATURE_2.supported) self.assertEqual(min_version, self.features.FEATURE_2.minimum_version) self.assertIn('FEATURE_2', self.features.defined_features) @ddt.data('True', 'False', 0, 1, 1.0, None, [], {}, (True,)) def test_add_feature_type_error(self, value): self.assertRaises(TypeError, self.features.add_feature, 'FEATURE_3', value) self.assertNotIn('FEATURE_3', self.features.defined_features) def test_get_attr_missing(self): self.assertRaises(AttributeError, getattr, self.features, 'FEATURE_4') @ddt.ddt class BitSetTestCase(test.TestCase): def test_default(self): self.assertEqual(na_utils.BitSet(0), na_utils.BitSet()) def test_set(self): bitset = na_utils.BitSet(0) bitset.set(16) self.assertEqual(na_utils.BitSet(1 << 16), bitset) def test_unset(self): bitset = na_utils.BitSet(1 << 16) bitset.unset(16) self.assertEqual(na_utils.BitSet(0), bitset) def test_is_set(self): bitset = na_utils.BitSet(1 << 16) self.assertTrue(bool(bitset.is_set(16))) def test_not_equal(self): set1 = na_utils.BitSet(1 << 15) set2 = na_utils.BitSet(1 << 16) self.assertNotEqual(set1, set2) def test_repr(self): raw_val = 1 << 16 actual = repr(na_utils.BitSet(raw_val)) expected = str(raw_val) self.assertEqual(actual, expected) def test_str(self): raw_val = 1 << 16 actual = str(na_utils.BitSet(raw_val)) expected = bin(raw_val) self.assertEqual(actual, expected) def test_int(self): val = 1 << 16 actual = int(int(na_utils.BitSet(val))) self.assertEqual(val, actual) def test_and(self): actual = na_utils.BitSet(1 << 16 | 1 << 15) actual &= 1 << 16 self.assertEqual(na_utils.BitSet(1 << 16), actual) def test_or(self): actual = na_utils.BitSet() actual |= 1 << 16 self.assertEqual(na_utils.BitSet(1 << 16), actual) def test_invert(self): actual = na_utils.BitSet(1 << 16) actual = ~actual self.assertEqual(~(1 << 16), actual) def test_xor(self): actual = na_utils.BitSet(1 << 16) actual ^= 1 << 16 self.assertEqual(na_utils.BitSet(), actual) def test_lshift(self): actual = na_utils.BitSet(1) actual <<= 16 self.assertEqual(na_utils.BitSet(1 << 16), actual) def test_rshift(self): actual = na_utils.BitSet(1 << 16) actual >>= 16 self.assertEqual(na_utils.BitSet(1), actual)