849 lines
32 KiB
Python
849 lines
32 KiB
Python
# Copyright (c) 2016 Dell Inc. or its subsidiaries.
|
|
# 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.
|
|
|
|
import contextlib
|
|
import functools
|
|
|
|
import mock
|
|
from oslo_utils import units
|
|
|
|
from cinder import exception
|
|
from cinder import test
|
|
from cinder.tests.unit.volume.drivers.dell_emc.unity \
|
|
import fake_exception as ex
|
|
from cinder.tests.unit.volume.drivers.dell_emc.unity import test_client
|
|
from cinder.volume.drivers.dell_emc.unity import adapter
|
|
|
|
|
|
########################
|
|
#
|
|
# Start of Mocks
|
|
#
|
|
########################
|
|
class MockConfig(object):
|
|
def __init__(self):
|
|
self.config_group = 'test_backend'
|
|
self.unity_storage_pool_names = ['pool1', 'pool2']
|
|
self.unity_io_ports = None
|
|
self.reserved_percentage = 5
|
|
self.max_over_subscription_ratio = 300
|
|
self.volume_backend_name = 'backend'
|
|
self.san_ip = '1.2.3.4'
|
|
self.san_login = 'user'
|
|
self.san_password = 'pass'
|
|
self.driver_ssl_cert_verify = False
|
|
self.driver_ssl_cert_path = None
|
|
|
|
def safe_get(self, name):
|
|
return getattr(self, name)
|
|
|
|
|
|
class MockConnector(object):
|
|
@staticmethod
|
|
def disconnect_volume(data, device):
|
|
pass
|
|
|
|
|
|
class MockDriver(object):
|
|
def __init__(self):
|
|
self.configuration = mock.Mock(volume_dd_blocksize='1M')
|
|
|
|
@staticmethod
|
|
def _connect_device(conn):
|
|
return {'connector': MockConnector(),
|
|
'device': {'path': 'dev'},
|
|
'conn': {'data': {}}}
|
|
|
|
|
|
class MockClient(object):
|
|
def __init__(self):
|
|
self._system = test_client.MockSystem()
|
|
|
|
@staticmethod
|
|
def get_pools():
|
|
return test_client.MockResourceList(['pool0', 'pool1'])
|
|
|
|
@staticmethod
|
|
def create_lun(name, size, pool, description=None, io_limit_policy=None):
|
|
return test_client.MockResource(_id=name, name=name)
|
|
|
|
@staticmethod
|
|
def get_lun(name=None, lun_id=None):
|
|
if lun_id is None:
|
|
lun_id = 'lun_4'
|
|
if lun_id in ('lun_43',): # for thin clone cases
|
|
return test_client.MockResource(_id=lun_id, name=name)
|
|
if name == 'not_exists':
|
|
ret = test_client.MockResource(name=lun_id)
|
|
ret.existed = False
|
|
else:
|
|
if name is None:
|
|
name = lun_id
|
|
ret = test_client.MockResource(_id=lun_id, name=name)
|
|
return ret
|
|
|
|
@staticmethod
|
|
def delete_lun(lun_id):
|
|
if lun_id != 'lun_4':
|
|
raise ex.UnexpectedLunDeletion()
|
|
|
|
@staticmethod
|
|
def get_serial():
|
|
return 'CLIENT_SERIAL'
|
|
|
|
@staticmethod
|
|
def create_snap(src_lun_id, name=None):
|
|
if src_lun_id in ('lun_53', 'lun_55'): # for thin clone cases
|
|
return test_client.MockResource(
|
|
_id='snap_clone_{}'.format(src_lun_id))
|
|
return test_client.MockResource(name=name, _id=src_lun_id)
|
|
|
|
@staticmethod
|
|
def get_snap(name=None):
|
|
if name in ('snap_50',): # for thin clone cases
|
|
return name
|
|
snap = test_client.MockResource(name=name, _id=name)
|
|
if name is not None:
|
|
ret = snap
|
|
else:
|
|
ret = [snap]
|
|
return ret
|
|
|
|
@staticmethod
|
|
def delete_snap(snap):
|
|
if snap.name in ('abc-def_snap',):
|
|
raise ex.SnapDeleteIsCalled()
|
|
|
|
@staticmethod
|
|
def create_host(name):
|
|
return test_client.MockResource(name=name)
|
|
|
|
@staticmethod
|
|
def attach(host, lun_or_snap):
|
|
return 10
|
|
|
|
@staticmethod
|
|
def detach(host, lun_or_snap):
|
|
error_ids = ['lun_43', 'snap_0']
|
|
if host.name == 'host1' and lun_or_snap.get_id() in error_ids:
|
|
raise ex.DetachIsCalled()
|
|
|
|
@staticmethod
|
|
def get_iscsi_target_info(allowed_ports=None):
|
|
return [{'portal': '1.2.3.4:1234', 'iqn': 'iqn.1-1.com.e:c.a.a0'},
|
|
{'portal': '1.2.3.5:1234', 'iqn': 'iqn.1-1.com.e:c.a.a1'}]
|
|
|
|
@staticmethod
|
|
def get_fc_target_info(host=None, logged_in_only=False,
|
|
allowed_ports=None):
|
|
if host and host.name == 'no_target':
|
|
ret = []
|
|
else:
|
|
ret = ['8899AABBCCDDEEFF', '8899AABBCCDDFFEE']
|
|
return ret
|
|
|
|
@staticmethod
|
|
def create_lookup_service():
|
|
return {}
|
|
|
|
@staticmethod
|
|
def get_io_limit_policy(specs):
|
|
return None
|
|
|
|
@staticmethod
|
|
def extend_lun(lun_id, size_gib):
|
|
if size_gib <= 0:
|
|
raise ex.ExtendLunError
|
|
|
|
@staticmethod
|
|
def get_fc_ports():
|
|
return test_client.MockResourceList(ids=['spa_iom_0_fc0',
|
|
'spa_iom_0_fc1'])
|
|
|
|
@staticmethod
|
|
def get_ethernet_ports():
|
|
return test_client.MockResourceList(ids=['spa_eth0', 'spb_eth0'])
|
|
|
|
@staticmethod
|
|
def thin_clone(obj, name, io_limit_policy, description, new_size_gb):
|
|
if (obj.name, name) in (
|
|
('snap_61', 'lun_60'), ('lun_63', 'lun_60')):
|
|
return test_client.MockResource(_id=name)
|
|
else:
|
|
raise ex.UnityThinCloneLimitExceededError
|
|
|
|
@staticmethod
|
|
def update_host_initiators(host, wwns):
|
|
return None
|
|
|
|
@property
|
|
def system(self):
|
|
return self._system
|
|
|
|
|
|
class MockLookupService(object):
|
|
@staticmethod
|
|
def get_device_mapping_from_network(initiator_wwns, target_wwns):
|
|
return {
|
|
'san_1': {
|
|
'initiator_port_wwn_list':
|
|
('200000051e55a100', '200000051e55a121'),
|
|
'target_port_wwn_list':
|
|
('100000051e55a100', '100000051e55a121')
|
|
}
|
|
}
|
|
|
|
|
|
class MockOSResource(mock.Mock):
|
|
def __init__(self, *args, **kwargs):
|
|
super(MockOSResource, self).__init__(*args, **kwargs)
|
|
if 'name' in kwargs:
|
|
self.name = kwargs['name']
|
|
|
|
|
|
def mock_adapter(driver_clz):
|
|
ret = driver_clz()
|
|
ret._client = MockClient()
|
|
with mock.patch('cinder.volume.drivers.dell_emc.unity.adapter.'
|
|
'CommonAdapter.validate_ports'), \
|
|
patch_storops():
|
|
ret.do_setup(MockDriver(), MockConfig())
|
|
ret.lookup_service = MockLookupService()
|
|
return ret
|
|
|
|
|
|
def get_backend_qos_specs(volume):
|
|
return None
|
|
|
|
|
|
def get_connector_properties():
|
|
return {'host': 'host1', 'wwpns': 'abcdefg'}
|
|
|
|
|
|
def get_lun_pl(name):
|
|
return 'id^%s|system^CLIENT_SERIAL|type^lun|version^None' % name
|
|
|
|
|
|
def get_snap_lun_pl(name):
|
|
return 'id^%s|system^CLIENT_SERIAL|type^snap_lun|version^None' % name
|
|
|
|
|
|
def get_snap_pl(name):
|
|
return 'id^%s|system^CLIENT_SERIAL|type^snapshot|version^None' % name
|
|
|
|
|
|
def get_connector_uids(adapter, connector):
|
|
return []
|
|
|
|
|
|
def get_connection_info(adapter, hlu, host, connector):
|
|
return {}
|
|
|
|
|
|
def patch_for_unity_adapter(func):
|
|
@functools.wraps(func)
|
|
@mock.patch('cinder.volume.drivers.dell_emc.unity.utils.'
|
|
'get_backend_qos_specs',
|
|
new=get_backend_qos_specs)
|
|
@mock.patch('cinder.utils.brick_get_connector_properties',
|
|
new=get_connector_properties)
|
|
def func_wrapper(*args, **kwargs):
|
|
return func(*args, **kwargs)
|
|
|
|
return func_wrapper
|
|
|
|
|
|
def patch_for_concrete_adapter(clz_str):
|
|
def inner_decorator(func):
|
|
@functools.wraps(func)
|
|
@mock.patch('%s.get_connector_uids' % clz_str,
|
|
new=get_connector_uids)
|
|
@mock.patch('%s.get_connection_info' % clz_str,
|
|
new=get_connection_info)
|
|
def func_wrapper(*args, **kwargs):
|
|
return func(*args, **kwargs)
|
|
return func_wrapper
|
|
|
|
return inner_decorator
|
|
|
|
|
|
patch_for_iscsi_adapter = patch_for_concrete_adapter(
|
|
'cinder.volume.drivers.dell_emc.unity.adapter.ISCSIAdapter')
|
|
|
|
|
|
patch_for_fc_adapter = patch_for_concrete_adapter(
|
|
'cinder.volume.drivers.dell_emc.unity.adapter.FCAdapter')
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def patch_thin_clone(cloned_lun):
|
|
with mock.patch.object(adapter.CommonAdapter, '_thin_clone') as tc:
|
|
tc.return_value = cloned_lun
|
|
yield tc
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def patch_dd_copy(copied_lun):
|
|
with mock.patch.object(adapter.CommonAdapter, '_dd_copy') as dd:
|
|
dd.return_value = copied_lun
|
|
yield dd
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def patch_copy_volume():
|
|
with mock.patch('cinder.volume.utils.copy_volume') as mocked:
|
|
yield mocked
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def patch_storops():
|
|
with mock.patch.object(adapter, 'storops') as storops:
|
|
storops.ThinCloneActionEnum = mock.Mock(DD_COPY='DD_COPY')
|
|
yield storops
|
|
|
|
|
|
class IdMatcher(object):
|
|
def __init__(self, obj):
|
|
self._obj = obj
|
|
|
|
def __eq__(self, other):
|
|
return self._obj._id == other._id
|
|
|
|
|
|
########################
|
|
#
|
|
# Start of Tests
|
|
#
|
|
########################
|
|
|
|
@mock.patch.object(adapter, 'storops_ex', new=ex)
|
|
class CommonAdapterTest(test.TestCase):
|
|
def setUp(self):
|
|
super(CommonAdapterTest, self).setUp()
|
|
self.adapter = mock_adapter(adapter.CommonAdapter)
|
|
|
|
def test_get_managed_pools(self):
|
|
ret = self.adapter.get_managed_pools()
|
|
self.assertIn('pool1', ret)
|
|
self.assertNotIn('pool0', ret)
|
|
self.assertNotIn('pool2', ret)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_create_volume(self):
|
|
volume = MockOSResource(name='lun_3', size=5, host='unity#pool1')
|
|
ret = self.adapter.create_volume(volume)
|
|
expected = get_lun_pl('lun_3')
|
|
self.assertEqual(expected, ret['provider_location'])
|
|
|
|
def test_create_snapshot(self):
|
|
volume = MockOSResource(provider_location='id^lun_43')
|
|
snap = MockOSResource(volume=volume, name='abc-def_snap')
|
|
result = self.adapter.create_snapshot(snap)
|
|
self.assertEqual(get_snap_pl('lun_43'), result['provider_location'])
|
|
self.assertEqual('lun_43', result['provider_id'])
|
|
|
|
def test_delete_snap(self):
|
|
def f():
|
|
snap = MockOSResource(name='abc-def_snap')
|
|
self.adapter.delete_snapshot(snap)
|
|
|
|
self.assertRaises(ex.SnapDeleteIsCalled, f)
|
|
|
|
def test_get_lun_id_has_location(self):
|
|
volume = MockOSResource(provider_location='id^lun_43')
|
|
self.assertEqual('lun_43', self.adapter.get_lun_id(volume))
|
|
|
|
def test_get_lun_id_no_location(self):
|
|
volume = MockOSResource(provider_location=None)
|
|
self.assertEqual('lun_4', self.adapter.get_lun_id(volume))
|
|
|
|
def test_delete_volume(self):
|
|
volume = MockOSResource(provider_location='id^lun_4')
|
|
self.adapter.delete_volume(volume)
|
|
|
|
def test_get_pool_stats(self):
|
|
stats_list = self.adapter.get_pools_stats()
|
|
self.assertEqual(1, len(stats_list))
|
|
|
|
stats = stats_list[0]
|
|
self.assertEqual('pool1', stats['pool_name'])
|
|
self.assertEqual(5, stats['total_capacity_gb'])
|
|
self.assertEqual('pool1|CLIENT_SERIAL', stats['location_info'])
|
|
self.assertEqual(6, stats['provisioned_capacity_gb'])
|
|
self.assertEqual(2, stats['free_capacity_gb'])
|
|
self.assertEqual(300, stats['max_over_subscription_ratio'])
|
|
self.assertEqual(5, stats['reserved_percentage'])
|
|
self.assertFalse(stats['thick_provisioning_support'])
|
|
self.assertTrue(stats['thin_provisioning_support'])
|
|
|
|
def test_update_volume_stats(self):
|
|
stats = self.adapter.update_volume_stats()
|
|
self.assertEqual('backend', stats['volume_backend_name'])
|
|
self.assertEqual('unknown', stats['storage_protocol'])
|
|
self.assertTrue(stats['thin_provisioning_support'])
|
|
self.assertFalse(stats['thick_provisioning_support'])
|
|
self.assertEqual(1, len(stats['pools']))
|
|
|
|
def test_serial_number(self):
|
|
self.assertEqual('CLIENT_SERIAL', self.adapter.serial_number)
|
|
|
|
def test_do_setup(self):
|
|
self.assertEqual('1.2.3.4', self.adapter.ip)
|
|
self.assertEqual('user', self.adapter.username)
|
|
self.assertEqual('pass', self.adapter.password)
|
|
self.assertFalse(self.adapter.array_cert_verify)
|
|
self.assertIsNone(self.adapter.array_ca_cert_path)
|
|
|
|
def test_do_setup_version_before_4_1(self):
|
|
def f():
|
|
with mock.patch('cinder.volume.drivers.dell_emc.unity.adapter.'
|
|
'CommonAdapter.validate_ports'):
|
|
self.adapter._client.system.system_version = '4.0.0'
|
|
self.adapter.do_setup(self.adapter.driver, MockConfig())
|
|
self.assertRaises(exception.VolumeBackendAPIException, f)
|
|
|
|
def test_verify_cert_false_path_none(self):
|
|
self.adapter.array_cert_verify = False
|
|
self.adapter.array_ca_cert_path = None
|
|
self.assertFalse(self.adapter.verify_cert)
|
|
|
|
def test_verify_cert_false_path_not_none(self):
|
|
self.adapter.array_cert_verify = False
|
|
self.adapter.array_ca_cert_path = '/tmp/array_ca.crt'
|
|
self.assertFalse(self.adapter.verify_cert)
|
|
|
|
def test_verify_cert_true_path_none(self):
|
|
self.adapter.array_cert_verify = True
|
|
self.adapter.array_ca_cert_path = None
|
|
self.assertTrue(self.adapter.verify_cert)
|
|
|
|
def test_verify_cert_true_path_valide(self):
|
|
self.adapter.array_cert_verify = True
|
|
self.adapter.array_ca_cert_path = '/tmp/array_ca.crt'
|
|
self.assertEqual(self.adapter.array_ca_cert_path,
|
|
self.adapter.verify_cert)
|
|
|
|
def test_terminate_connection_volume(self):
|
|
def f():
|
|
volume = MockOSResource(provider_location='id^lun_43', id='id_43')
|
|
connector = {'host': 'host1'}
|
|
self.adapter.terminate_connection(volume, connector)
|
|
|
|
self.assertRaises(ex.DetachIsCalled, f)
|
|
|
|
def test_terminate_connection_snapshot(self):
|
|
def f():
|
|
connector = {'host': 'host1'}
|
|
snap = MockOSResource(name='snap_0', id='snap_0')
|
|
self.adapter.terminate_connection_snapshot(snap, connector)
|
|
|
|
self.assertRaises(ex.DetachIsCalled, f)
|
|
|
|
def test_manage_existing_by_name(self):
|
|
ref = {'source-id': 12}
|
|
volume = MockOSResource(name='lun1')
|
|
ret = self.adapter.manage_existing(volume, ref)
|
|
expected = get_lun_pl('12')
|
|
self.assertEqual(expected, ret['provider_location'])
|
|
|
|
def test_manage_existing_by_id(self):
|
|
ref = {'source-name': 'lunx'}
|
|
volume = MockOSResource(name='lun1')
|
|
ret = self.adapter.manage_existing(volume, ref)
|
|
expected = get_lun_pl('lun_4')
|
|
self.assertEqual(expected, ret['provider_location'])
|
|
|
|
def test_manage_existing_invalid_ref(self):
|
|
def f():
|
|
ref = {}
|
|
volume = MockOSResource(name='lun1')
|
|
self.adapter.manage_existing(volume, ref)
|
|
|
|
self.assertRaises(exception.ManageExistingInvalidReference, f)
|
|
|
|
def test_manage_existing_lun_not_found(self):
|
|
def f():
|
|
ref = {'source-name': 'not_exists'}
|
|
volume = MockOSResource(name='lun1')
|
|
self.adapter.manage_existing(volume, ref)
|
|
|
|
self.assertRaises(exception.ManageExistingInvalidReference, f)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_manage_existing_get_size_invalid_backend(self):
|
|
def f():
|
|
volume = MockOSResource(volume_type_id='thin',
|
|
host='host@backend#pool1')
|
|
ref = {'source-id': 12}
|
|
self.adapter.manage_existing_get_size(volume, ref)
|
|
|
|
self.assertRaises(exception.ManageExistingInvalidReference, f)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_manage_existing_get_size_success(self):
|
|
volume = MockOSResource(volume_type_id='thin',
|
|
host='host@backend#pool0')
|
|
ref = {'source-id': 12}
|
|
volume_size = self.adapter.manage_existing_get_size(volume, ref)
|
|
self.assertEqual(5, volume_size)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_create_volume_from_snapshot(self):
|
|
lun_id = 'lun_50'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1')
|
|
snap_id = 'snap_50'
|
|
snap = MockOSResource(name=snap_id)
|
|
with patch_thin_clone(test_client.MockResource(_id=lun_id)) as tc:
|
|
ret = self.adapter.create_volume_from_snapshot(volume, snap)
|
|
self.assertEqual(get_snap_lun_pl(lun_id),
|
|
ret['provider_location'])
|
|
tc.assert_called_with(adapter.VolumeParams(self.adapter, volume),
|
|
snap_id)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_create_cloned_volume_attached(self):
|
|
lun_id = 'lun_51'
|
|
src_lun_id = 'lun_53'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1')
|
|
src_vref = MockOSResource(id=src_lun_id, name=src_lun_id,
|
|
provider_location=get_lun_pl(src_lun_id),
|
|
volume_attachment=['not_care'])
|
|
with patch_dd_copy(test_client.MockResource(_id=lun_id)) as dd:
|
|
ret = self.adapter.create_cloned_volume(volume, src_vref)
|
|
dd.assert_called_with(
|
|
adapter.VolumeParams(self.adapter, volume),
|
|
IdMatcher(test_client.MockResource(
|
|
_id='snap_clone_{}'.format(src_lun_id))),
|
|
src_lun=IdMatcher(test_client.MockResource(_id=src_lun_id)))
|
|
self.assertEqual(get_lun_pl(lun_id), ret['provider_location'])
|
|
|
|
@patch_for_unity_adapter
|
|
def test_create_cloned_volume_available(self):
|
|
lun_id = 'lun_54'
|
|
src_lun_id = 'lun_55'
|
|
volume = MockOSResource(id=lun_id, host='unity#pool1', size=3,
|
|
provider_location=get_lun_pl(lun_id))
|
|
src_vref = MockOSResource(id=src_lun_id, name=src_lun_id,
|
|
provider_location=get_lun_pl(src_lun_id),
|
|
volume_attachment=None)
|
|
with patch_thin_clone(test_client.MockResource(_id=lun_id)) as tc:
|
|
ret = self.adapter.create_cloned_volume(volume, src_vref)
|
|
tc.assert_called_with(
|
|
adapter.VolumeParams(self.adapter, volume),
|
|
IdMatcher(test_client.MockResource(
|
|
_id='snap_clone_{}'.format(src_lun_id))),
|
|
src_lun=IdMatcher(test_client.MockResource(_id=src_lun_id)))
|
|
self.assertEqual(get_snap_lun_pl(lun_id), ret['provider_location'])
|
|
|
|
@patch_for_unity_adapter
|
|
def test_dd_copy_with_src_lun(self):
|
|
lun_id = 'lun_56'
|
|
src_lun_id = 'lun_57'
|
|
src_snap_id = 'snap_57'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1',
|
|
provider_location=get_lun_pl(lun_id))
|
|
src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
|
|
src_lun = test_client.MockResource(name=src_lun_id, _id=src_lun_id)
|
|
src_lun.size_total = 6 * units.Gi
|
|
with patch_copy_volume() as copy_volume:
|
|
ret = self.adapter._dd_copy(
|
|
adapter.VolumeParams(self.adapter, volume), src_snap,
|
|
src_lun=src_lun)
|
|
copy_volume.assert_called_with('dev', 'dev', 6144, '1M',
|
|
sparse=True)
|
|
self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)),
|
|
ret)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_dd_copy_wo_src_lun(self):
|
|
lun_id = 'lun_58'
|
|
src_lun_id = 'lun_59'
|
|
src_snap_id = 'snap_59'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1',
|
|
provider_location=get_lun_pl(lun_id))
|
|
src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
|
|
src_snap.storage_resource = test_client.MockResource(name=src_lun_id,
|
|
_id=src_lun_id)
|
|
with patch_copy_volume() as copy_volume:
|
|
ret = self.adapter._dd_copy(
|
|
adapter.VolumeParams(self.adapter, volume), src_snap)
|
|
copy_volume.assert_called_with('dev', 'dev', 5120, '1M',
|
|
sparse=True)
|
|
self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)),
|
|
ret)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_dd_copy_raise(self):
|
|
lun_id = 'lun_58'
|
|
src_snap_id = 'snap_59'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, host='unity#pool1',
|
|
provider_location=get_lun_pl(lun_id))
|
|
src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
|
|
with patch_copy_volume() as copy_volume:
|
|
copy_volume.side_effect = AttributeError
|
|
self.assertRaises(AttributeError,
|
|
self.adapter._dd_copy, volume, src_snap)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_thin_clone(self):
|
|
lun_id = 'lun_60'
|
|
src_snap_id = 'snap_61'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, size=1,
|
|
provider_location=get_snap_lun_pl(lun_id))
|
|
src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
|
|
ret = self.adapter._thin_clone(volume, src_snap)
|
|
self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)), ret)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_thin_clone_downgraded_with_src_lun(self):
|
|
lun_id = 'lun_60'
|
|
src_snap_id = 'snap_62'
|
|
src_lun_id = 'lun_62'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, size=1,
|
|
provider_location=get_snap_lun_pl(lun_id))
|
|
src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
|
|
src_lun = test_client.MockResource(name=src_lun_id, _id=src_lun_id)
|
|
new_dd_lun = test_client.MockResource(name='lun_63')
|
|
with patch_storops() as mocked_storops, \
|
|
patch_dd_copy(new_dd_lun) as dd:
|
|
ret = self.adapter._thin_clone(
|
|
adapter.VolumeParams(self.adapter, volume),
|
|
src_snap, src_lun=src_lun)
|
|
vol_params = adapter.VolumeParams(self.adapter, volume)
|
|
vol_params.name = 'hidden-{}'.format(volume.name)
|
|
vol_params.description = 'hidden-{}'.format(volume.description)
|
|
dd.assert_called_with(vol_params, src_snap, src_lun=src_lun)
|
|
mocked_storops.TCHelper.notify.assert_called_with(src_lun,
|
|
'DD_COPY',
|
|
new_dd_lun)
|
|
self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)), ret)
|
|
|
|
@patch_for_unity_adapter
|
|
def test_thin_clone_downgraded_wo_src_lun(self):
|
|
lun_id = 'lun_60'
|
|
src_snap_id = 'snap_62'
|
|
volume = MockOSResource(name=lun_id, id=lun_id, size=1,
|
|
provider_location=get_snap_lun_pl(lun_id))
|
|
src_snap = test_client.MockResource(name=src_snap_id, _id=src_snap_id)
|
|
new_dd_lun = test_client.MockResource(name='lun_63')
|
|
with patch_storops() as mocked_storops, \
|
|
patch_dd_copy(new_dd_lun) as dd:
|
|
ret = self.adapter._thin_clone(
|
|
adapter.VolumeParams(self.adapter, volume), src_snap)
|
|
vol_params = adapter.VolumeParams(self.adapter, volume)
|
|
vol_params.name = 'hidden-{}'.format(volume.name)
|
|
vol_params.description = 'hidden-{}'.format(volume.description)
|
|
dd.assert_called_with(vol_params, src_snap, src_lun=None)
|
|
mocked_storops.TCHelper.notify.assert_called_with(src_snap,
|
|
'DD_COPY',
|
|
new_dd_lun)
|
|
self.assertEqual(IdMatcher(test_client.MockResource(_id=lun_id)), ret)
|
|
|
|
def test_extend_volume_error(self):
|
|
def f():
|
|
volume = MockOSResource(id='l56',
|
|
provider_location=get_lun_pl('lun56'))
|
|
self.adapter.extend_volume(volume, -1)
|
|
|
|
self.assertRaises(ex.ExtendLunError, f)
|
|
|
|
def test_extend_volume_no_id(self):
|
|
def f():
|
|
volume = MockOSResource(provider_location='type^lun')
|
|
self.adapter.extend_volume(volume, 5)
|
|
|
|
self.assertRaises(exception.VolumeBackendAPIException, f)
|
|
|
|
def test_normalize_config(self):
|
|
config = MockConfig()
|
|
config.unity_storage_pool_names = [' pool_1 ', '', ' ']
|
|
config.unity_io_ports = [' spa_eth2 ', '', ' ']
|
|
normalized = self.adapter.normalize_config(config)
|
|
self.assertEqual(['pool_1'], normalized.unity_storage_pool_names)
|
|
self.assertEqual(['spa_eth2'], normalized.unity_io_ports)
|
|
|
|
def test_normalize_config_raise(self):
|
|
with self.assertRaisesRegexp(exception.InvalidConfigurationValue,
|
|
'unity_storage_pool_names'):
|
|
config = MockConfig()
|
|
config.unity_storage_pool_names = ['', ' ']
|
|
self.adapter.normalize_config(config)
|
|
with self.assertRaisesRegexp(exception.InvalidConfigurationValue,
|
|
'unity_io_ports'):
|
|
config = MockConfig()
|
|
config.unity_io_ports = ['', ' ']
|
|
self.adapter.normalize_config(config)
|
|
|
|
|
|
class FCAdapterTest(test.TestCase):
|
|
def setUp(self):
|
|
super(FCAdapterTest, self).setUp()
|
|
self.adapter = mock_adapter(adapter.FCAdapter)
|
|
|
|
def test_setup(self):
|
|
self.assertIsNotNone(self.adapter.lookup_service)
|
|
|
|
def test_auto_zone_enabled(self):
|
|
self.assertTrue(self.adapter.auto_zone_enabled)
|
|
|
|
def test_fc_protocol(self):
|
|
stats = mock_adapter(adapter.FCAdapter).update_volume_stats()
|
|
self.assertEqual('FC', stats['storage_protocol'])
|
|
|
|
def test_get_connector_uids(self):
|
|
connector = {'host': 'fake_host',
|
|
'wwnns': ['1111111111111111',
|
|
'2222222222222222'],
|
|
'wwpns': ['3333333333333333',
|
|
'4444444444444444']
|
|
}
|
|
expected = ['11:11:11:11:11:11:11:11:33:33:33:33:33:33:33:33',
|
|
'22:22:22:22:22:22:22:22:44:44:44:44:44:44:44:44']
|
|
ret = self.adapter.get_connector_uids(connector)
|
|
self.assertListEqual(expected, ret)
|
|
|
|
def test_get_connection_info_no_targets(self):
|
|
def f():
|
|
host = test_client.MockResource('no_target')
|
|
self.adapter.get_connection_info(12, host, {})
|
|
|
|
self.assertRaises(exception.VolumeBackendAPIException, f)
|
|
|
|
def test_get_connection_info_auto_zone_enabled(self):
|
|
host = test_client.MockResource('host1')
|
|
connector = {'wwpns': 'abcdefg'}
|
|
ret = self.adapter.get_connection_info(10, host, connector)
|
|
target_wwns = ['100000051e55a100', '100000051e55a121']
|
|
self.assertListEqual(target_wwns, ret['target_wwn'])
|
|
init_target_map = {
|
|
'200000051e55a100': ('100000051e55a100', '100000051e55a121'),
|
|
'200000051e55a121': ('100000051e55a100', '100000051e55a121')}
|
|
self.assertDictEqual(init_target_map, ret['initiator_target_map'])
|
|
self.assertEqual(10, ret['target_lun'])
|
|
|
|
def test_get_connection_info_auto_zone_disabled(self):
|
|
self.adapter.lookup_service = None
|
|
host = test_client.MockResource('host1')
|
|
connector = {'wwpns': 'abcdefg'}
|
|
ret = self.adapter.get_connection_info(10, host, connector)
|
|
self.assertEqual(10, ret['target_lun'])
|
|
wwns = ['8899AABBCCDDEEFF', '8899AABBCCDDFFEE']
|
|
self.assertListEqual(wwns, ret['target_wwn'])
|
|
|
|
@patch_for_fc_adapter
|
|
def test_initialize_connection_volume(self):
|
|
volume = MockOSResource(provider_location='id^lun_43', id='id_43')
|
|
connector = {'host': 'host1'}
|
|
conn_info = self.adapter.initialize_connection(volume, connector)
|
|
self.assertEqual('fibre_channel', conn_info['driver_volume_type'])
|
|
self.assertTrue(conn_info['data']['target_discovered'])
|
|
self.assertEqual('id_43', conn_info['data']['volume_id'])
|
|
|
|
@patch_for_fc_adapter
|
|
def test_initialize_connection_snapshot(self):
|
|
snap = MockOSResource(id='snap_1', name='snap_1')
|
|
connector = {'host': 'host1'}
|
|
conn_info = self.adapter.initialize_connection_snapshot(
|
|
snap, connector)
|
|
self.assertEqual('fibre_channel', conn_info['driver_volume_type'])
|
|
self.assertTrue(conn_info['data']['target_discovered'])
|
|
self.assertEqual('snap_1', conn_info['data']['volume_id'])
|
|
|
|
def test_terminate_connection_auto_zone_enabled(self):
|
|
connector = {'host': 'host1', 'wwpns': 'abcdefg'}
|
|
volume = MockOSResource(provider_location='id^lun_41', id='id_41')
|
|
ret = self.adapter.terminate_connection(volume, connector)
|
|
self.assertEqual('fibre_channel', ret['driver_volume_type'])
|
|
data = ret['data']
|
|
target_map = {
|
|
'200000051e55a100': ('100000051e55a100', '100000051e55a121'),
|
|
'200000051e55a121': ('100000051e55a100', '100000051e55a121')}
|
|
self.assertDictEqual(target_map, data['initiator_target_map'])
|
|
target_wwn = ['100000051e55a100', '100000051e55a121']
|
|
self.assertListEqual(target_wwn, data['target_wwn'])
|
|
|
|
def test_validate_ports_whitelist_none(self):
|
|
ports = self.adapter.validate_ports(None)
|
|
self.assertEqual(set(('spa_iom_0_fc0', 'spa_iom_0_fc1')), set(ports))
|
|
|
|
def test_validate_ports(self):
|
|
ports = self.adapter.validate_ports(['spa_iom_0_fc0'])
|
|
self.assertEqual(set(('spa_iom_0_fc0',)), set(ports))
|
|
|
|
def test_validate_ports_asterisk(self):
|
|
ports = self.adapter.validate_ports(['spa*'])
|
|
self.assertEqual(set(('spa_iom_0_fc0', 'spa_iom_0_fc1')), set(ports))
|
|
|
|
def test_validate_ports_question_mark(self):
|
|
ports = self.adapter.validate_ports(['spa_iom_0_fc?'])
|
|
self.assertEqual(set(('spa_iom_0_fc0', 'spa_iom_0_fc1')), set(ports))
|
|
|
|
def test_validate_ports_no_matched(self):
|
|
with self.assertRaisesRegexp(exception.InvalidConfigurationValue,
|
|
'unity_io_ports'):
|
|
self.adapter.validate_ports(['spc_invalid'])
|
|
|
|
def test_validate_ports_unmatched_whitelist(self):
|
|
with self.assertRaisesRegexp(exception.InvalidConfigurationValue,
|
|
'unity_io_ports'):
|
|
self.adapter.validate_ports(['spa_iom*', 'spc_invalid'])
|
|
|
|
|
|
class ISCSIAdapterTest(test.TestCase):
|
|
def setUp(self):
|
|
super(ISCSIAdapterTest, self).setUp()
|
|
self.adapter = mock_adapter(adapter.ISCSIAdapter)
|
|
|
|
def test_iscsi_protocol(self):
|
|
stats = self.adapter.update_volume_stats()
|
|
self.assertEqual('iSCSI', stats['storage_protocol'])
|
|
|
|
def test_get_connector_uids(self):
|
|
connector = {'host': 'fake_host', 'initiator': 'fake_iqn'}
|
|
ret = self.adapter.get_connector_uids(connector)
|
|
self.assertListEqual(['fake_iqn'], ret)
|
|
|
|
def test_get_connection_info(self):
|
|
connector = {'host': 'fake_host', 'initiator': 'fake_iqn'}
|
|
hlu = 10
|
|
info = self.adapter.get_connection_info(hlu, None, connector)
|
|
target_iqns = ['iqn.1-1.com.e:c.a.a0', 'iqn.1-1.com.e:c.a.a1']
|
|
target_portals = ['1.2.3.4:1234', '1.2.3.5:1234']
|
|
self.assertListEqual(target_iqns, info['target_iqns'])
|
|
self.assertListEqual([hlu, hlu], info['target_luns'])
|
|
self.assertListEqual(target_portals, info['target_portals'])
|
|
self.assertEqual(hlu, info['target_lun'])
|
|
self.assertTrue(info['target_portal'] in target_portals)
|
|
self.assertTrue(info['target_iqn'] in target_iqns)
|
|
|
|
@patch_for_iscsi_adapter
|
|
def test_initialize_connection_volume(self):
|
|
volume = MockOSResource(provider_location='id^lun_43', id='id_43')
|
|
connector = {'host': 'host1'}
|
|
conn_info = self.adapter.initialize_connection(volume, connector)
|
|
self.assertEqual('iscsi', conn_info['driver_volume_type'])
|
|
self.assertTrue(conn_info['data']['target_discovered'])
|
|
self.assertEqual('id_43', conn_info['data']['volume_id'])
|
|
|
|
@patch_for_iscsi_adapter
|
|
def test_initialize_connection_snapshot(self):
|
|
snap = MockOSResource(id='snap_1', name='snap_1')
|
|
connector = {'host': 'host1'}
|
|
conn_info = self.adapter.initialize_connection_snapshot(
|
|
snap, connector)
|
|
self.assertEqual('iscsi', conn_info['driver_volume_type'])
|
|
self.assertTrue(conn_info['data']['target_discovered'])
|
|
self.assertEqual('snap_1', conn_info['data']['volume_id'])
|