¡@
OpenStack Study: test_nexenta.py**** CubicPower OpenStack Study ****
# Copyright 2011 Nexenta Systems, Inc. # 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. """ Unit tests for OpenStack Cinder volume driver """ import base64 import urllib2 import mox as mox_lib from cinder import context from cinder import db from cinder import test from cinder import units from cinder.volume import configuration as conf from cinder.volume.drivers import nexenta from cinder.volume.drivers.nexenta import iscsi from cinder.volume.drivers.nexenta import jsonrpc from cinder.volume.drivers.nexenta import nfs from cinder.volume.drivers.nexenta import utils
**** CubicPower OpenStack Study **** class TestNexentaISCSIDriver(test.TestCase): TEST_VOLUME_NAME = 'volume1' TEST_VOLUME_NAME2 = 'volume2' TEST_SNAPSHOT_NAME = 'snapshot1' TEST_VOLUME_REF = { 'name': TEST_VOLUME_NAME, 'size': 1, 'id': '1', 'status': 'available' } TEST_VOLUME_REF2 = { 'name': TEST_VOLUME_NAME2, 'size': 1, 'id': '2', 'status': 'in-use' } TEST_SNAPSHOT_REF = { 'name': TEST_SNAPSHOT_NAME, 'volume_name': TEST_VOLUME_NAME, }
**** CubicPower OpenStack Study **** def __init__(self, method): super(TestNexentaISCSIDriver, self).__init__(method)
**** CubicPower OpenStack Study **** def setUp(self): super(TestNexentaISCSIDriver, self).setUp() self.configuration = mox_lib.MockObject(conf.Configuration) self.configuration.nexenta_host = '1.1.1.1' self.configuration.nexenta_user = 'admin' self.configuration.nexenta_password = 'nexenta' self.configuration.nexenta_volume = 'cinder' self.configuration.nexenta_rest_port = 2000 self.configuration.nexenta_rest_protocol = 'http' self.configuration.nexenta_iscsi_target_portal_port = 3260 self.configuration.nexenta_target_prefix = 'iqn:' self.configuration.nexenta_target_group_prefix = 'cinder/' self.configuration.nexenta_blocksize = '8K' self.configuration.nexenta_sparse = True self.configuration.nexenta_rrmgr_compression = 1 self.configuration.nexenta_rrmgr_tcp_buf_size = 1024 self.configuration.nexenta_rrmgr_connections = 2 self.nms_mock = self.mox.CreateMockAnything() for mod in ['volume', 'zvol', 'iscsitarget', 'appliance', 'stmf', 'scsidisk', 'snapshot']: setattr(self.nms_mock, mod, self.mox.CreateMockAnything()) self.stubs.Set(jsonrpc, 'NexentaJSONProxy', lambda *_, **__: self.nms_mock) self.drv = iscsi.NexentaISCSIDriver(configuration=self.configuration) self.drv.do_setup({})
**** CubicPower OpenStack Study **** def test_setup_error(self): self.nms_mock.volume.object_exists('cinder').AndReturn(True) self.mox.ReplayAll() self.drv.check_for_setup_error()
**** CubicPower OpenStack Study **** def test_setup_error_fail(self): self.nms_mock.volume.object_exists('cinder').AndReturn(False) self.mox.ReplayAll() self.assertRaises(LookupError, self.drv.check_for_setup_error)
**** CubicPower OpenStack Study **** def test_local_path(self): self.assertRaises(NotImplementedError, self.drv.local_path, '')
**** CubicPower OpenStack Study **** def test_create_volume(self): self.nms_mock.zvol.create('cinder/volume1', '1G', '8K', True) self.nms_mock.stmf.list_targets() self.nms_mock.iscsitarget.create_target({'target_name': 'iqn:volume1'}) self.nms_mock.stmf.list_targetgroups() self.nms_mock.stmf.create_targetgroup('cinder/volume1') self.nms_mock.stmf.list_targetgroup_members('cinder/volume1') self.nms_mock.stmf.add_targetgroup_member('cinder/volume1', 'iqn:volume1') self.nms_mock.scsidisk.lu_exists('cinder/volume1') self.nms_mock.scsidisk.create_lu('cinder/volume1', {}) self.nms_mock.scsidisk.lu_shared('cinder/volume1') self.nms_mock.scsidisk.add_lun_mapping_entry( 'cinder/volume1', {'target_group': 'cinder/volume1', 'lun': '0'}) self.mox.ReplayAll() self.drv.create_volume(self.TEST_VOLUME_REF)
**** CubicPower OpenStack Study **** def test_delete_volume(self): self.nms_mock.zvol.get_child_props('cinder/volume1', 'origin').AndReturn({}) self.nms_mock.zvol.destroy('cinder/volume1', '') self.mox.ReplayAll() self.drv.delete_volume(self.TEST_VOLUME_REF) self.mox.ResetAll() c = self.nms_mock.zvol.get_child_props('cinder/volume1', 'origin') c.AndReturn({'origin': 'cinder/volume0@snapshot'}) self.nms_mock.zvol.destroy('cinder/volume1', '') self.mox.ReplayAll() self.drv.delete_volume(self.TEST_VOLUME_REF) self.mox.ResetAll() c = self.nms_mock.zvol.get_child_props('cinder/volume1', 'origin') c.AndReturn({'origin': 'cinder/volume0@cinder-clone-snapshot-1'}) self.nms_mock.zvol.destroy('cinder/volume1', '') self.nms_mock.snapshot.destroy( 'cinder/volume0@cinder-clone-snapshot-1', '') self.mox.ReplayAll() self.drv.delete_volume(self.TEST_VOLUME_REF) self.mox.ResetAll()
**** CubicPower OpenStack Study **** def test_create_cloned_volume(self): vol = self.TEST_VOLUME_REF2 src_vref = self.TEST_VOLUME_REF snapshot = { 'volume_name': src_vref['name'], 'name': 'cinder-clone-snapshot-%s' % vol['id'], } self.nms_mock.zvol.create_snapshot('cinder/%s' % src_vref['name'], snapshot['name'], '') self.nms_mock.zvol.clone('cinder/%s@%s' % (src_vref['name'], snapshot['name']), 'cinder/%s' % vol['name']) self.mox.ReplayAll() self.drv.create_cloned_volume(vol, src_vref)
**** CubicPower OpenStack Study **** def test_migrate_volume(self): volume = self.TEST_VOLUME_REF host = { 'capabilities': { 'vendor_name': 'Nexenta', 'location_info': 'NexentaISCSIDriver:1.1.1.1:cinder', 'free_capacity_gb': 1, 'iscsi_target_portal_port': 3260, 'nms_url': 'http://admin:password@1.1.1.1:2000' } } snapshot = { 'volume_name': volume['name'], 'name': 'cinder-migrate-snapshot-%s' % volume['id'], } self.nms_mock.appliance.ssh_list_bindings().AndReturn([]) self.nms_mock.zvol.create_snapshot('cinder/%s' % volume['name'], snapshot['name'], '') src = '%(volume)s/%(zvol)s@%(snapshot)s' % { 'volume': 'cinder', 'zvol': volume['name'], 'snapshot': snapshot['name'] } dst = '1.1.1.1:cinder' cmd = ' '.join(['rrmgr -s zfs -c 1 -q -e -w 1024 -n 2', src, dst]) self.nms_mock.appliance.execute(cmd) snapshot_name = 'cinder/%(volume)s@%(snapshot)s' % { 'volume': volume['name'], 'snapshot': snapshot['name'] } self.nms_mock.snapshot.destroy(snapshot_name, '') volume_name = 'cinder/%s' % volume['name'] self.nms_mock.zvol.get_child_props(volume_name, 'origin').AndReturn(None) self.nms_mock.zvol.destroy(volume_name, '') self.nms_mock.snapshot.destroy('cinder/%(volume)s@%(snapshot)s' % { 'volume': volume['name'], 'snapshot': snapshot['name'] }, '') self.mox.ReplayAll() self.drv.migrate_volume(None, volume, host)
**** CubicPower OpenStack Study **** def test_create_snapshot(self): self.nms_mock.zvol.create_snapshot('cinder/volume1', 'snapshot1', '') self.mox.ReplayAll() self.drv.create_snapshot(self.TEST_SNAPSHOT_REF)
**** CubicPower OpenStack Study **** def test_create_volume_from_snapshot(self): self.nms_mock.zvol.clone('cinder/volume1@snapshot1', 'cinder/volume2') self.mox.ReplayAll() self.drv.create_volume_from_snapshot(self.TEST_VOLUME_REF2, self.TEST_SNAPSHOT_REF)
**** CubicPower OpenStack Study **** def test_delete_snapshot(self): self.nms_mock.snapshot.destroy('cinder/volume1@snapshot1', '') self.mox.ReplayAll() self.drv.delete_snapshot(self.TEST_SNAPSHOT_REF) self.mox.ResetAll() # Check that exception not raised if snapshot does not exist mock = self.nms_mock.snapshot.destroy('cinder/volume1@snapshot1', '') mock.AndRaise(nexenta.NexentaException( 'Snapshot cinder/volume1@snapshot1 does not exist')) self.mox.ReplayAll() self.drv.delete_snapshot(self.TEST_SNAPSHOT_REF) _CREATE_EXPORT_METHODS = [ ('stmf', 'list_targets', tuple(), [], False, ), ('iscsitarget', 'create_target', ({'target_name': 'iqn:volume1'},), u'Unable to create iscsi target\n' u' iSCSI target iqn.1986-03.com.sun:02:cinder-volume1 already' u' configured\n' u' itadm create-target failed with error 17\n', True, ), ('stmf', 'list_targetgroups', tuple(), [], False, ), ('stmf', 'create_targetgroup', ('cinder/volume1',), u'Unable to create targetgroup: stmfadm: cinder/volume1:' u' already exists\n', True, ), ('stmf', 'list_targetgroup_members', ('cinder/volume1', ), [], False, ), ('stmf', 'add_targetgroup_member', ('cinder/volume1', 'iqn:volume1'), u'Unable to add member to targetgroup: stmfadm:' u' iqn.1986-03.com.sun:02:cinder-volume1: already exists\n', True, ), ('scsidisk', 'lu_exists', ('cinder/volume1', ), 0, False, ), ('scsidisk', 'create_lu', ('cinder/volume1', {}), u"Unable to create lu with zvol 'cinder/volume1':\n" u" sbdadm: filename /dev/zvol/rdsk/cinder/volume1: in use\n", True, ), ('scsidisk', 'lu_shared', ('cinder/volume1', ), 0, False, ), ('scsidisk', 'add_lun_mapping_entry', ('cinder/volume1', { 'target_group': 'cinder/volume1', 'lun': '0'}), u"Unable to add view to zvol 'cinder/volume1' (LUNs in use: ):\n" u" stmfadm: view entry exists\n", True, ), ]
**** CubicPower OpenStack Study **** def _stub_export_method(self, module, method, args, error, raise_exception, fail=False): m = getattr(self.nms_mock, module) m = getattr(m, method) mock = m(*args) if raise_exception and fail: mock.AndRaise(nexenta.NexentaException(error)) else: mock.AndReturn(error)
**** CubicPower OpenStack Study **** def _stub_all_export_methods(self, fail=False): for params in self._CREATE_EXPORT_METHODS: self._stub_export_method(*params, fail=fail)
**** CubicPower OpenStack Study **** def test_create_export(self): self._stub_all_export_methods() self.mox.ReplayAll() retval = self.drv.create_export({}, self.TEST_VOLUME_REF) location = '%(host)s:%(port)s,1 %(prefix)s%(volume)s 0' % { 'host': self.configuration.nexenta_host, 'port': self.configuration.nexenta_iscsi_target_portal_port, 'prefix': self.configuration.nexenta_target_prefix, 'volume': self.TEST_VOLUME_NAME } self.assertEqual(retval, {'provider_location': location})
**** CubicPower OpenStack Study **** def __get_test(i): def _test_create_export_fail(self): for params in self._CREATE_EXPORT_METHODS[:i]: self._stub_export_method(*params) self._stub_export_method(*self._CREATE_EXPORT_METHODS[i], fail=True) self.mox.ReplayAll() self.assertRaises(nexenta.NexentaException, self.drv.create_export, {}, self.TEST_VOLUME_REF) return _test_create_export_fail for i in range(len(_CREATE_EXPORT_METHODS)): if i % 2: locals()['test_create_export_fail_%d' % i] = __get_test(i)
**** CubicPower OpenStack Study **** def _test_create_export_fail(self): for params in self._CREATE_EXPORT_METHODS[:i]: self._stub_export_method(*params) self._stub_export_method(*self._CREATE_EXPORT_METHODS[i], fail=True) self.mox.ReplayAll() self.assertRaises(nexenta.NexentaException, self.drv.create_export, {}, self.TEST_VOLUME_REF) return _test_create_export_fail for i in range(len(_CREATE_EXPORT_METHODS)): if i % 2: locals()['test_create_export_fail_%d' % i] = __get_test(i)
**** CubicPower OpenStack Study **** def test_ensure_export(self): self._stub_all_export_methods(fail=True) self.mox.ReplayAll() self.drv.ensure_export({}, self.TEST_VOLUME_REF)
**** CubicPower OpenStack Study **** def test_remove_export(self): self.nms_mock.scsidisk.delete_lu('cinder/volume1') self.nms_mock.stmf.destroy_targetgroup('cinder/volume1') self.nms_mock.iscsitarget.delete_target('iqn:volume1') self.mox.ReplayAll() self.drv.remove_export({}, self.TEST_VOLUME_REF)
**** CubicPower OpenStack Study **** def test_remove_export_fail_0(self): self.nms_mock.scsidisk.delete_lu('cinder/volume1') self.nms_mock.stmf.destroy_targetgroup( 'cinder/volume1').AndRaise(nexenta.NexentaException()) self.nms_mock.iscsitarget.delete_target('iqn:volume1') self.mox.ReplayAll() self.drv.remove_export({}, self.TEST_VOLUME_REF)
**** CubicPower OpenStack Study **** def test_remove_export_fail_1(self): self.nms_mock.scsidisk.delete_lu('cinder/volume1') self.nms_mock.stmf.destroy_targetgroup('cinder/volume1') self.nms_mock.iscsitarget.delete_target( 'iqn:volume1').AndRaise(nexenta.NexentaException()) self.mox.ReplayAll() self.drv.remove_export({}, self.TEST_VOLUME_REF)
**** CubicPower OpenStack Study **** def test_get_volume_stats(self): stats = {'size': '5368709120G', 'used': '5368709120G', 'available': '5368709120G', 'health': 'ONLINE'} self.nms_mock.volume.get_child_props( self.configuration.nexenta_volume, 'health|size|used|available').AndReturn(stats) self.mox.ReplayAll() stats = self.drv.get_volume_stats(True) self.assertEqual(stats['storage_protocol'], 'iSCSI') self.assertEqual(stats['total_capacity_gb'], 5368709120.0) self.assertEqual(stats['free_capacity_gb'], 5368709120.0) self.assertEqual(stats['reserved_percentage'], 0) self.assertEqual(stats['QoS_support'], False)
**** CubicPower OpenStack Study **** class TestNexentaJSONRPC(test.TestCase): HOST = 'example.com' URL = 'http://%s/' % HOST URL_S = 'https://%s/' % HOST USER = 'user' PASSWORD = 'password' HEADERS = { 'Authorization': 'Basic %s' % base64.b64encode('%s:%s' % (USER, PASSWORD)), 'Content-Type': 'application/json' } REQUEST = 'the request'
**** CubicPower OpenStack Study **** def setUp(self): super(TestNexentaJSONRPC, self).setUp() self.proxy = jsonrpc.NexentaJSONProxy( 'http', self.HOST, 2000, '/', self.USER, self.PASSWORD, auto=True) self.mox.StubOutWithMock(urllib2, 'Request', True) self.mox.StubOutWithMock(urllib2, 'urlopen') self.resp_mock = self.mox.CreateMockAnything() self.resp_info_mock = self.mox.CreateMockAnything() self.resp_mock.info().AndReturn(self.resp_info_mock) urllib2.urlopen(self.REQUEST).AndReturn(self.resp_mock)
**** CubicPower OpenStack Study **** def test_call(self): urllib2.Request( 'http://%s:2000/' % self.HOST, '{"object": null, "params": ["arg1", "arg2"], "method": null}', self.HEADERS).AndReturn(self.REQUEST) self.resp_info_mock.status = '' self.resp_mock.read().AndReturn( '{"error": null, "result": "the result"}') self.mox.ReplayAll() result = self.proxy('arg1', 'arg2') self.assertEqual("the result", result)
**** CubicPower OpenStack Study **** def test_call_deep(self): urllib2.Request( 'http://%s:2000/' % self.HOST, '{"object": "obj1.subobj", "params": ["arg1", "arg2"],' ' "method": "meth"}', self.HEADERS).AndReturn(self.REQUEST) self.resp_info_mock.status = '' self.resp_mock.read().AndReturn( '{"error": null, "result": "the result"}') self.mox.ReplayAll() result = self.proxy.obj1.subobj.meth('arg1', 'arg2') self.assertEqual("the result", result)
**** CubicPower OpenStack Study **** def test_call_auto(self): urllib2.Request( 'http://%s:2000/' % self.HOST, '{"object": null, "params": ["arg1", "arg2"], "method": null}', self.HEADERS).AndReturn(self.REQUEST) urllib2.Request( 'https://%s:2000/' % self.HOST, '{"object": null, "params": ["arg1", "arg2"], "method": null}', self.HEADERS).AndReturn(self.REQUEST) self.resp_info_mock.status = 'EOF in headers' self.resp_mock.read().AndReturn( '{"error": null, "result": "the result"}') urllib2.urlopen(self.REQUEST).AndReturn(self.resp_mock) self.mox.ReplayAll() result = self.proxy('arg1', 'arg2') self.assertEqual("the result", result)
**** CubicPower OpenStack Study **** def test_call_error(self): urllib2.Request( 'http://%s:2000/' % self.HOST, '{"object": null, "params": ["arg1", "arg2"], "method": null}', self.HEADERS).AndReturn(self.REQUEST) self.resp_info_mock.status = '' self.resp_mock.read().AndReturn( '{"error": {"message": "the error"}, "result": "the result"}') self.mox.ReplayAll() self.assertRaises(jsonrpc.NexentaJSONException, self.proxy, 'arg1', 'arg2')
**** CubicPower OpenStack Study **** def test_call_fail(self): urllib2.Request( 'http://%s:2000/' % self.HOST, '{"object": null, "params": ["arg1", "arg2"], "method": null}', self.HEADERS).AndReturn(self.REQUEST) self.resp_info_mock.status = 'EOF in headers' self.proxy.auto = False self.mox.ReplayAll() self.assertRaises(jsonrpc.NexentaJSONException, self.proxy, 'arg1', 'arg2')
**** CubicPower OpenStack Study **** class TestNexentaNfsDriver(test.TestCase): TEST_EXPORT1 = 'host1:/volumes/stack/share' TEST_NMS1 = 'http://admin:nexenta@host1:2000' TEST_EXPORT2 = 'host2:/volumes/stack/share' TEST_NMS2 = 'http://admin:nexenta@host2:2000' TEST_EXPORT2_OPTIONS = '-o intr' TEST_FILE_NAME = 'test.txt' TEST_SHARES_CONFIG_FILE = '/etc/cinder/nexenta-shares.conf' TEST_SHARE_SVC = 'svc:/network/nfs/server:
**** CubicPower OpenStack Study **** def _create_volume_db_entry(self): vol = { 'id': '1', 'size': 1, 'status': 'available', 'provider_location': self.TEST_EXPORT1 } return db.volume_create(self.ctxt, vol)['id']
**** CubicPower OpenStack Study **** def setUp(self): super(TestNexentaNfsDriver, self).setUp() self.ctxt = context.get_admin_context() self.configuration = mox_lib.MockObject(conf.Configuration) self.configuration.nexenta_shares_config = None self.configuration.nexenta_mount_point_base = '$state_path/mnt' self.configuration.nexenta_sparsed_volumes = True self.configuration.nexenta_volume_compression = 'on' self.configuration.nfs_mount_point_base = '/mnt/test' self.configuration.nfs_mount_options = None self.configuration.nexenta_nms_cache_volroot = False self.nms_mock = self.mox.CreateMockAnything() for mod in ('appliance', 'folder', 'server', 'volume', 'netstorsvc', 'snapshot'): setattr(self.nms_mock, mod, self.mox.CreateMockAnything()) self.nms_mock.__hash__ = lambda *_, **__: 1 self.stubs.Set(jsonrpc, 'NexentaJSONProxy', lambda *_, **__: self.nms_mock) self.drv = nfs.NexentaNfsDriver(configuration=self.configuration) self.drv.shares = {} self.drv.share2nms = {}
**** CubicPower OpenStack Study **** def test_check_for_setup_error(self): self.drv.share2nms = { 'host1:/volumes/stack/share': self.nms_mock } self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.volume.object_exists('stack').AndReturn(True) self.nms_mock.folder.object_exists('stack/share').AndReturn(True) share_opts = { 'read_write': '*', 'read_only': '', 'root': 'nobody', 'extra_options': 'anon=0', 'recursive': 'true', 'anonymous_rw': 'true', } self.nms_mock.netstorsvc.share_folder( 'svc:/network/nfs/server:default', 'stack/share', share_opts) self.mox.ReplayAll() self.drv.check_for_setup_error() self.mox.ResetAll() self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.volume.object_exists('stack').AndReturn(False) self.mox.ReplayAll() self.assertRaises(LookupError, self.drv.check_for_setup_error) self.mox.ResetAll() self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.volume.object_exists('stack').AndReturn(True) self.nms_mock.folder.object_exists('stack/share').AndReturn(False) self.mox.ReplayAll() self.assertRaises(LookupError, self.drv.check_for_setup_error)
**** CubicPower OpenStack Study **** def test_initialize_connection(self): self.drv.shares = { self.TEST_EXPORT1: None } volume = { 'provider_location': self.TEST_EXPORT1, 'name': 'volume' } result = self.drv.initialize_connection(volume, None) self.assertEqual(result['data']['export'], '%s/volume' % self.TEST_EXPORT1)
**** CubicPower OpenStack Study **** def test_do_create_volume(self): volume = { 'provider_location': self.TEST_EXPORT1, 'size': 1, 'name': 'volume-1' } self.drv.shares = {self.TEST_EXPORT1: None} self.drv.share2nms = {self.TEST_EXPORT1: self.nms_mock} compression = self.configuration.nexenta_volume_compression self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.folder.create_with_props( 'stack', 'share/volume-1', {'compression': compression}) self.nms_mock.netstorsvc.share_folder(self.TEST_SHARE_SVC, 'stack/share/volume-1', self.TEST_SHARE_OPTS) self.nms_mock.appliance.execute( 'dd if=/dev/zero of=/volumes/stack/share/volume-1/volume bs=1M ' 'count=0 seek=1024' ) self.nms_mock.appliance.execute('chmod ugo+rw ' '/volumes/stack/share/volume-1/volume') self.mox.ReplayAll() self.drv._do_create_volume(volume) self.mox.ResetAll() self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.folder.create_with_props( 'stack', 'share/volume-1', {'compression': compression}) self.nms_mock.netstorsvc.share_folder( self.TEST_SHARE_SVC, 'stack/share/volume-1', self.TEST_SHARE_OPTS).AndRaise(nexenta.NexentaException('-')) self.nms_mock.folder.destroy('stack/share/volume-1') self.mox.ReplayAll() self.assertRaises(nexenta.NexentaException, self.drv._do_create_volume, volume)
**** CubicPower OpenStack Study **** def test_create_sparsed_file(self): self.nms_mock.appliance.execute('dd if=/dev/zero of=/tmp/path bs=1M ' 'count=0 seek=1024') self.mox.ReplayAll() self.drv._create_sparsed_file(self.nms_mock, '/tmp/path', 1)
**** CubicPower OpenStack Study **** def test_create_regular_file(self): self.nms_mock.appliance.execute('dd if=/dev/zero of=/tmp/path bs=1M ' 'count=1024') self.mox.ReplayAll() self.drv._create_regular_file(self.nms_mock, '/tmp/path', 1)
**** CubicPower OpenStack Study **** def test_set_rw_permissions_for_all(self): path = '/tmp/path' self.nms_mock.appliance.execute('chmod ugo+rw %s' % path) self.mox.ReplayAll() self.drv._set_rw_permissions_for_all(self.nms_mock, path)
**** CubicPower OpenStack Study **** def test_local_path(self): volume = {'provider_location': self.TEST_EXPORT1, 'name': 'volume-1'} path = self.drv.local_path(volume) self.assertEqual( path, '$state_path/mnt/b3f660847a52b29ac330d8555e4ad669/volume-1/volume' )
**** CubicPower OpenStack Study **** def test_remote_path(self): volume = {'provider_location': self.TEST_EXPORT1, 'name': 'volume-1'} path = self.drv.remote_path(volume) self.assertEqual(path, '/volumes/stack/share/volume-1/volume')
**** CubicPower OpenStack Study **** def test_share_folder(self): path = 'stack/share/folder' self.nms_mock.netstorsvc.share_folder(self.TEST_SHARE_SVC, path, self.TEST_SHARE_OPTS) self.mox.ReplayAll() self.drv._share_folder(self.nms_mock, 'stack', 'share/folder')
**** CubicPower OpenStack Study **** def test_load_shares_config(self): self.drv.configuration.nfs_shares_config = self.TEST_SHARES_CONFIG_FILE self.mox.StubOutWithMock(self.drv, '_read_config_file') config_data = [ '%s %s' % (self.TEST_EXPORT1, self.TEST_NMS1), '# %s %s' % (self.TEST_EXPORT2, self.TEST_NMS2), '', '%s %s %s' % (self.TEST_EXPORT2, self.TEST_NMS2, self.TEST_EXPORT2_OPTIONS) ] self.drv._read_config_file(self.TEST_SHARES_CONFIG_FILE).\ AndReturn(config_data) self.mox.ReplayAll() self.drv._load_shares_config(self.drv.configuration.nfs_shares_config) self.assertIn(self.TEST_EXPORT1, self.drv.shares) self.assertIn(self.TEST_EXPORT2, self.drv.shares) self.assertEqual(len(self.drv.shares), 2) self.assertIn(self.TEST_EXPORT1, self.drv.share2nms) self.assertIn(self.TEST_EXPORT2, self.drv.share2nms) self.assertEqual(len(self.drv.share2nms.keys()), 2) self.assertEqual(self.drv.shares[self.TEST_EXPORT2], self.TEST_EXPORT2_OPTIONS) self.mox.VerifyAll()
**** CubicPower OpenStack Study **** def test_get_capacity_info(self): self.drv.share2nms = {self.TEST_EXPORT1: self.nms_mock} self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.folder.get_child_props('stack/share', '').AndReturn({ 'available': '1G', 'used': '2G' }) self.mox.ReplayAll() total, free, allocated = self.drv._get_capacity_info(self.TEST_EXPORT1) self.assertEqual(total, 3 * units.GiB) self.assertEqual(free, units.GiB) self.assertEqual(allocated, 2 * units.GiB)
**** CubicPower OpenStack Study **** def test_get_share_datasets(self): self.drv.share2nms = {self.TEST_EXPORT1: self.nms_mock} self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.mox.ReplayAll() volume_name, folder_name = \ self.drv._get_share_datasets(self.TEST_EXPORT1) self.assertEqual(volume_name, 'stack') self.assertEqual(folder_name, 'share')
**** CubicPower OpenStack Study **** def test_delete_snapshot(self): self.drv.share2nms = {self.TEST_EXPORT1: self.nms_mock} self._create_volume_db_entry() self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.snapshot.destroy('stack/share/volume-1@snapshot1', '') self.mox.ReplayAll() self.drv.delete_snapshot({'volume_id': '1', 'name': 'snapshot1'}) self.mox.ResetAll() # Check that exception not raised if snapshot does not exist on # NexentaStor appliance. self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') mock = self.nms_mock.snapshot.destroy('stack/share/volume-1@snapshot1', '') mock.AndRaise(nexenta.NexentaException("Snapshot does not exist")) self.mox.ReplayAll() self.drv.delete_snapshot({'volume_id': '1', 'name': 'snapshot1'}) self.mox.ResetAll()
**** CubicPower OpenStack Study **** def test_delete_volume(self): self.drv.share2nms = {self.TEST_EXPORT1: self.nms_mock} self._create_volume_db_entry() self.drv._ensure_share_mounted = lambda *_, **__: 0 self.drv._execute = lambda *_, **__: 0 self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.folder.get_child_props('stack/share/volume-1', 'origin').AndReturn(None) self.nms_mock.folder.destroy('stack/share/volume-1', '-r') self.mox.ReplayAll() self.drv.delete_volume({ 'id': '1', 'name': 'volume-1', 'provider_location': self.TEST_EXPORT1 }) self.mox.ResetAll() # Check that exception not raised if folder does not exist on # NexentaStor appliance. self.nms_mock.server.get_prop('volroot').AndReturn('/volumes') self.nms_mock.folder.get_child_props('stack/share/volume-1', 'origin').AndReturn(None) mock = self.nms_mock.folder.destroy('stack/share/volume-1', '-r') mock.AndRaise(nexenta.NexentaException("Folder does not exist")) self.mox.ReplayAll() self.drv.delete_volume({ 'id': '1', 'name': 'volume-1', 'provider_location': self.TEST_EXPORT1 }) self.mox.ResetAll()
**** CubicPower OpenStack Study **** class TestNexentaUtils(test.TestCase):
**** CubicPower OpenStack Study **** def test_str2size(self): values_to_test = ( # Test empty value (None, 0), ('', 0), ('0', 0), ('12', 12), # Test int and long values (10, 10), (long(10), 10), # Test bytes string ('1b', 1), ('1B', 1), ('1023b', 1023), ('0B', 0), # Test other units ('1M', units.MiB), ('1.0M', units.MiB), ) for value, result in values_to_test: self.assertEqual(utils.str2size(value), result) # Invalid format value self.assertRaises(ValueError, utils.str2size, 'A')
**** CubicPower OpenStack Study **** def test_str2gib_size(self): self.assertEqual(utils.str2gib_size('1024M'), 1) self.assertEqual(utils.str2gib_size('300M'), 300 * units.MiB // units.GiB) self.assertEqual(utils.str2gib_size('1.2T'), 1.2 * units.TiB // units.GiB) self.assertRaises(ValueError, utils.str2gib_size, 'A')
**** CubicPower OpenStack Study **** def test_parse_nms_url(self): urls = ( ('auto://192.168.1.1/', (True, 'http', 'admin', 'nexenta', '192.168.1.1', '2000', '/rest/nms/')), ('http://192.168.1.1/', (False, 'http', 'admin', 'nexenta', '192.168.1.1', '2000', '/rest/nms/')), ('http://192.168.1.1:8080', (False, 'http', 'admin', 'nexenta', '192.168.1.1', '8080', '/rest/nms/')), ('https://root:password@192.168.1.1:8080', (False, 'https', 'root', 'password', '192.168.1.1', '8080', '/rest/nms/')), ) for url, result in urls: self.assertEqual(utils.parse_nms_url(url), result)
|