From: Jenkins Date: Thu, 3 Mar 2016 02:24:24 +0000 (+0000) Subject: Merge "Remove duplicated code in volume manager and base driver" X-Git-Url: https://review.fuel-infra.org/gitweb?a=commitdiff_plain;h=8893ad83bed3c0ae0cee422e28af109a5a87a524;p=openstack-build%2Fcinder-build.git Merge "Remove duplicated code in volume manager and base driver" --- 8893ad83bed3c0ae0cee422e28af109a5a87a524 diff --cc cinder/tests/unit/test_volume.py index 6cf65569b,b82952ea1..2b9708175 --- a/cinder/tests/unit/test_volume.py +++ b/cinder/tests/unit/test_volume.py @@@ -6479,111 -6422,118 +6479,43 @@@ class GenericVolumeDriverTestCase(Drive volume_file) self.assertEqual(i, backup_service.restore.call_count) - def test_enable_replication_invalid_state(self): - volume_api = cinder.volume.api.API() - ctxt = context.get_admin_context() - volume = tests_utils.create_volume(ctxt, - size=1, - host=CONF.host, - replication_status='enabled') - - self.assertRaises(exception.InvalidVolume, - volume_api.enable_replication, - ctxt, volume) - - def test_enable_replication_invalid_type(self): - volume_api = cinder.volume.api.API() - ctxt = context.get_admin_context() - - volume = tests_utils.create_volume(self.context, - size=1, - host=CONF.host, - replication_status='disabled') - volume['volume_type_id'] = 'dab02f01-b50f-4ed6-8d42-2b5b9680996e' - fake_specs = {} - with mock.patch.object(volume_types, - 'get_volume_type_extra_specs', - return_value = fake_specs): - self.assertRaises(exception.InvalidVolume, - volume_api.enable_replication, - ctxt, - volume) - - def test_enable_replication(self): - volume_api = cinder.volume.api.API() - ctxt = context.get_admin_context() - - volume = tests_utils.create_volume(self.context, - size=1, - host=CONF.host, - replication_status='disabled') - volume['volume_type_id'] = 'dab02f01-b50f-4ed6-8d42-2b5b9680996e' - fake_specs = {'replication_enabled': ' True'} - with mock.patch.object(volume_rpcapi.VolumeAPI, - 'enable_replication') as mock_enable_rep,\ - mock.patch.object(volume_types, - 'get_volume_type_extra_specs', - return_value = fake_specs): - - volume_api.enable_replication(ctxt, volume) - self.assertTrue(mock_enable_rep.called) - - def test_enable_replication_driver_initialized(self): - volume = tests_utils.create_volume(self.context, - size=1, - host=CONF.host, - replication_status='enabling') - # set initialized to False - self.volume.driver._initialized = False - - # start test - self.assertRaises(exception.DriverNotInitialized, - self.volume.enable_replication, - self.context, - volume) - - def test_disable_replication_invalid_state(self): - volume_api = cinder.volume.api.API() - ctxt = context.get_admin_context() - volume = tests_utils.create_volume(ctxt, - size=1, - host=CONF.host, - replication_status='invalid-state') - - self.assertRaises(exception.InvalidVolume, - volume_api.disable_replication, - ctxt, volume) - - def test_disable_replication(self): - volume_api = cinder.volume.api.API() - ctxt = context.get_admin_context() - - volume = tests_utils.create_volume(self.context, - size=1, - host=CONF.host, - replication_status='disabled') - - volume['volume_type_id'] = 'dab02f01-b50f-4ed6-8d42-2b5b9680996e' - fake_specs = {'replication_enabled': ' True'} - with mock.patch.object(volume_rpcapi.VolumeAPI, - 'disable_replication') as mock_disable_rep,\ - mock.patch.object(volume_types, - 'get_volume_type_extra_specs', - return_value = fake_specs): - volume_api.disable_replication(ctxt, volume) - self.assertTrue(mock_disable_rep.called) - - volume['replication_status'] = 'enabled' - volume_api.disable_replication(ctxt, volume) - self.assertTrue(mock_disable_rep.called) - - def test_disable_replication_driver_initialized(self): - volume = tests_utils.create_volume(self.context, - size=1, - host=CONF.host, - replication_status='disabling') - # set initialized to False - self.volume.driver._initialized = False + def test_get_backup_device_available(self): + 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) + (backup_device, is_snapshot) = self.volume.driver.get_backup_device( + self.context, backup_obj) + volume = objects.Volume.get_by_id(self.context, vol.id) + self.assertEqual(volume, backup_device) + self.assertFalse(is_snapshot) + backup_obj = objects.Backup.get_by_id(self.context, backup.id) + self.assertIsNone(backup.temp_volume_id) - # start test - self.assertRaises(exception.DriverNotInitialized, - self.volume.disable_replication, - self.context, - volume) + def test_get_backup_device_in_use(self): + vol = tests_utils.create_volume(self.context, + status='backing-up', + previous_status='in-use') + temp_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) + with mock.patch.object( + self.volume.driver, + '_create_temp_cloned_volume') as mock_create_temp: + mock_create_temp.return_value = temp_vol + (backup_device, is_snapshot) = ( + self.volume.driver.get_backup_device(self.context, + backup_obj)) + self.assertEqual(temp_vol, backup_device) + self.assertFalse(is_snapshot) + backup_obj = objects.Backup.get_by_id(self.context, backup.id) + self.assertEqual(temp_vol.id, backup_obj.temp_volume_id) - @mock.patch.object(utils, 'brick_get_connector_properties') - @mock.patch.object(cinder.volume.driver.VolumeDriver, '_attach_volume') - @mock.patch.object(cinder.volume.driver.VolumeDriver, '_detach_volume') - @mock.patch.object(volutils, 'copy_volume') - @mock.patch.object(volume_rpcapi.VolumeAPI, 'get_capabilities') - def test_copy_volume_data(self, - mock_get_capabilities, - mock_copy, - mock_detach, - mock_attach, - mock_get_connector): - - src_vol = tests_utils.create_volume(self.context, size=1, - host=CONF.host) - dest_vol = tests_utils.create_volume(self.context, size=1, - host=CONF.host) - mock_get_connector.return_value = {} - self.volume.driver._throttle = mock.MagicMock() - - attach_expected = [ - mock.call(self.context, dest_vol, {}, remote=False), - mock.call(self.context, src_vol, {}, remote=False)] - - detach_expected = [ - mock.call(self.context, {'device': {'path': 'bar'}}, - dest_vol, {}, force=False, remote=False), - mock.call(self.context, {'device': {'path': 'foo'}}, - src_vol, {}, force=False, remote=False)] - - attach_volume_returns = [ - ({'device': {'path': 'bar'}}, dest_vol), - ({'device': {'path': 'foo'}}, src_vol), - ] - - # Test case for sparse_copy_volume = False - mock_attach.side_effect = attach_volume_returns - mock_get_capabilities.return_value = {} - self.volume.driver.copy_volume_data(self.context, - src_vol, - dest_vol) - - self.assertEqual(attach_expected, mock_attach.mock_calls) - mock_copy.assert_called_with( - 'foo', 'bar', 1024, '1M', - throttle=self.volume.driver._throttle, - sparse=False) - self.assertEqual(detach_expected, mock_detach.mock_calls) - - # Test case for sparse_copy_volume = True - mock_attach.reset_mock() - mock_detach.reset_mock() - mock_attach.side_effect = attach_volume_returns - mock_get_capabilities.return_value = {'sparse_copy_volume': True} - self.volume.driver.copy_volume_data(self.context, - src_vol, - dest_vol) - - self.assertEqual(attach_expected, mock_attach.mock_calls) - mock_copy.assert_called_with( - 'foo', 'bar', 1024, '1M', - throttle=self.volume.driver._throttle, - sparse=True) - self.assertEqual(detach_expected, mock_detach.mock_calls) - - # cleanup resource - db.volume_destroy(self.context, src_vol['id']) - db.volume_destroy(self.context, dest_vol['id']) - @mock.patch.object(utils, 'brick_get_connector_properties') @mock.patch.object(cinder.volume.manager.VolumeManager, '_attach_volume') @mock.patch.object(cinder.volume.manager.VolumeManager, '_detach_volume')