from cinder.volume.drivers import lvm
from cinder.volume import manager as vol_manager
from cinder.volume import rpcapi as volume_rpcapi
-from cinder.volume.targets import tgt
from cinder.volume import utils as volutils
from cinder.volume import volume_types
cfg.StrOpt('fake_opt1', default='fake', help='fake opts')
]
-FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa'
-
class FakeImageService(object):
def __init__(self, db_driver=None, image_service=None):
class BaseVolumeTestCase(test.TestCase):
"""Test Case for volumes."""
+
+ FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa'
+
def setUp(self):
super(BaseVolumeTestCase, self).setUp()
self.extension_manager = extension.ExtensionManager(
self.flags(volumes_dir=vol_tmpdir,
notification_driver=["test"])
self.addCleanup(self._cleanup)
- with mock.patch("osprofiler.profiler.trace_cls") as mock_trace_cls:
- side_effect = lambda value: value
- mock_decorator = mock.MagicMock(side_effect=side_effect)
- mock_trace_cls.return_value = mock_decorator
- self.volume = importutils.import_object(CONF.volume_manager)
+ self.volume = importutils.import_object(CONF.volume_manager)
self.configuration = mock.Mock(conf.Configuration)
self.context = context.get_admin_context()
self.context.user_id = 'fake'
class VolumeTestCase(BaseVolumeTestCase):
- def _fake_create_iscsi_target(self, name, tid,
- lun, path, chap_auth=None,
- **kwargs):
- return 1
-
def setUp(self):
super(VolumeTestCase, self).setUp()
- self.stubs.Set(volutils, 'clear_volume',
- lambda a, b, volume_clear=mox.IgnoreArg(),
- volume_clear_size=mox.IgnoreArg(),
- lvm_type=mox.IgnoreArg(),
- throttle=mox.IgnoreArg(): None)
- self.stubs.Set(tgt.TgtAdm,
- 'create_iscsi_target',
- self._fake_create_iscsi_target)
+ self._clear_patch = mock.patch('cinder.volume.utils.clear_volume',
+ autospec=True)
+ self._clear_patch.start()
+
+ def tearDown(self):
+ super(VolumeTestCase, self).tearDown()
+ self._clear_patch.stop()
def test_init_host_clears_downloads(self):
"""Test that init_host will unwedge a volume stuck in downloading."""
@mock.patch.object(QUOTAS, 'commit')
@mock.patch.object(QUOTAS, 'reserve')
def test_delete_driver_not_initialized(self, reserve, commit, rollback):
- # NOTE(flaper87): Set initialized to False
self.volume.driver._initialized = False
def fake_reserve(context, expire=None, project_id=None, **deltas):
self.assertEqual("error_deleting", volume.status)
db.volume_destroy(context.get_admin_context(), volume_id)
- def test_create_delete_volume(self):
+ @mock.patch('cinder.quota.QUOTAS.rollback', new=mock.Mock())
+ @mock.patch('cinder.quota.QUOTAS.commit', new=mock.Mock())
+ @mock.patch('cinder.quota.QUOTAS.reserve', return_value=['RESERVATION'])
+ def test_create_delete_volume(self, _mock_reserve):
"""Test volume can be created and deleted."""
- # Need to stub out reserve, commit, and rollback
- def fake_reserve(context, expire=None, project_id=None, **deltas):
- return ["RESERVATION"]
-
- def fake_commit(context, reservations, project_id=None):
- pass
-
- def fake_rollback(context, reservations, project_id=None):
- pass
-
- self.stubs.Set(QUOTAS, "reserve", fake_reserve)
- self.stubs.Set(QUOTAS, "commit", fake_commit)
- self.stubs.Set(QUOTAS, "rollback", fake_rollback)
-
volume = tests_utils.create_volume(
self.context,
availability_zone=CONF.storage_availability_zone,
'fake_key1',
FAKE_METADATA_TYPE.fake_type)
- def test_create_volume_uses_default_availability_zone(self):
+ @mock.patch.object(cinder.volume.api.API, 'list_availability_zones')
+ def test_create_volume_uses_default_availability_zone(self, mock_list_az):
"""Test setting availability_zone correctly during volume create."""
- volume_api = cinder.volume.api.API()
-
- def fake_list_availability_zones(enable_cache=False):
- return ({'name': 'az1', 'available': True},
- {'name': 'az2', 'available': True},
- {'name': 'default-az', 'available': True})
+ mock_list_az.return_value = ({'name': 'az1', 'available': True},
+ {'name': 'az2', 'available': True},
+ {'name': 'default-az', 'available': True})
- self.stubs.Set(volume_api,
- 'list_availability_zones',
- fake_list_availability_zones)
+ volume_api = cinder.volume.api.API()
# Test backwards compatibility, default_availability_zone not set
self.override_config('storage_availability_zone', 'az2')
'description')
self.assertEqual('default-az', volume['availability_zone'])
- def test_create_volume_with_volume_type(self):
+ @mock.patch('cinder.quota.QUOTAS.rollback', new=mock.MagicMock())
+ @mock.patch('cinder.quota.QUOTAS.commit', new=mock.MagicMock())
+ @mock.patch('cinder.quota.QUOTAS.reserve', return_value=["RESERVATION"])
+ def test_create_volume_with_volume_type(self, _mock_reserve):
"""Test volume creation with default volume type."""
- def fake_reserve(context, expire=None, project_id=None, **deltas):
- return ["RESERVATION"]
-
- def fake_commit(context, reservations, project_id=None):
- pass
-
- def fake_rollback(context, reservations, project_id=None):
- pass
-
- self.stubs.Set(QUOTAS, "reserve", fake_reserve)
- self.stubs.Set(QUOTAS, "commit", fake_commit)
- self.stubs.Set(QUOTAS, "rollback", fake_rollback)
-
volume_api = cinder.volume.api.API()
# Create volume with default volume type while default
volume_type=db_vol_type)
self.assertEqual(db_vol_type.get('id'), volume['volume_type_id'])
+ @mock.patch.object(keymgr, 'API', fake_keymgr.fake_api)
def test_create_volume_with_encrypted_volume_type(self):
- self.stubs.Set(keymgr, "API", fake_keymgr.fake_api)
-
ctxt = context.get_admin_context()
db.volume_type_create(ctxt,
self.volume.create_volume(self.context, volume['id'])
self.assertEqual('1111-aaaa', volume['provider_id'])
+ @mock.patch.object(keymgr, 'API', new=fake_keymgr.fake_api)
def test_create_delete_volume_with_encrypted_volume_type(self):
- self.stubs.Set(keymgr, "API", fake_keymgr.fake_api)
-
ctxt = context.get_admin_context()
db.volume_type_create(ctxt,
size=volume_src['size'])['id']
snapshot_obj = objects.Snapshot.get_by_id(self.context, snapshot_id)
- # NOTE(flaper87): Set initialized to False
self.volume.driver._initialized = False
self.assertRaises(exception.DriverNotInitialized,
snapshot = db.snapshot_get(context.get_admin_context(), snapshot_id)
self.assertEqual("error", snapshot.status)
- # NOTE(flaper87): Set initialized to True,
# lets cleanup the mess
self.volume.driver._initialized = True
self.volume.delete_snapshot(self.context, snapshot_obj)
def _fake_execute(self, *cmd, **kwargs):
pass
- def test_create_volume_from_snapshot_check_locks(self):
+ @mock.patch.object(cinder.volume.drivers.lvm.LVMISCSIDriver,
+ 'create_volume_from_snapshot')
+ def test_create_volume_from_snapshot_check_locks(
+ self, mock_lvm_create):
# mock the synchroniser so we can record events
self.stubs.Set(utils, 'synchronized', self._mock_synchronized)
-
- self.stubs.Set(self.volume.driver, 'create_volume_from_snapshot',
- lambda *args, **kwargs: None)
-
orig_flow = engine.ActionEngine.run
def mock_flow_run(*args, **kwargs):
'unlock-%s' % ('%s-delete_volume' % (src_vol_id))],
self.called)
+ self.assertTrue(mock_lvm_create.called)
+
def test_create_volume_from_volume_check_locks(self):
# mock the synchroniser so we can record events
self.stubs.Set(utils, 'synchronized', self._mock_synchronized)
self.assertRaises(exception.VolumeNotFound, db.volume_get,
self.context, src_vol_id)
+ @mock.patch.object(keymgr, 'API', fake_keymgr.fake_api)
def test_create_volume_from_snapshot_with_encryption(self):
"""Test volume can be created from a snapshot of
an encrypted volume.
"""
- self.stubs.Set(keymgr, 'API', fake_keymgr.fake_api)
-
ctxt = context.get_admin_context()
db.volume_type_create(ctxt,
@mock.patch.object(db, 'volume_get')
def test_reserve_volume_success(self, volume_get, volume_update):
fake_volume = {
- 'id': FAKE_UUID,
+ 'id': self.FAKE_UUID,
'status': 'available'
}
def test_reserve_volume_bad_status(self):
fake_volume = {
- 'id': FAKE_UUID,
+ 'id': self.FAKE_UUID,
'status': 'attaching'
}
volume_attachment_get_used_by_volume_id,
volume_update):
fake_volume = {
- 'id': FAKE_UUID,
+ 'id': self.FAKE_UUID,
'status': 'attaching'
}
- fake_attachments = [{'volume_id': FAKE_UUID,
+ fake_attachments = [{'volume_id': self.FAKE_UUID,
'instance_uuid': 'fake_instance_uuid'}]
volume_get.return_value = fake_volume
self.context,
volume_id, None, None, None,
None,
- FAKE_UUID)
+ self.FAKE_UUID)
volume = db.volume_get(self.context, volume_id)
self.assertEqual("error", volume['status'])
self.assertFalse(volume['bootable'])
self.context, 2,
'name', 'description', image_id=1)
- def _do_test_create_volume_with_size(self, size):
- def fake_reserve(context, expire=None, project_id=None, **deltas):
- return ["RESERVATION"]
-
- def fake_commit(context, reservations, project_id=None):
- pass
-
- def fake_rollback(context, reservations, project_id=None):
- pass
-
- self.stubs.Set(QUOTAS, "reserve", fake_reserve)
- self.stubs.Set(QUOTAS, "commit", fake_commit)
- self.stubs.Set(QUOTAS, "rollback", fake_rollback)
-
+ @mock.patch.object(QUOTAS, "rollback")
+ @mock.patch.object(QUOTAS, "commit")
+ @mock.patch.object(QUOTAS, "reserve", return_value=["RESERVATION"])
+ def _do_test_create_volume_with_size(self, size, *_unused_quota_mocks):
volume_api = cinder.volume.api.API()
volume = volume_api.create(self.context,
"""Test volume creation with string size."""
self._do_test_create_volume_with_size('2')
- def test_create_volume_with_bad_size(self):
- def fake_reserve(context, expire=None, project_id=None, **deltas):
- return ["RESERVATION"]
-
- def fake_commit(context, reservations, project_id=None):
- pass
-
- def fake_rollback(context, reservations, project_id=None):
- pass
-
- self.stubs.Set(QUOTAS, "reserve", fake_reserve)
- self.stubs.Set(QUOTAS, "commit", fake_commit)
- self.stubs.Set(QUOTAS, "rollback", fake_rollback)
-
+ @mock.patch.object(QUOTAS, "rollback")
+ @mock.patch.object(QUOTAS, "commit")
+ @mock.patch.object(QUOTAS, "reserve", return_value=["RESERVATION"])
+ def test_create_volume_with_bad_size(self, *_unused_quota_mocks):
volume_api = cinder.volume.api.API()
self.assertRaises(exception.InvalidInput,
host=CONF.host)
self.volume.create_volume(self.context, volume['id'])
- # NOTE(flaper87): Set initialized to False
self.volume.driver._initialized = False
self.assertRaises(exception.DriverNotInitialized,
volume = db.volume_get(context.get_admin_context(), volume['id'])
self.assertEqual('error_extending', volume.status)
- # NOTE(flaper87): Set initialized to True,
# lets cleanup the mess.
self.volume.driver._initialized = True
self.volume.delete_volume(self.context, volume['id'])
self.volume.delete_volume(self.context, volume_dst['id'])
self.volume.delete_volume(self.context, volume_src['id'])
- def test_create_volume_from_sourcevol_fail_wrong_az(self):
+ @mock.patch('cinder.volume.api.API.list_availability_zones',
+ return_value=({'name': 'nova', 'available': True},
+ {'name': 'az2', 'available': True}))
+ def test_create_volume_from_sourcevol_fail_wrong_az(self, _mock_laz):
"""Test volume can't be cloned from an other volume in different az."""
volume_api = cinder.volume.api.API()
- def fake_list_availability_zones(enable_cache=False):
- return ({'name': 'nova', 'available': True},
- {'name': 'az2', 'available': True})
-
- self.stubs.Set(volume_api,
- 'list_availability_zones',
- fake_list_availability_zones)
-
volume_src = tests_utils.create_volume(self.context,
availability_zone='az2',
**self.volume_params)
volume = db.volume_get(context.get_admin_context(), volume['id'])
self.assertEqual('error', volume.migration_status)
- # NOTE(flaper87): Set initialized to True,
# lets cleanup the mess.
self.volume.driver._initialized = True
self.volume.delete_volume(self.context, volume['id'])
return cgsnapshot, snapshot
- def test_create_delete_cgsnapshot(self):
+ @mock.patch('cinder.volume.driver.VolumeDriver.create_consistencygroup',
+ autospec=True,
+ return_value={'status': 'available'})
+ @mock.patch('cinder.volume.driver.VolumeDriver.delete_consistencygroup',
+ autospec=True,
+ return_value=({'status': 'deleted'}, []))
+ @mock.patch('cinder.volume.driver.VolumeDriver.create_cgsnapshot',
+ autospec=True,
+ return_value=({'status': 'available'}, []))
+ @mock.patch('cinder.volume.driver.VolumeDriver.delete_cgsnapshot',
+ autospec=True,
+ return_value=({'status': 'deleted'}, []))
+ def test_create_delete_cgsnapshot(self,
+ mock_del_cgsnap, mock_create_cgsnap,
+ mock_del_cg, _mock_create_cg):
"""Test cgsnapshot can be created and deleted."""
- rval = {'status': 'available'}
- driver.VolumeDriver.create_consistencygroup = \
- mock.Mock(return_value=rval)
-
- rval = {'status': 'deleted'}, []
- driver.VolumeDriver.delete_consistencygroup = \
- mock.Mock(return_value=rval)
-
- rval = {'status': 'available'}, []
- driver.VolumeDriver.create_cgsnapshot = \
- mock.Mock(return_value=rval)
-
- rval = {'status': 'deleted'}, []
- driver.VolumeDriver.delete_cgsnapshot = \
- mock.Mock(return_value=rval)
-
group = tests_utils.create_consistencygroup(
self.context,
availability_zone=CONF.storage_availability_zone,
self.volume.delete_consistencygroup(self.context, group_id)
- def test_delete_consistencygroup_correct_host(self):
+ self.assertTrue(mock_create_cgsnap.called)
+ self.assertTrue(mock_del_cgsnap.called)
+ self.assertTrue(mock_del_cg.called)
+
+ @mock.patch('cinder.volume.driver.VolumeDriver.create_consistencygroup',
+ return_value={'status': 'available'})
+ @mock.patch('cinder.volume.driver.VolumeDriver.delete_consistencygroup',
+ return_value=({'status': 'deleted'}, []))
+ def test_delete_consistencygroup_correct_host(self,
+ mock_del_cg,
+ _mock_create_cg):
"""Test consistencygroup can be deleted.
Test consistencygroup can be deleted when volumes are on
the correct volume node.
"""
- rval = {'status': 'available'}
- driver.VolumeDriver.create_consistencygroup = \
- mock.Mock(return_value=rval)
-
- rval = {'status': 'deleted'}, []
- driver.VolumeDriver.delete_consistencygroup = \
- mock.Mock(return_value=rval)
-
group = tests_utils.create_consistencygroup(
self.context,
availability_zone=CONF.storage_availability_zone,
self.context,
group_id)
- def test_delete_consistencygroup_wrong_host(self):
+ self.assertTrue(mock_del_cg.called)
+
+ @mock.patch('cinder.volume.driver.VolumeDriver.create_consistencygroup',
+ return_value={'status': 'available'})
+ def test_delete_consistencygroup_wrong_host(self, *_mock_create_cg):
"""Test consistencygroup cannot be deleted.
Test consistencygroup cannot be deleted when volumes in the
group are not local to the volume node.
"""
- rval = {'status': 'available'}
- driver.VolumeDriver.create_consistencygroup = \
- mock.Mock(return_value=rval)
-
group = tests_utils.create_consistencygroup(
self.context,
availability_zone=CONF.storage_availability_zone,
ret_flag = self.volume.driver.secure_file_operations_enabled()
self.assertFalse(ret_flag)
- @mock.patch('cinder.volume.flows.common.make_pretty_name')
- @mock.patch('cinder.scheduler.rpcapi.SchedulerAPI.create_volume')
+ @mock.patch('cinder.volume.flows.common.make_pretty_name',
+ new=mock.MagicMock())
+ @mock.patch('cinder.scheduler.rpcapi.SchedulerAPI.create_volume',
+ return_value=None)
@mock.patch('cinder.volume.flows.manager.create_volume.'
- 'CreateVolumeFromSpecTask.execute')
+ 'CreateVolumeFromSpecTask.execute',
+ side_effect=exception.DriverNotInitialized())
def test_create_volume_raise_rescheduled_exception(self, mock_execute,
- mock_reschedule,
- mock_make_name):
+ mock_reschedule):
# Create source volume
- mock_execute.side_effect = exception.DriverNotInitialized()
- mock_reschedule.return_value = None
test_vol = tests_utils.create_volume(self.context,
**self.volume_params)
test_vol_id = test_vol['id']
self.assertEqual('available', volume['status'])
def test_copy_volume_to_image_exception(self):
- self.image_meta['id'] = FAKE_UUID
+ self.image_meta['id'] = self.FAKE_UUID
# creating volume testdata
self.volume_attrs['status'] = 'in-use'
db.volume_create(self.context, self.volume_attrs)
cinder.policy.init()
self.context = context.get_admin_context()
- self.stubs.Set(brick_lvm.LVM, '_vg_exists', lambda x: True)
def test_check_policy(self):
self.mox.StubOutWithMock(cinder.policy, 'enforce')
Code Review / openstack-build / cinder-build.git / commitdiff