¡@
OpenStack Study: test_ibm_xiv_ds8k.py**** CubicPower OpenStack Study ****
# Copyright (c) 2013 OpenStack Foundation # 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. # # Authors: # Erik Zaadi # Avishay Traeger import mox from oslo.config import cfg from cinder import exception from cinder import test from cinder.volume import configuration as conf from cinder.volume.drivers.ibm import xiv_ds8k FAKE = "fake" VOLUME = {'size': 16, 'name': FAKE, 'id': 1} CONNECTOR = {'initiator': "iqn.2012-07.org.fake:01:948f189c4695", } CONF = cfg.CONF
**** CubicPower OpenStack Study **** class XIVDS8KFakeProxyDriver(object): """Fake IBM XIV and DS8K Proxy Driver."""
**** CubicPower OpenStack Study **** def __init__(self, xiv_ds8k_info, logger, expt, driver=None): """Initialize Proxy.""" self.xiv_ds8k_info = xiv_ds8k_info self.logger = logger self.exception = expt self.xiv_ds8k_portal = \ self.xiv_ds8k_iqn = FAKE self.volumes = {} self.driver = driver
**** CubicPower OpenStack Study **** def setup(self, context): if self.xiv_ds8k_info['xiv_ds8k_user'] != self.driver\ .configuration.san_login: raise self.exception.NotAuthorized() if self.xiv_ds8k_info['xiv_ds8k_address'] != self.driver\ .configuration.san_ip: raise self.exception.HostNotFound(host='fake')
**** CubicPower OpenStack Study **** def create_volume(self, volume): if volume['size'] > 100: raise self.exception.VolumeBackendAPIException(data='blah') self.volumes[volume['name']] = volume
**** CubicPower OpenStack Study **** def volume_exists(self, volume): return self.volumes.get(volume['name'], None) is not None
**** CubicPower OpenStack Study **** def delete_volume(self, volume): if self.volumes.get(volume['name'], None) is not None: del self.volumes[volume['name']]
**** CubicPower OpenStack Study **** def initialize_connection(self, volume, connector): if not self.volume_exists(volume): raise self.exception.VolumeNotFound(volume_id=volume['id']) lun_id = volume['id'] self.volumes[volume['name']]['attached'] = connector return {'driver_volume_type': 'iscsi', 'data': {'target_discovered': True, 'target_discovered': True, 'target_portal': self.xiv_ds8k_portal, 'target_iqn': self.xiv_ds8k_iqn, 'target_lun': lun_id, 'volume_id': volume['id'], 'multipath': True, 'provider_location': "%s,1 %s %s" % ( self.xiv_ds8k_portal, self.xiv_ds8k_iqn, lun_id), }, }
**** CubicPower OpenStack Study **** def terminate_connection(self, volume, connector): if not self.volume_exists(volume): raise self.exception.VolumeNotFound(volume_id=volume['id']) if not self.is_volume_attached(volume, connector): raise self.exception.NotFound(_('Volume not found for ' 'instance %(instance_id)s.') % {'instance_id': 'fake'}) del self.volumes[volume['name']]['attached']
**** CubicPower OpenStack Study **** def is_volume_attached(self, volume, connector): if not self.volume_exists(volume): raise self.exception.VolumeNotFound(volume_id=volume['id']) return (self.volumes[volume['name']].get('attached', None) == connector)
**** CubicPower OpenStack Study **** class XIVDS8KVolumeDriverTest(test.TestCase): """Test IBM XIV and DS8K volume driver."""
**** CubicPower OpenStack Study **** def setUp(self): """Initialize IBM XIV and DS8K Driver.""" super(XIVDS8KVolumeDriverTest, self).setUp() configuration = mox.MockObject(conf.Configuration) configuration.san_is_local = False configuration.xiv_ds8k_proxy = \ 'cinder.tests.test_ibm_xiv_ds8k.XIVDS8KFakeProxyDriver' configuration.xiv_ds8k_connection_type = 'iscsi' configuration.xiv_chap = 'disabled' configuration.san_ip = FAKE configuration.san_login = FAKE configuration.san_clustername = FAKE configuration.san_password = FAKE configuration.append_config_values(mox.IgnoreArg()) self.driver = xiv_ds8k.XIVDS8KDriver( configuration=configuration)
**** CubicPower OpenStack Study **** def test_initialized_should_set_xiv_ds8k_info(self): """Test that the san flags are passed to the IBM proxy.""" self.assertEqual( self.driver.xiv_ds8k_proxy.xiv_ds8k_info['xiv_ds8k_user'], self.driver.configuration.san_login) self.assertEqual( self.driver.xiv_ds8k_proxy.xiv_ds8k_info['xiv_ds8k_pass'], self.driver.configuration.san_password) self.assertEqual( self.driver.xiv_ds8k_proxy.xiv_ds8k_info['xiv_ds8k_address'], self.driver.configuration.san_ip) self.assertEqual( self.driver.xiv_ds8k_proxy.xiv_ds8k_info['xiv_ds8k_vol_pool'], self.driver.configuration.san_clustername)
**** CubicPower OpenStack Study **** def test_setup_should_fail_if_credentials_are_invalid(self): """Test that the xiv_ds8k_proxy validates credentials.""" self.driver.xiv_ds8k_proxy.xiv_ds8k_info['xiv_ds8k_user'] = 'invalid' self.assertRaises(exception.NotAuthorized, self.driver.do_setup, None)
**** CubicPower OpenStack Study **** def test_setup_should_fail_if_connection_is_invalid(self): """Test that the xiv_ds8k_proxy validates connection.""" self.driver.xiv_ds8k_proxy.xiv_ds8k_info['xiv_ds8k_address'] = \ 'invalid' self.assertRaises(exception.HostNotFound, self.driver.do_setup, None)
**** CubicPower OpenStack Study **** def test_create_volume(self): """Test creating a volume.""" self.driver.do_setup(None) self.driver.create_volume(VOLUME) has_volume = self.driver.xiv_ds8k_proxy.volume_exists(VOLUME) self.assertTrue(has_volume) self.driver.delete_volume(VOLUME)
**** CubicPower OpenStack Study **** def test_volume_exists(self): """Test the volume exist method with a volume that doesn't exist.""" self.driver.do_setup(None) self.assertFalse( self.driver.xiv_ds8k_proxy.volume_exists({'name': FAKE}))
**** CubicPower OpenStack Study **** def test_delete_volume(self): """Verify that a volume is deleted.""" self.driver.do_setup(None) self.driver.create_volume(VOLUME) self.driver.delete_volume(VOLUME) has_volume = self.driver.xiv_ds8k_proxy.volume_exists(VOLUME) self.assertFalse(has_volume)
**** CubicPower OpenStack Study **** def test_delete_volume_should_fail_for_not_existing_volume(self): """Verify that deleting a non-existing volume is OK.""" self.driver.do_setup(None) self.driver.delete_volume(VOLUME)
**** CubicPower OpenStack Study **** def test_create_volume_should_fail_if_no_pool_space_left(self): """Vertify that the xiv_ds8k_proxy validates volume pool space.""" self.driver.do_setup(None) self.assertRaises(exception.VolumeBackendAPIException, self.driver.create_volume, {'name': FAKE, 'id': 1, 'size': 12000})
**** CubicPower OpenStack Study **** def test_initialize_connection(self): """Test that inititialize connection attaches volume to host.""" self.driver.do_setup(None) self.driver.create_volume(VOLUME) self.driver.initialize_connection(VOLUME, CONNECTOR) self.assertTrue( self.driver.xiv_ds8k_proxy.is_volume_attached(VOLUME, CONNECTOR)) self.driver.terminate_connection(VOLUME, CONNECTOR) self.driver.delete_volume(VOLUME)
**** CubicPower OpenStack Study **** def test_initialize_connection_should_fail_for_non_existing_volume(self): """Verify that initialize won't work for non-existing volume.""" self.driver.do_setup(None) self.assertRaises(exception.VolumeNotFound, self.driver.initialize_connection, VOLUME, CONNECTOR)
**** CubicPower OpenStack Study **** def test_terminate_connection(self): """Test terminating a connection.""" self.driver.do_setup(None) self.driver.create_volume(VOLUME) self.driver.initialize_connection(VOLUME, CONNECTOR) self.driver.terminate_connection(VOLUME, CONNECTOR) self.assertFalse(self.driver.xiv_ds8k_proxy.is_volume_attached( VOLUME, CONNECTOR)) self.driver.delete_volume(VOLUME)
**** CubicPower OpenStack Study **** def test_terminate_connection_should_fail_on_non_existing_volume(self): """Test that terminate won't work for non-existing volumes.""" self.driver.do_setup(None) self.assertRaises(exception.VolumeNotFound, self.driver.terminate_connection, VOLUME, CONNECTOR)
|