def _fake_execute(self, *cmd, **kwargs):
pass
- @mock.patch.object(cinder.volume.drivers.lvm.LVMISCSIDriver,
+ @mock.patch.object(cinder.volume.drivers.lvm.LVMVolumeDriver,
'create_volume_from_snapshot')
def test_create_volume_from_snapshot_check_locks(
self, mock_lvm_create):
db.volume_destroy(self.context, dest_vol['id'])
-class LVMISCSIVolumeDriverTestCase(DriverTestCase):
+class LVMVolumeDriverTestCase(DriverTestCase):
"""Test case for VolumeDriver"""
- driver_name = "cinder.volume.drivers.lvm.LVMISCSIDriver"
-
- def test_delete_busy_volume(self):
- """Test deleting a busy volume."""
- self.stubs.Set(self.volume.driver, '_volume_not_present',
- lambda x: False)
- self.stubs.Set(self.volume.driver, '_delete_volume',
- lambda x: False)
+ driver_name = "cinder.volume.drivers.lvm.LVMVolumeDriver"
+ FAKE_VOLUME = {'name': 'test1',
+ 'id': 'test1'}
- self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
+ @mock.patch.object(fake_driver.FakeISCSIDriver, 'create_export')
+ def test_delete_volume_invalid_parameter(self, _mock_create_export):
+ self.configuration.volume_clear = 'zero'
+ self.configuration.volume_clear_size = 0
+ lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
+ db=db)
+ self.mox.StubOutWithMock(os.path, 'exists')
- self.stubs.Set(self.volume.driver.vg, 'lv_has_snapshot',
- lambda x: True)
- self.assertRaises(exception.VolumeIsBusy,
- self.volume.driver.delete_volume,
- {'name': 'test1', 'size': 1024})
+ os.path.exists(mox.IgnoreArg()).AndReturn(True)
- self.stubs.Set(self.volume.driver.vg, 'lv_has_snapshot',
- lambda x: False)
- self.output = 'x'
- self.volume.driver.delete_volume(
- {'name': 'test1',
- 'size': 1024,
- 'id': '478e14bc-a6a9-11e4-89d3-123b93f75cba'})
+ self.mox.ReplayAll()
+ # Test volume without 'size' field and 'volume_size' field
+ self.assertRaises(exception.InvalidParameterValue,
+ lvm_driver._delete_volume,
+ self.FAKE_VOLUME)
- def test_lvm_migrate_volume_no_loc_info(self):
- host = {'capabilities': {}}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- moved, model_update = self.volume.driver.migrate_volume(self.context,
- vol, host)
- self.assertFalse(moved)
- self.assertIsNone(model_update)
+ @mock.patch.object(fake_driver.FakeISCSIDriver, 'create_export')
+ def test_delete_volume_bad_path(self, _mock_create_export):
+ self.configuration.volume_clear = 'zero'
+ self.configuration.volume_clear_size = 0
+ self.configuration.volume_type = 'default'
- def test_lvm_migrate_volume_bad_loc_info(self):
- capabilities = {'location_info': 'foo'}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- moved, model_update = self.volume.driver.migrate_volume(self.context,
- vol, host)
- self.assertFalse(moved)
- self.assertIsNone(model_update)
+ volume = dict(self.FAKE_VOLUME, size=1)
+ lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
+ db=db)
- def test_lvm_migrate_volume_diff_driver(self):
- capabilities = {'location_info': 'FooDriver:foo:bar:default:0'}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- moved, model_update = self.volume.driver.migrate_volume(self.context,
- vol, host)
- self.assertFalse(moved)
- self.assertIsNone(model_update)
+ self.mox.StubOutWithMock(os.path, 'exists')
+ os.path.exists(mox.IgnoreArg()).AndReturn(False)
+ self.mox.ReplayAll()
- def test_lvm_migrate_volume_diff_host(self):
- capabilities = {'location_info': 'LVMVolumeDriver:foo:bar:default:0'}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- moved, model_update = self.volume.driver.migrate_volume(self.context,
- vol, host)
- self.assertFalse(moved)
- self.assertIsNone(model_update)
+ self.assertRaises(exception.VolumeBackendAPIException,
+ lvm_driver._delete_volume, volume)
- def test_lvm_migrate_volume_in_use(self):
- hostname = socket.gethostname()
- capabilities = {'location_info': 'LVMVolumeDriver:%s:bar' % hostname}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'in-use'}
- moved, model_update = self.volume.driver.migrate_volume(self.context,
- vol, host)
- self.assertFalse(moved)
- self.assertIsNone(model_update)
+ @mock.patch.object(fake_driver.FakeISCSIDriver, 'create_export')
+ def test_delete_volume_thinlvm_snap(self, _mock_create_export):
+ self.configuration.volume_clear = 'zero'
+ self.configuration.volume_clear_size = 0
+ self.configuration.lvm_type = 'thin'
+ self.configuration.iscsi_helper = 'tgtadm'
+ lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
+ vg_obj=mox.MockAnything(),
+ db=db)
- @mock.patch.object(volutils, 'get_all_volume_groups',
- return_value=[{'name': 'cinder-volumes'}])
- def test_lvm_migrate_volume_same_volume_group(self, vgs):
- hostname = socket.gethostname()
- capabilities = {'location_info': 'LVMVolumeDriver:%s:'
- 'cinder-volumes:default:0' % hostname}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
+ # Ensures that copy_volume is not called for ThinLVM
+ self.mox.StubOutWithMock(volutils, 'copy_volume')
+ self.mox.StubOutWithMock(volutils, 'clear_volume')
+ self.mox.StubOutWithMock(lvm_driver, '_execute')
+ self.mox.ReplayAll()
- self.assertRaises(exception.VolumeBackendAPIException,
- self.volume.driver.migrate_volume, self.context,
- vol, host)
+ uuid = '00000000-0000-0000-0000-c3aa7ee01536'
- @mock.patch.object(lvm.LVMVolumeDriver, '_create_volume')
- @mock.patch.object(brick_lvm.LVM, 'get_all_physical_volumes')
- @mock.patch.object(brick_lvm.LVM, 'delete')
- @mock.patch.object(volutils, 'copy_volume',
- side_effect=processutils.ProcessExecutionError)
- @mock.patch.object(volutils, 'get_all_volume_groups',
- return_value=[{'name': 'cinder-volumes'}])
- def test_lvm_migrate_volume_volume_copy_error(self, vgs, copy_volume,
- mock_delete, mock_pvs,
- mock_create):
+ fake_snapshot = {'name': 'volume-' + uuid,
+ 'id': uuid,
+ 'size': 123}
- hostname = socket.gethostname()
- capabilities = {'location_info': 'LVMVolumeDriver:%s:'
- 'cinder-volumes:default:0' % hostname}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes-old',
- False, None, 'default')
- self.assertRaises(processutils.ProcessExecutionError,
- self.volume.driver.migrate_volume, self.context,
- vol, host)
- mock_delete.assert_called_once_with(vol)
+ lvm_driver._delete_volume(fake_snapshot, is_snapshot=True)
- def test_lvm_volume_group_missing(self):
- hostname = socket.gethostname()
- capabilities = {'location_info': 'LVMVolumeDriver:%s:'
- 'cinder-volumes-3:default:0' % hostname}
- host = {'capabilities': capabilities}
- vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ def test_check_for_setup_error(self):
- def get_all_volume_groups():
- return [{'name': 'cinder-volumes-2'}]
+ def get_all_volume_groups(vg):
+ return [{'name': 'cinder-volumes'}]
self.stubs.Set(volutils, 'get_all_volume_groups',
get_all_volume_groups)
- self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
+ vg_obj = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
- moved, model_update = self.volume.driver.migrate_volume(self.context,
- vol, host)
- self.assertFalse(moved)
- self.assertIsNone(model_update)
+ configuration = conf.Configuration(fake_opt, 'fake_group')
+ lvm_driver = lvm.LVMVolumeDriver(configuration=configuration,
+ vg_obj=vg_obj, db=db)
- def test_lvm_migrate_volume_proceed(self):
- hostname = socket.gethostname()
- capabilities = {'location_info': 'LVMVolumeDriver:%s:'
- 'cinder-volumes-2:default:0' % hostname}
- host = {'capabilities': capabilities}
- vol = {'name': 'testvol', 'id': 1, 'size': 2, 'status': 'available'}
+ lvm_driver.delete_snapshot = mock.Mock()
+ self.stubs.Set(volutils, 'get_all_volume_groups',
+ get_all_volume_groups)
- def fake_execute(*args, **kwargs):
- pass
+ volume = tests_utils.create_volume(self.context,
+ host=socket.gethostname())
+ volume_id = volume['id']
- def get_all_volume_groups():
- # NOTE(flaper87) Return just the destination
- # host to test the check of dest VG existence.
- return [{'name': 'cinder-volumes-2'}]
+ backup = {}
+ backup['volume_id'] = volume_id
+ backup['user_id'] = 'fake'
+ backup['project_id'] = 'fake'
+ backup['host'] = socket.gethostname()
+ backup['availability_zone'] = '1'
+ backup['display_name'] = 'test_check_for_setup_error'
+ backup['display_description'] = 'test_check_for_setup_error'
+ backup['container'] = 'fake'
+ backup['status'] = 'creating'
+ backup['fail_reason'] = ''
+ backup['service'] = 'fake'
+ backup['parent_id'] = None
+ backup['size'] = 5 * 1024 * 1024
+ backup['object_count'] = 22
+ db.backup_create(self.context, backup)
- def _fake_get_all_physical_volumes(obj, root_helper, vg_name):
- return [{}]
+ lvm_driver.check_for_setup_error()
- with mock.patch.object(brick_lvm.LVM, 'get_all_physical_volumes',
- return_value = [{}]), \
- mock.patch.object(self.volume.driver, '_execute') \
- as mock_execute, \
- mock.patch.object(volutils, 'copy_volume') as mock_copy, \
- mock.patch.object(volutils, 'get_all_volume_groups',
- side_effect = get_all_volume_groups), \
- mock.patch.object(self.volume.driver, '_delete_volume'):
+ @mock.patch.object(utils, 'temporary_chown')
+ @mock.patch('six.moves.builtins.open')
+ @mock.patch.object(os_brick.initiator.connector,
+ 'get_connector_properties')
+ @mock.patch.object(db.sqlalchemy.api, 'volume_get')
+ def test_backup_volume(self, mock_volume_get,
+ mock_get_connector_properties,
+ mock_file_open,
+ mock_temporary_chown):
+ vol = tests_utils.create_volume(self.context)
+ self.context.user_id = 'fake'
+ self.context.project_id = 'fake'
+ backup = tests_utils.create_backup(self.context,
+ vol['id'])
+ backup_obj = objects.Backup.get_by_id(self.context, backup.id)
- self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
- moved, model_update = \
- self.volume.driver.migrate_volume(self.context, vol, host)
- self.assertTrue(moved)
- self.assertIsNone(model_update)
- mock_copy.assert_called_once_with(
- '/dev/mapper/cinder--volumes-testvol',
- '/dev/mapper/cinder--volumes--2-testvol',
- 2048,
- '1M',
- execute=mock_execute,
- sparse=False)
+ properties = {}
+ attach_info = {'device': {'path': '/dev/null'}}
+ backup_service = mock.Mock()
- def test_lvm_migrate_volume_proceed_with_thin(self):
- hostname = socket.gethostname()
- capabilities = {'location_info': 'LVMVolumeDriver:%s:'
- 'cinder-volumes-2:default:0' % hostname}
- host = {'capabilities': capabilities}
- vol = {'name': 'testvol', 'id': 1, 'size': 2, 'status': 'available'}
+ self.volume.driver._detach_volume = mock.MagicMock()
+ self.volume.driver._attach_volume = mock.MagicMock()
+ self.volume.driver.terminate_connection = mock.MagicMock()
- def fake_execute(*args, **kwargs):
- pass
+ mock_volume_get.return_value = vol
+ mock_get_connector_properties.return_value = properties
+ f = mock_file_open.return_value = open('/dev/null', 'rb')
- def get_all_volume_groups():
- # NOTE(flaper87) Return just the destination
- # host to test the check of dest VG existence.
- return [{'name': 'cinder-volumes-2'}]
+ backup_service.backup(backup_obj, f, None)
+ self.volume.driver._attach_volume.return_value = attach_info
- def _fake_get_all_physical_volumes(obj, root_helper, vg_name):
- return [{}]
+ self.volume.driver.backup_volume(self.context, backup_obj,
+ backup_service)
- self.configuration.lvm_type = 'thin'
- lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
- db=db)
+ mock_volume_get.assert_called_with(self.context, vol['id'])
- with mock.patch.object(brick_lvm.LVM, 'get_all_physical_volumes',
- return_value = [{}]), \
- mock.patch.object(lvm_driver, '_execute') \
- as mock_execute, \
- mock.patch.object(volutils, 'copy_volume') as mock_copy, \
- mock.patch.object(volutils, 'get_all_volume_groups',
- side_effect = get_all_volume_groups), \
- mock.patch.object(lvm_driver, '_delete_volume'):
+ def test_retype_volume(self):
+ vol = tests_utils.create_volume(self.context)
+ new_type = 'fake'
+ diff = {}
+ host = 'fake_host'
+ retyped = self.volume.driver.retype(self.context, vol, new_type,
+ diff, host)
+ self.assertTrue(retyped)
- lvm_driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
- lvm_driver._sparse_copy_volume = True
- moved, model_update = \
- lvm_driver.migrate_volume(self.context, vol, host)
- self.assertTrue(moved)
- self.assertIsNone(model_update)
- mock_copy.assert_called_once_with(
- '/dev/mapper/cinder--volumes-testvol',
- '/dev/mapper/cinder--volumes--2-testvol',
- 2048,
- '1M',
- execute=mock_execute,
- sparse=True)
+ def test_update_migrated_volume(self):
+ fake_volume_id = 'vol1'
+ fake_new_volume_id = 'vol2'
+ fake_provider = 'fake_provider'
+ original_volume_name = CONF.volume_name_template % fake_volume_id
+ current_name = CONF.volume_name_template % fake_new_volume_id
+ fake_volume = tests_utils.create_volume(self.context)
+ fake_volume['id'] = fake_volume_id
+ fake_new_volume = tests_utils.create_volume(self.context)
+ fake_new_volume['id'] = fake_new_volume_id
+ fake_new_volume['provider_location'] = fake_provider
+ fake_vg = fake_lvm.FakeBrickLVM('cinder-volumes', False,
+ None, 'default')
+ with mock.patch.object(self.volume.driver, 'vg') as vg:
+ vg.return_value = fake_vg
+ vg.rename_volume.return_value = None
+ update = self.volume.driver.update_migrated_volume(self.context,
+ fake_volume,
+ fake_new_volume,
+ 'available')
+ vg.rename_volume.assert_called_once_with(current_name,
+ original_volume_name)
+ self.assertEqual({'_name_id': None,
+ 'provider_location': None}, update)
- @staticmethod
- def _get_manage_existing_lvs(name):
- """Helper method used by the manage_existing tests below."""
- lvs = [{'name': 'fake_lv', 'size': '1.75'},
- {'name': 'fake_lv_bad_size', 'size': 'Not a float'}]
- for lv in lvs:
- if lv['name'] == name:
- return lv
+ vg.rename_volume.reset_mock()
+ vg.rename_volume.side_effect = processutils.ProcessExecutionError
+ update = self.volume.driver.update_migrated_volume(self.context,
+ fake_volume,
+ fake_new_volume,
+ 'available')
+ vg.rename_volume.assert_called_once_with(current_name,
+ original_volume_name)
+ self.assertEqual({'_name_id': fake_new_volume_id,
+ 'provider_location': fake_provider},
+ update)
- def _setup_stubs_for_manage_existing(self):
- """Helper to set up common stubs for the manage_existing tests."""
- self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
- self.stubs.Set(self.volume.driver.vg, 'get_volume',
- self._get_manage_existing_lvs)
+ @mock.patch.object(utils, 'temporary_chown')
+ @mock.patch('six.moves.builtins.open')
+ @mock.patch.object(os_brick.initiator.connector,
+ 'get_connector_properties')
+ @mock.patch.object(db.sqlalchemy.api, 'volume_get')
+ def test_backup_volume_inuse(self, mock_volume_get,
+ mock_get_connector_properties,
+ mock_file_open,
+ mock_temporary_chown):
- @mock.patch.object(db.sqlalchemy.api, 'volume_get',
- side_effect=exception.VolumeNotFound(
- volume_id='d8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'))
- def test_lvm_manage_existing_not_found(self, mock_vol_get):
- self._setup_stubs_for_manage_existing()
+ vol = tests_utils.create_volume(self.context,
+ status='backing-up',
+ previous_status='in-use')
+ self.context.user_id = 'fake'
+ self.context.project_id = 'fake'
- vol_name = 'volume-d8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
- ref = {'source-name': 'fake_lv'}
- vol = {'name': vol_name, 'id': 1, 'size': 0}
+ mock_volume_get.return_value = vol
+ temp_snapshot = tests_utils.create_snapshot(self.context, vol['id'])
+ backup = tests_utils.create_backup(self.context,
+ vol['id'])
+ backup_obj = objects.Backup.get_by_id(self.context, backup.id)
+ properties = {}
+ attach_info = {'device': {'path': '/dev/null'}}
+ backup_service = mock.Mock()
- with mock.patch.object(self.volume.driver.vg, 'rename_volume'):
- model_update = self.volume.driver.manage_existing(vol, ref)
- self.assertIsNone(model_update)
+ self.volume.driver._detach_volume = mock.MagicMock()
+ self.volume.driver._attach_volume = mock.MagicMock()
+ self.volume.driver.terminate_connection = mock.MagicMock()
+ self.volume.driver._create_temp_snapshot = mock.MagicMock()
+ self.volume.driver._delete_temp_snapshot = mock.MagicMock()
- @mock.patch.object(db.sqlalchemy.api, 'volume_get')
- def test_lvm_manage_existing_already_managed(self, mock_conf):
- self._setup_stubs_for_manage_existing()
+ mock_get_connector_properties.return_value = properties
+ f = mock_file_open.return_value = open('/dev/null', 'rb')
- mock_conf.volume_name_template = 'volume-%s'
- vol_name = 'volume-d8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
- ref = {'source-name': vol_name}
- vol = {'name': 'test', 'id': 1, 'size': 0}
+ backup_service.backup(backup_obj, f, None)
+ self.volume.driver._attach_volume.return_value = attach_info
+ self.volume.driver._create_temp_snapshot.return_value = temp_snapshot
- with mock.patch.object(self.volume.driver.vg, 'rename_volume'):
- self.assertRaises(exception.ManageExistingAlreadyManaged,
- self.volume.driver.manage_existing,
- vol, ref)
+ self.volume.driver.backup_volume(self.context, backup_obj,
+ backup_service)
- def test_lvm_manage_existing(self):
- """Good pass on managing an LVM volume.
+ mock_volume_get.assert_called_with(self.context, vol['id'])
+ self.volume.driver._create_temp_snapshot.assert_called_once_with(
+ self.context, vol)
+ self.volume.driver._delete_temp_snapshot.assert_called_once_with(
+ self.context, temp_snapshot)
- This test case ensures that, when a logical volume with the
- specified name exists, and the size is as expected, no error is
- returned from driver.manage_existing, and that the rename_volume
- function is called in the Brick LVM code with the correct arguments.
- """
- self._setup_stubs_for_manage_existing()
+ def test_create_volume_from_snapshot_none_sparse(self):
- ref = {'source-name': 'fake_lv'}
- vol = {'name': 'test', 'id': 1, 'size': 0}
+ with mock.patch.object(self.volume.driver, 'vg'), \
+ mock.patch.object(self.volume.driver, '_create_volume'), \
+ mock.patch.object(volutils, 'copy_volume') as mock_copy:
- def _rename_volume(old_name, new_name):
- self.assertEqual(ref['source-name'], old_name)
- self.assertEqual(vol['name'], new_name)
+ # Test case for thick LVM
+ src_volume = tests_utils.create_volume(self.context)
+ snapshot_ref = tests_utils.create_snapshot(self.context,
+ src_volume['id'])
+ dst_volume = tests_utils.create_volume(self.context)
+ self.volume.driver.create_volume_from_snapshot(dst_volume,
+ snapshot_ref)
- self.stubs.Set(self.volume.driver.vg, 'rename_volume',
- _rename_volume)
+ volume_path = self.volume.driver.local_path(dst_volume)
+ snapshot_path = self.volume.driver.local_path(snapshot_ref)
+ volume_size = 1024
+ block_size = '1M'
+ mock_copy.assert_called_with(snapshot_path,
+ volume_path,
+ volume_size,
+ block_size,
+ execute=self.volume.driver._execute,
+ sparse=False)
- size = self.volume.driver.manage_existing_get_size(vol, ref)
- self.assertEqual(2, size)
- model_update = self.volume.driver.manage_existing(vol, ref)
- self.assertIsNone(model_update)
+ def test_create_volume_from_snapshot_sparse(self):
- def test_lvm_manage_existing_bad_size(self):
- """Make sure correct exception on bad size returned from LVM.
+ self.configuration.lvm_type = 'thin'
+ lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
+ db=db)
- This test case ensures that the correct exception is raised when
- the information returned for the existing LVs is not in the format
- that the manage_existing code expects.
- """
- self._setup_stubs_for_manage_existing()
+ with mock.patch.object(lvm_driver, 'vg'), \
+ mock.patch.object(lvm_driver, '_create_volume'), \
+ mock.patch.object(volutils, 'copy_volume') as mock_copy:
- ref = {'source-name': 'fake_lv_bad_size'}
- vol = {'name': 'test', 'id': 1, 'size': 2}
+ # Test case for thin LVM
+ lvm_driver._sparse_copy_volume = True
+ src_volume = tests_utils.create_volume(self.context)
+ snapshot_ref = tests_utils.create_snapshot(self.context,
+ src_volume['id'])
+ dst_volume = tests_utils.create_volume(self.context)
+ lvm_driver.create_volume_from_snapshot(dst_volume,
+ snapshot_ref)
- self.assertRaises(exception.VolumeBackendAPIException,
- self.volume.driver.manage_existing_get_size,
- vol, ref)
+ volume_path = lvm_driver.local_path(dst_volume)
+ snapshot_path = lvm_driver.local_path(snapshot_ref)
+ volume_size = 1024
+ block_size = '1M'
+ mock_copy.assert_called_with(snapshot_path,
+ volume_path,
+ volume_size,
+ block_size,
+ execute=lvm_driver._execute,
+ sparse=True)
- def test_lvm_manage_existing_bad_ref(self):
- """Error case where specified LV doesn't exist.
+ @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
+ return_value=[{'name': 'cinder-volumes'}])
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
+ return_value=True)
+ def test_lvm_type_auto_thin_pool_exists(self, *_unused_mocks):
+ configuration = conf.Configuration(fake_opt, 'fake_group')
+ configuration.lvm_type = 'auto'
- This test case ensures that the correct exception is raised when
- the caller attempts to manage a volume that does not exist.
- """
- self._setup_stubs_for_manage_existing()
+ vg_obj = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
- ref = {'source-name': 'fake_nonexistent_lv'}
- vol = {'name': 'test', 'id': 1, 'size': 0, 'status': 'available'}
+ lvm_driver = lvm.LVMVolumeDriver(configuration=configuration,
+ vg_obj=vg_obj)
- self.assertRaises(exception.ManageExistingInvalidReference,
- self.volume.driver.manage_existing_get_size,
- vol, ref)
+ lvm_driver.check_for_setup_error()
- def test_lvm_manage_existing_snapshot(self):
- """Good pass on managing an LVM snapshot.
-
- This test case ensures that, when a logical volume's snapshot with the
- specified name exists, and the size is as expected, no error is
- returned from driver.manage_existing_snapshot, and that the
- rename_volume function is called in the Brick LVM code with the correct
- arguments.
- """
- self._setup_stubs_for_manage_existing()
-
- ref = {'source-name': 'fake_lv'}
- snp = {'name': 'test', 'id': 1, 'size': 0}
-
- def _rename_volume(old_name, new_name):
- self.assertEqual(ref['source-name'], old_name)
- self.assertEqual(snp['name'], new_name)
+ self.assertEqual('thin', lvm_driver.configuration.lvm_type)
- with mock.patch.object(self.volume.driver.vg, 'rename_volume') as \
- mock_rename_volume:
- mock_rename_volume.return_value = _rename_volume
+ @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
+ return_value=[{'name': 'cinder-volumes'}])
+ @mock.patch.object(cinder.brick.local_dev.lvm.LVM, 'get_volumes',
+ return_value=[])
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
+ return_value=True)
+ def test_lvm_type_auto_no_lvs(self, *_unused_mocks):
+ configuration = conf.Configuration(fake_opt, 'fake_group')
+ configuration.lvm_type = 'auto'
- size = self.volume.driver.manage_existing_snapshot_get_size(snp,
- ref)
- self.assertEqual(2, size)
- model_update = self.volume.driver.manage_existing_snapshot(snp,
- ref)
- self.assertIsNone(model_update)
+ vg_obj = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
- def test_lvm_manage_existing_snapshot_bad_ref(self):
- """Error case where specified LV snapshot doesn't exist.
+ lvm_driver = lvm.LVMVolumeDriver(configuration=configuration,
+ vg_obj=vg_obj)
- This test case ensures that the correct exception is raised when
- the caller attempts to manage a snapshot that does not exist.
- """
- self._setup_stubs_for_manage_existing()
+ lvm_driver.check_for_setup_error()
- ref = {'source-name': 'fake_nonexistent_lv'}
- snp = {'name': 'test', 'id': 1, 'size': 0, 'status': 'available'}
+ self.assertEqual('thin', lvm_driver.configuration.lvm_type)
- self.assertRaises(exception.ManageExistingInvalidReference,
- self.volume.driver.manage_existing_snapshot_get_size,
- snp, ref)
+ @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
+ return_value=[{'name': 'cinder-volumes'}])
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
+ return_value=False)
+ def test_lvm_type_auto_no_thin_support(self, *_unused_mocks):
+ configuration = conf.Configuration(fake_opt, 'fake_group')
+ configuration.lvm_type = 'auto'
- def test_lvm_manage_existing_snapshot_bad_size(self):
- """Make sure correct exception on bad size returned from LVM.
+ lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
- This test case ensures that the correct exception is raised when
- the information returned for the existing LVs is not in the format
- that the manage_existing_snapshot code expects.
- """
- self._setup_stubs_for_manage_existing()
+ lvm_driver.check_for_setup_error()
- ref = {'source-name': 'fake_lv_bad_size'}
- snp = {'name': 'test', 'id': 1, 'size': 2}
+ self.assertEqual('default', lvm_driver.configuration.lvm_type)
- self.assertRaises(exception.VolumeBackendAPIException,
- self.volume.driver.manage_existing_snapshot_get_size,
- snp, ref)
+ @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
+ return_value=[{'name': 'cinder-volumes'}])
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.get_volume')
+ @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
+ return_value=False)
+ def test_lvm_type_auto_no_thin_pool(self, *_unused_mocks):
+ configuration = conf.Configuration(fake_opt, 'fake_group')
+ configuration.lvm_type = 'auto'
- def test_lvm_unmanage(self):
- volume = tests_utils.create_volume(self.context, status='available',
- size=1, host=CONF.host)
- ret = self.volume.driver.unmanage(volume)
- self.assertEqual(ret, None)
+ lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
+ lvm_driver.check_for_setup_error()
-class LVMVolumeDriverTestCase(DriverTestCase):
- """Test case for VolumeDriver"""
- driver_name = "cinder.volume.drivers.lvm.LVMVolumeDriver"
- FAKE_VOLUME = {'name': 'test1',
- 'id': 'test1'}
+ self.assertEqual('default', lvm_driver.configuration.lvm_type)
- @mock.patch.object(fake_driver.FakeISCSIDriver, 'create_export')
- def test_delete_volume_invalid_parameter(self, _mock_create_export):
- self.configuration.volume_clear = 'zero'
- self.configuration.volume_clear_size = 0
+ @mock.patch.object(lvm.LVMVolumeDriver, 'extend_volume')
+ def test_create_cloned_volume_by_thin_snapshot(self, mock_extend):
+ self.configuration.lvm_type = 'thin'
+ fake_vg = mock.Mock(fake_lvm.FakeBrickLVM('cinder-volumes', False,
+ None, 'default'))
lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
+ vg_obj=fake_vg,
db=db)
- self.mox.StubOutWithMock(os.path, 'exists')
-
- os.path.exists(mox.IgnoreArg()).AndReturn(True)
+ fake_volume = tests_utils.create_volume(self.context, size=1)
+ fake_new_volume = tests_utils.create_volume(self.context, size=2)
- self.mox.ReplayAll()
- # Test volume without 'size' field and 'volume_size' field
- self.assertRaises(exception.InvalidParameterValue,
- lvm_driver._delete_volume,
- self.FAKE_VOLUME)
+ lvm_driver.create_cloned_volume(fake_new_volume, fake_volume)
+ fake_vg.create_lv_snapshot.assert_called_once_with(
+ fake_new_volume['name'], fake_volume['name'], 'thin')
+ mock_extend.assert_called_once_with(fake_new_volume, 2)
+ fake_vg.activate_lv.assert_called_once_with(
+ fake_new_volume['name'], is_snapshot=True, permanent=True)
- @mock.patch.object(fake_driver.FakeISCSIDriver, 'create_export')
- def test_delete_volume_bad_path(self, _mock_create_export):
- self.configuration.volume_clear = 'zero'
- self.configuration.volume_clear_size = 0
- self.configuration.volume_type = 'default'
+ def test_lvm_migrate_volume_no_loc_info(self):
+ host = {'capabilities': {}}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ moved, model_update = self.volume.driver.migrate_volume(self.context,
+ vol, host)
+ self.assertFalse(moved)
+ self.assertIsNone(model_update)
- volume = dict(self.FAKE_VOLUME, size=1)
- lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
- db=db)
+ def test_lvm_migrate_volume_bad_loc_info(self):
+ capabilities = {'location_info': 'foo'}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ moved, model_update = self.volume.driver.migrate_volume(self.context,
+ vol, host)
+ self.assertFalse(moved)
+ self.assertIsNone(model_update)
- self.mox.StubOutWithMock(os.path, 'exists')
- os.path.exists(mox.IgnoreArg()).AndReturn(False)
- self.mox.ReplayAll()
+ def test_lvm_migrate_volume_diff_driver(self):
+ capabilities = {'location_info': 'FooDriver:foo:bar:default:0'}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ moved, model_update = self.volume.driver.migrate_volume(self.context,
+ vol, host)
+ self.assertFalse(moved)
+ self.assertIsNone(model_update)
- self.assertRaises(exception.VolumeBackendAPIException,
- lvm_driver._delete_volume, volume)
+ def test_lvm_migrate_volume_diff_host(self):
+ capabilities = {'location_info': 'LVMVolumeDriver:foo:bar:default:0'}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ moved, model_update = self.volume.driver.migrate_volume(self.context,
+ vol, host)
+ self.assertFalse(moved)
+ self.assertIsNone(model_update)
- @mock.patch.object(fake_driver.FakeISCSIDriver, 'create_export')
- def test_delete_volume_thinlvm_snap(self, _mock_create_export):
- self.configuration.volume_clear = 'zero'
- self.configuration.volume_clear_size = 0
- self.configuration.lvm_type = 'thin'
- self.configuration.iscsi_helper = 'tgtadm'
- lvm_driver = lvm.LVMISCSIDriver(configuration=self.configuration,
- vg_obj=mox.MockAnything(),
- db=db)
+ def test_lvm_migrate_volume_in_use(self):
+ hostname = socket.gethostname()
+ capabilities = {'location_info': 'LVMVolumeDriver:%s:bar' % hostname}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'in-use'}
+ moved, model_update = self.volume.driver.migrate_volume(self.context,
+ vol, host)
+ self.assertFalse(moved)
+ self.assertIsNone(model_update)
- # Ensures that copy_volume is not called for ThinLVM
- self.mox.StubOutWithMock(volutils, 'copy_volume')
- self.mox.StubOutWithMock(volutils, 'clear_volume')
- self.mox.StubOutWithMock(lvm_driver, '_execute')
- self.mox.ReplayAll()
+ @mock.patch.object(volutils, 'get_all_volume_groups',
+ return_value=[{'name': 'cinder-volumes'}])
+ def test_lvm_migrate_volume_same_volume_group(self, vgs):
+ hostname = socket.gethostname()
+ capabilities = {'location_info': 'LVMVolumeDriver:%s:'
+ 'cinder-volumes:default:0' % hostname}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
- uuid = '00000000-0000-0000-0000-c3aa7ee01536'
+ self.assertRaises(exception.VolumeBackendAPIException,
+ self.volume.driver.migrate_volume, self.context,
+ vol, host)
- fake_snapshot = {'name': 'volume-' + uuid,
- 'id': uuid,
- 'size': 123}
+ @mock.patch.object(lvm.LVMVolumeDriver, '_create_volume')
+ @mock.patch.object(brick_lvm.LVM, 'get_all_physical_volumes')
+ @mock.patch.object(brick_lvm.LVM, 'delete')
+ @mock.patch.object(volutils, 'copy_volume',
+ side_effect=processutils.ProcessExecutionError)
+ @mock.patch.object(volutils, 'get_all_volume_groups',
+ return_value=[{'name': 'cinder-volumes'}])
+ def test_lvm_migrate_volume_volume_copy_error(self, vgs, copy_volume,
+ mock_delete, mock_pvs,
+ mock_create):
- lvm_driver._delete_volume(fake_snapshot, is_snapshot=True)
+ hostname = socket.gethostname()
+ capabilities = {'location_info': 'LVMVolumeDriver:%s:'
+ 'cinder-volumes:default:0' % hostname}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
+ self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes-old',
+ False, None, 'default')
+ self.assertRaises(processutils.ProcessExecutionError,
+ self.volume.driver.migrate_volume, self.context,
+ vol, host)
+ mock_delete.assert_called_once_with(vol)
- def test_check_for_setup_error(self):
+ def test_lvm_volume_group_missing(self):
+ hostname = socket.gethostname()
+ capabilities = {'location_info': 'LVMVolumeDriver:%s:'
+ 'cinder-volumes-3:default:0' % hostname}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'test', 'id': 1, 'size': 1, 'status': 'available'}
- def get_all_volume_groups(vg):
- return [{'name': 'cinder-volumes'}]
+ def get_all_volume_groups():
+ return [{'name': 'cinder-volumes-2'}]
self.stubs.Set(volutils, 'get_all_volume_groups',
get_all_volume_groups)
- vg_obj = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
-
- configuration = conf.Configuration(fake_opt, 'fake_group')
- lvm_driver = lvm.LVMVolumeDriver(configuration=configuration,
- vg_obj=vg_obj, db=db)
-
- lvm_driver.delete_snapshot = mock.Mock()
- self.stubs.Set(volutils, 'get_all_volume_groups',
- get_all_volume_groups)
+ self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
- volume = tests_utils.create_volume(self.context,
- host=socket.gethostname())
- volume_id = volume['id']
+ moved, model_update = self.volume.driver.migrate_volume(self.context,
+ vol, host)
+ self.assertFalse(moved)
+ self.assertIsNone(model_update)
- backup = {}
- backup['volume_id'] = volume_id
- backup['user_id'] = 'fake'
- backup['project_id'] = 'fake'
- backup['host'] = socket.gethostname()
- backup['availability_zone'] = '1'
- backup['display_name'] = 'test_check_for_setup_error'
- backup['display_description'] = 'test_check_for_setup_error'
- backup['container'] = 'fake'
- backup['status'] = 'creating'
- backup['fail_reason'] = ''
- backup['service'] = 'fake'
- backup['parent_id'] = None
- backup['size'] = 5 * 1024 * 1024
- backup['object_count'] = 22
- db.backup_create(self.context, backup)
+ def test_lvm_migrate_volume_proceed(self):
+ hostname = socket.gethostname()
+ capabilities = {'location_info': 'LVMVolumeDriver:%s:'
+ 'cinder-volumes-2:default:0' % hostname}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'testvol', 'id': 1, 'size': 2, 'status': 'available'}
- lvm_driver.check_for_setup_error()
+ def fake_execute(*args, **kwargs):
+ pass
- @mock.patch.object(utils, 'temporary_chown')
- @mock.patch('six.moves.builtins.open')
- @mock.patch.object(os_brick.initiator.connector,
- 'get_connector_properties')
- @mock.patch.object(db.sqlalchemy.api, 'volume_get')
- def test_backup_volume(self, mock_volume_get,
- mock_get_connector_properties,
- mock_file_open,
- mock_temporary_chown):
- vol = tests_utils.create_volume(self.context)
- self.context.user_id = 'fake'
- self.context.project_id = 'fake'
- backup = tests_utils.create_backup(self.context,
- vol['id'])
- backup_obj = objects.Backup.get_by_id(self.context, backup.id)
+ def get_all_volume_groups():
+ # NOTE(flaper87) Return just the destination
+ # host to test the check of dest VG existence.
+ return [{'name': 'cinder-volumes-2'}]
- properties = {}
- attach_info = {'device': {'path': '/dev/null'}}
- backup_service = mock.Mock()
+ def _fake_get_all_physical_volumes(obj, root_helper, vg_name):
+ return [{}]
- self.volume.driver._detach_volume = mock.MagicMock()
- self.volume.driver._attach_volume = mock.MagicMock()
- self.volume.driver.terminate_connection = mock.MagicMock()
+ with mock.patch.object(brick_lvm.LVM, 'get_all_physical_volumes',
+ return_value = [{}]), \
+ mock.patch.object(self.volume.driver, '_execute') \
+ as mock_execute, \
+ mock.patch.object(volutils, 'copy_volume') as mock_copy, \
+ mock.patch.object(volutils, 'get_all_volume_groups',
+ side_effect = get_all_volume_groups), \
+ mock.patch.object(self.volume.driver, '_delete_volume'):
- mock_volume_get.return_value = vol
- mock_get_connector_properties.return_value = properties
- f = mock_file_open.return_value = open('/dev/null', 'rb')
+ self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
+ moved, model_update = \
+ self.volume.driver.migrate_volume(self.context, vol, host)
+ self.assertTrue(moved)
+ self.assertIsNone(model_update)
+ mock_copy.assert_called_once_with(
+ '/dev/mapper/cinder--volumes-testvol',
+ '/dev/mapper/cinder--volumes--2-testvol',
+ 2048,
+ '1M',
+ execute=mock_execute,
+ sparse=False)
- backup_service.backup(backup_obj, f, None)
- self.volume.driver._attach_volume.return_value = attach_info
+ def test_lvm_migrate_volume_proceed_with_thin(self):
+ hostname = socket.gethostname()
+ capabilities = {'location_info': 'LVMVolumeDriver:%s:'
+ 'cinder-volumes-2:default:0' % hostname}
+ host = {'capabilities': capabilities}
+ vol = {'name': 'testvol', 'id': 1, 'size': 2, 'status': 'available'}
- self.volume.driver.backup_volume(self.context, backup_obj,
- backup_service)
+ def fake_execute(*args, **kwargs):
+ pass
- mock_volume_get.assert_called_with(self.context, vol['id'])
+ def get_all_volume_groups():
+ # NOTE(flaper87) Return just the destination
+ # host to test the check of dest VG existence.
+ return [{'name': 'cinder-volumes-2'}]
- def test_retype_volume(self):
- vol = tests_utils.create_volume(self.context)
- new_type = 'fake'
- diff = {}
- host = 'fake_host'
- retyped = self.volume.driver.retype(self.context, vol, new_type,
- diff, host)
- self.assertTrue(retyped)
+ def _fake_get_all_physical_volumes(obj, root_helper, vg_name):
+ return [{}]
- def test_update_migrated_volume(self):
- fake_volume_id = 'vol1'
- fake_new_volume_id = 'vol2'
- fake_provider = 'fake_provider'
- original_volume_name = CONF.volume_name_template % fake_volume_id
- current_name = CONF.volume_name_template % fake_new_volume_id
- fake_volume = tests_utils.create_volume(self.context)
- fake_volume['id'] = fake_volume_id
- fake_new_volume = tests_utils.create_volume(self.context)
- fake_new_volume['id'] = fake_new_volume_id
- fake_new_volume['provider_location'] = fake_provider
- fake_vg = fake_lvm.FakeBrickLVM('cinder-volumes', False,
- None, 'default')
- with mock.patch.object(self.volume.driver, 'vg') as vg:
- vg.return_value = fake_vg
- vg.rename_volume.return_value = None
- update = self.volume.driver.update_migrated_volume(self.context,
- fake_volume,
- fake_new_volume,
- 'available')
- vg.rename_volume.assert_called_once_with(current_name,
- original_volume_name)
- self.assertEqual({'_name_id': None,
- 'provider_location': None}, update)
+ self.configuration.lvm_type = 'thin'
+ lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
+ db=db)
- vg.rename_volume.reset_mock()
- vg.rename_volume.side_effect = processutils.ProcessExecutionError
- update = self.volume.driver.update_migrated_volume(self.context,
- fake_volume,
- fake_new_volume,
- 'available')
- vg.rename_volume.assert_called_once_with(current_name,
- original_volume_name)
- self.assertEqual({'_name_id': fake_new_volume_id,
- 'provider_location': fake_provider},
- update)
+ with mock.patch.object(brick_lvm.LVM, 'get_all_physical_volumes',
+ return_value = [{}]), \
+ mock.patch.object(lvm_driver, '_execute') \
+ as mock_execute, \
+ mock.patch.object(volutils, 'copy_volume') as mock_copy, \
+ mock.patch.object(volutils, 'get_all_volume_groups',
+ side_effect = get_all_volume_groups), \
+ mock.patch.object(lvm_driver, '_delete_volume'):
- @mock.patch.object(utils, 'temporary_chown')
- @mock.patch('six.moves.builtins.open')
- @mock.patch.object(os_brick.initiator.connector,
- 'get_connector_properties')
- @mock.patch.object(db.sqlalchemy.api, 'volume_get')
- def test_backup_volume_inuse(self, mock_volume_get,
- mock_get_connector_properties,
- mock_file_open,
- mock_temporary_chown):
+ lvm_driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
+ lvm_driver._sparse_copy_volume = True
+ moved, model_update = \
+ lvm_driver.migrate_volume(self.context, vol, host)
+ self.assertTrue(moved)
+ self.assertIsNone(model_update)
+ mock_copy.assert_called_once_with(
+ '/dev/mapper/cinder--volumes-testvol',
+ '/dev/mapper/cinder--volumes--2-testvol',
+ 2048,
+ '1M',
+ execute=mock_execute,
+ sparse=True)
- vol = tests_utils.create_volume(self.context,
- status='backing-up',
- previous_status='in-use')
- self.context.user_id = 'fake'
- self.context.project_id = 'fake'
+ @staticmethod
+ def _get_manage_existing_lvs(name):
+ """Helper method used by the manage_existing tests below."""
+ lvs = [{'name': 'fake_lv', 'size': '1.75'},
+ {'name': 'fake_lv_bad_size', 'size': 'Not a float'}]
+ for lv in lvs:
+ if lv['name'] == name:
+ return lv
- mock_volume_get.return_value = vol
- temp_snapshot = tests_utils.create_snapshot(self.context, vol['id'])
- backup = tests_utils.create_backup(self.context,
- vol['id'])
- backup_obj = objects.Backup.get_by_id(self.context, backup.id)
- properties = {}
- attach_info = {'device': {'path': '/dev/null'}}
- backup_service = mock.Mock()
+ def _setup_stubs_for_manage_existing(self):
+ """Helper to set up common stubs for the manage_existing tests."""
+ self.volume.driver.vg = fake_lvm.FakeBrickLVM('cinder-volumes',
+ False,
+ None,
+ 'default')
+ self.stubs.Set(self.volume.driver.vg, 'get_volume',
+ self._get_manage_existing_lvs)
- self.volume.driver._detach_volume = mock.MagicMock()
- self.volume.driver._attach_volume = mock.MagicMock()
- self.volume.driver.terminate_connection = mock.MagicMock()
- self.volume.driver._create_temp_snapshot = mock.MagicMock()
- self.volume.driver._delete_temp_snapshot = mock.MagicMock()
+ @mock.patch.object(db.sqlalchemy.api, 'volume_get',
+ side_effect=exception.VolumeNotFound(
+ volume_id='d8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'))
+ def test_lvm_manage_existing_not_found(self, mock_vol_get):
+ self._setup_stubs_for_manage_existing()
- mock_get_connector_properties.return_value = properties
- f = mock_file_open.return_value = open('/dev/null', 'rb')
+ vol_name = 'volume-d8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+ ref = {'source-name': 'fake_lv'}
+ vol = {'name': vol_name, 'id': 1, 'size': 0}
- backup_service.backup(backup_obj, f, None)
- self.volume.driver._attach_volume.return_value = attach_info
- self.volume.driver._create_temp_snapshot.return_value = temp_snapshot
+ with mock.patch.object(self.volume.driver.vg, 'rename_volume'):
+ model_update = self.volume.driver.manage_existing(vol, ref)
+ self.assertIsNone(model_update)
- self.volume.driver.backup_volume(self.context, backup_obj,
- backup_service)
+ @mock.patch.object(db.sqlalchemy.api, 'volume_get')
+ def test_lvm_manage_existing_already_managed(self, mock_conf):
+ self._setup_stubs_for_manage_existing()
- mock_volume_get.assert_called_with(self.context, vol['id'])
- self.volume.driver._create_temp_snapshot.assert_called_once_with(
- self.context, vol)
- self.volume.driver._delete_temp_snapshot.assert_called_once_with(
- self.context, temp_snapshot)
+ mock_conf.volume_name_template = 'volume-%s'
+ vol_name = 'volume-d8cd1feb-2dcc-404d-9b15-b86fe3bec0a1'
+ ref = {'source-name': vol_name}
+ vol = {'name': 'test', 'id': 1, 'size': 0}
- def test_create_volume_from_snapshot_none_sparse(self):
+ with mock.patch.object(self.volume.driver.vg, 'rename_volume'):
+ self.assertRaises(exception.ManageExistingAlreadyManaged,
+ self.volume.driver.manage_existing,
+ vol, ref)
- with mock.patch.object(self.volume.driver, 'vg'), \
- mock.patch.object(self.volume.driver, '_create_volume'), \
- mock.patch.object(volutils, 'copy_volume') as mock_copy:
+ def test_lvm_manage_existing(self):
+ """Good pass on managing an LVM volume.
- # Test case for thick LVM
- src_volume = tests_utils.create_volume(self.context)
- snapshot_ref = tests_utils.create_snapshot(self.context,
- src_volume['id'])
- dst_volume = tests_utils.create_volume(self.context)
- self.volume.driver.create_volume_from_snapshot(dst_volume,
- snapshot_ref)
+ This test case ensures that, when a logical volume with the
+ specified name exists, and the size is as expected, no error is
+ returned from driver.manage_existing, and that the rename_volume
+ function is called in the Brick LVM code with the correct arguments.
+ """
+ self._setup_stubs_for_manage_existing()
- volume_path = self.volume.driver.local_path(dst_volume)
- snapshot_path = self.volume.driver.local_path(snapshot_ref)
- volume_size = 1024
- block_size = '1M'
- mock_copy.assert_called_with(snapshot_path,
- volume_path,
- volume_size,
- block_size,
- execute=self.volume.driver._execute,
- sparse=False)
+ ref = {'source-name': 'fake_lv'}
+ vol = {'name': 'test', 'id': 1, 'size': 0}
- def test_create_volume_from_snapshot_sparse(self):
+ def _rename_volume(old_name, new_name):
+ self.assertEqual(ref['source-name'], old_name)
+ self.assertEqual(vol['name'], new_name)
- self.configuration.lvm_type = 'thin'
- lvm_driver = lvm.LVMVolumeDriver(configuration=self.configuration,
- db=db)
+ self.stubs.Set(self.volume.driver.vg, 'rename_volume',
+ _rename_volume)
- with mock.patch.object(lvm_driver, 'vg'), \
- mock.patch.object(lvm_driver, '_create_volume'), \
- mock.patch.object(volutils, 'copy_volume') as mock_copy:
+ size = self.volume.driver.manage_existing_get_size(vol, ref)
+ self.assertEqual(2, size)
+ model_update = self.volume.driver.manage_existing(vol, ref)
+ self.assertIsNone(model_update)
- # Test case for thin LVM
- lvm_driver._sparse_copy_volume = True
- src_volume = tests_utils.create_volume(self.context)
- snapshot_ref = tests_utils.create_snapshot(self.context,
- src_volume['id'])
- dst_volume = tests_utils.create_volume(self.context)
- lvm_driver.create_volume_from_snapshot(dst_volume,
- snapshot_ref)
+ def test_lvm_manage_existing_bad_size(self):
+ """Make sure correct exception on bad size returned from LVM.
- volume_path = lvm_driver.local_path(dst_volume)
- snapshot_path = lvm_driver.local_path(snapshot_ref)
- volume_size = 1024
- block_size = '1M'
- mock_copy.assert_called_with(snapshot_path,
- volume_path,
- volume_size,
- block_size,
- execute=lvm_driver._execute,
- sparse=True)
+ This test case ensures that the correct exception is raised when
+ the information returned for the existing LVs is not in the format
+ that the manage_existing code expects.
+ """
+ self._setup_stubs_for_manage_existing()
- @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
- return_value=[{'name': 'cinder-volumes'}])
- @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
- return_value=True)
- def test_lvm_type_auto_thin_pool_exists(self, *_unused_mocks):
- configuration = conf.Configuration(fake_opt, 'fake_group')
- configuration.lvm_type = 'auto'
+ ref = {'source-name': 'fake_lv_bad_size'}
+ vol = {'name': 'test', 'id': 1, 'size': 2}
- vg_obj = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
+ self.assertRaises(exception.VolumeBackendAPIException,
+ self.volume.driver.manage_existing_get_size,
+ vol, ref)
- lvm_driver = lvm.LVMVolumeDriver(configuration=configuration,
- vg_obj=vg_obj)
+ def test_lvm_manage_existing_bad_ref(self):
+ """Error case where specified LV doesn't exist.
- lvm_driver.check_for_setup_error()
+ This test case ensures that the correct exception is raised when
+ the caller attempts to manage a volume that does not exist.
+ """
+ self._setup_stubs_for_manage_existing()
- self.assertEqual('thin', lvm_driver.configuration.lvm_type)
+ ref = {'source-name': 'fake_nonexistent_lv'}
+ vol = {'name': 'test', 'id': 1, 'size': 0, 'status': 'available'}
- @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
- return_value=[{'name': 'cinder-volumes'}])
- @mock.patch.object(cinder.brick.local_dev.lvm.LVM, 'get_volumes',
- return_value=[])
- @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
- return_value=True)
- def test_lvm_type_auto_no_lvs(self, *_unused_mocks):
- configuration = conf.Configuration(fake_opt, 'fake_group')
- configuration.lvm_type = 'auto'
+ self.assertRaises(exception.ManageExistingInvalidReference,
+ self.volume.driver.manage_existing_get_size,
+ vol, ref)
- vg_obj = fake_lvm.FakeBrickLVM('cinder-volumes',
- False,
- None,
- 'default')
+ def test_lvm_manage_existing_snapshot(self):
+ """Good pass on managing an LVM snapshot.
- lvm_driver = lvm.LVMVolumeDriver(configuration=configuration,
- vg_obj=vg_obj)
+ This test case ensures that, when a logical volume's snapshot with the
+ specified name exists, and the size is as expected, no error is
+ returned from driver.manage_existing_snapshot, and that the
+ rename_volume function is called in the Brick LVM code with the correct
+ arguments.
+ """
+ self._setup_stubs_for_manage_existing()
- lvm_driver.check_for_setup_error()
+ ref = {'source-name': 'fake_lv'}
+ snp = {'name': 'test', 'id': 1, 'size': 0}
- self.assertEqual('thin', lvm_driver.configuration.lvm_type)
+ def _rename_volume(old_name, new_name):
+ self.assertEqual(ref['source-name'], old_name)
+ self.assertEqual(snp['name'], new_name)
- @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
- return_value=[{'name': 'cinder-volumes'}])
- @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
- return_value=False)
- def test_lvm_type_auto_no_thin_support(self, *_unused_mocks):
- configuration = conf.Configuration(fake_opt, 'fake_group')
- configuration.lvm_type = 'auto'
+ with mock.patch.object(self.volume.driver.vg, 'rename_volume') as \
+ mock_rename_volume:
+ mock_rename_volume.return_value = _rename_volume
- lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
+ size = self.volume.driver.manage_existing_snapshot_get_size(snp,
+ ref)
+ self.assertEqual(2, size)
+ model_update = self.volume.driver.manage_existing_snapshot(snp,
+ ref)
+ self.assertIsNone(model_update)
- lvm_driver.check_for_setup_error()
+ def test_lvm_manage_existing_snapshot_bad_ref(self):
+ """Error case where specified LV snapshot doesn't exist.
- self.assertEqual('default', lvm_driver.configuration.lvm_type)
+ This test case ensures that the correct exception is raised when
+ the caller attempts to manage a snapshot that does not exist.
+ """
+ self._setup_stubs_for_manage_existing()
- @mock.patch.object(cinder.volume.utils, 'get_all_volume_groups',
- return_value=[{'name': 'cinder-volumes'}])
- @mock.patch('cinder.brick.local_dev.lvm.LVM.update_volume_group_info')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.get_all_physical_volumes')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.get_volume')
- @mock.patch('cinder.brick.local_dev.lvm.LVM.supports_thin_provisioning',
- return_value=False)
- def test_lvm_type_auto_no_thin_pool(self, *_unused_mocks):
- configuration = conf.Configuration(fake_opt, 'fake_group')
- configuration.lvm_type = 'auto'
+ ref = {'source-name': 'fake_nonexistent_lv'}
+ snp = {'name': 'test', 'id': 1, 'size': 0, 'status': 'available'}
- lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
+ self.assertRaises(exception.ManageExistingInvalidReference,
+ self.volume.driver.manage_existing_snapshot_get_size,
+ snp, ref)
- lvm_driver.check_for_setup_error()
+ def test_lvm_manage_existing_snapshot_bad_size(self):
+ """Make sure correct exception on bad size returned from LVM.
- self.assertEqual('default', lvm_driver.configuration.lvm_type)
+ This test case ensures that the correct exception is raised when
+ the information returned for the existing LVs is not in the format
+ that the manage_existing_snapshot code expects.
+ """
+ self._setup_stubs_for_manage_existing()
- @mock.patch.object(lvm.LVMISCSIDriver, 'extend_volume')
- def test_create_cloned_volume_by_thin_snapshot(self, mock_extend):
- self.configuration.lvm_type = 'thin'
- fake_vg = mock.Mock(fake_lvm.FakeBrickLVM('cinder-volumes', False,
- None, 'default'))
- lvm_driver = lvm.LVMISCSIDriver(configuration=self.configuration,
- vg_obj=fake_vg,
- db=db)
- fake_volume = tests_utils.create_volume(self.context, size=1)
- fake_new_volume = tests_utils.create_volume(self.context, size=2)
+ ref = {'source-name': 'fake_lv_bad_size'}
+ snp = {'name': 'test', 'id': 1, 'size': 2}
- lvm_driver.create_cloned_volume(fake_new_volume, fake_volume)
- fake_vg.create_lv_snapshot.assert_called_once_with(
- fake_new_volume['name'], fake_volume['name'], 'thin')
- mock_extend.assert_called_once_with(fake_new_volume, 2)
- fake_vg.activate_lv.assert_called_once_with(
- fake_new_volume['name'], is_snapshot=True, permanent=True)
+ self.assertRaises(exception.VolumeBackendAPIException,
+ self.volume.driver.manage_existing_snapshot_get_size,
+ snp, ref)
+
+ def test_lvm_unmanage(self):
+ volume = tests_utils.create_volume(self.context, status='available',
+ size=1, host=CONF.host)
+ ret = self.volume.driver.unmanage(volume)
+ self.assertEqual(ret, None)
class ISCSITestCase(DriverTestCase):
"""Test Case for ISCSIDriver"""
- driver_name = "cinder.volume.drivers.lvm.LVMISCSIDriver"
+ driver_name = "cinder.volume.drivers.lvm.LVMVolumeDriver"
def setUp(self):
super(ISCSITestCase, self).setUp()
iscsi_driver.validate_connector, connector)
-class ISERTestCase(DriverTestCase):
- """Test Case for ISERDriver."""
- driver_name = "cinder.volume.drivers.lvm.LVMISERDriver"
-
- def setUp(self):
- super(ISERTestCase, self).setUp()
- self.configuration = mock.Mock(conf.Configuration)
- self.configuration.safe_get.return_value = None
- self.configuration.num_iser_scan_tries = 3
- self.configuration.iser_target_prefix = 'iqn.2010-10.org.openstack:'
- self.configuration.iser_ip_address = '0.0.0.0'
- self.configuration.iser_port = 3260
- self.configuration.target_driver = \
- 'cinder.volume.targets.iser.ISERTgtAdm'
-
- @test.testtools.skip("SKIP until ISER driver is removed or fixed")
- def test_get_volume_stats(self):
- def _fake_get_all_physical_volumes(obj, root_helper, vg_name):
- return [{}]
-
- def _fake_get_all_volume_groups(obj, vg_name=None, no_suffix=True):
- return [{'name': 'cinder-volumes',
- 'size': '5.52',
- 'available': '0.52',
- 'lv_count': '2',
- 'uuid': 'vR1JU3-FAKE-C4A9-PQFh-Mctm-9FwA-Xwzc1m'}]
-
- self.stubs.Set(brick_lvm.LVM,
- 'get_all_physical_volumes',
- _fake_get_all_physical_volumes)
-
- self.stubs.Set(brick_lvm.LVM,
- 'get_all_volume_groups',
- _fake_get_all_volume_groups)
-
- self.volume_driver = \
- lvm.LVMISERDriver(configuration=self.configuration)
- self.volume.driver.vg = brick_lvm.LVM('cinder-volumes', 'sudo')
-
- stats = self.volume.driver.get_volume_stats(refresh=True)
-
- self.assertEqual(
- float('5.52'), stats['pools'][0]['total_capacity_gb'])
- self.assertEqual(
- float('0.52'), stats['pools'][0]['free_capacity_gb'])
- self.assertEqual(
- float('5.0'), stats['pools'][0]['provisioned_capacity_gb'])
- self.assertEqual('iSER', stats['storage_protocol'])
-
- @test.testtools.skip("SKIP until ISER driver is removed or fixed")
- def test_get_volume_stats2(self):
- iser_driver = lvm.LVMISERDriver(configuration=self.configuration)
-
- stats = iser_driver.get_volume_stats(refresh=True)
-
- self.assertEqual(
- 0, stats['pools'][0]['total_capacity_gb'])
- self.assertEqual(
- 0, stats['pools'][0]['free_capacity_gb'])
- self.assertEqual(
- float('5.0'), stats['pools'][0]['provisioned_capacity_gb'])
- self.assertEqual('iSER', stats['storage_protocol'])
-
-
class FibreChannelTestCase(DriverTestCase):
"""Test Case for FibreChannelDriver."""
driver_name = "cinder.volume.driver.FibreChannelDriver"
Code Review / openstack-build / cinder-build.git / commitdiff