"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "cinder/tests/unit/volume/drivers/test_rbd.py" between
cinder-15.4.1.tar.gz and cinder-15.5.0.tar.gz

About: OpenStack Cinder (Core Service: Block Storage) provides persistent block storage to running instances. Its pluggable driver architecture facilitates the creation and management of block storage devices.
The "Train" series (maintained release).

test_rbd.py  (cinder-15.4.1):test_rbd.py  (cinder-15.5.0)
skipping to change at line 877 skipping to change at line 877
proxy.__enter__.return_value = proxy proxy.__enter__.return_value = proxy
self.driver.create_snapshot(self.snapshot) self.driver.create_snapshot(self.snapshot)
args = [str(self.snapshot.name)] args = [str(self.snapshot.name)]
proxy.create_snap.assert_called_with(*args) proxy.create_snap.assert_called_with(*args)
proxy.protect_snap.assert_called_with(*args) proxy.protect_snap.assert_called_with(*args)
@common_mocks @common_mocks
@mock.patch('cinder.objects.Volume.get_by_id') @mock.patch('cinder.objects.Volume.get_by_id')
@mock.patch.object(driver.RBDDriver, '_resize', mock.Mock())
def test_log_create_vol_from_snap_w_v2_clone_api(self, volume_get_by_id):
volume_get_by_id.return_value = self.volume_a
self.mock_proxy().__enter__().volume.op_features.return_value = 1
self.mock_rbd.RBD_OPERATION_FEATURE_CLONE_PARENT = 1
snapshot = self.snapshot
self.cfg.rbd_flatten_volume_from_snapshot = False
with mock.patch.object(driver, 'LOG') as \
mock_log:
self.driver.create_volume_from_snapshot(self.volume_a, snapshot)
mock_log.info.assert_called_once()
self.assertTrue(self.driver._clone_v2_api_checked)
@common_mocks
@mock.patch('cinder.objects.Volume.get_by_id')
@mock.patch.object(driver.RBDDriver, '_resize', mock.Mock())
def test_log_create_vol_from_snap_without_v2_clone_api(self,
volume_get_by_id):
volume_get_by_id.return_value = self.volume_a
self.mock_proxy().__enter__().volume.op_features.return_value = 0
self.mock_rbd.RBD_OPERATION_FEATURE_CLONE_PARENT = 1
snapshot = self.snapshot
self.cfg.rbd_flatten_volume_from_snapshot = False
with mock.patch.object(driver, 'LOG') as \
mock_log:
self.driver.create_volume_from_snapshot(self.volume_a, snapshot)
mock_log.warning.assert_called_once()
self.assertTrue(self.driver._clone_v2_api_checked)
@common_mocks
@mock.patch('cinder.objects.Volume.get_by_id')
def test_delete_snapshot(self, volume_get_by_id): def test_delete_snapshot(self, volume_get_by_id):
volume_get_by_id.return_value = self.volume_a volume_get_by_id.return_value = self.volume_a
proxy = self.mock_proxy.return_value proxy = self.mock_proxy.return_value
proxy.__enter__.return_value = proxy proxy.__enter__.return_value = proxy
self.driver.delete_snapshot(self.snapshot) self.driver.delete_snapshot(self.snapshot)
proxy.remove_snap.assert_called_with(self.snapshot.name) proxy.remove_snap.assert_called_with(self.snapshot.name)
proxy.unprotect_snap.assert_called_with(self.snapshot.name) proxy.unprotect_snap.assert_called_with(self.snapshot.name)
skipping to change at line 1298 skipping to change at line 1339
def test_uncloneable_bad_format(self): def test_uncloneable_bad_format(self):
with mock.patch.object(self.driver, '_get_fsid') as mock_get_fsid: with mock.patch.object(self.driver, '_get_fsid') as mock_get_fsid:
mock_get_fsid.return_value = 'abc' mock_get_fsid.return_value = 'abc'
location = 'rbd://abc/pool/image/snap' location = 'rbd://abc/pool/image/snap'
formats = ['qcow2', 'vmdk', 'vdi'] formats = ['qcow2', 'vmdk', 'vdi']
for f in formats: for f in formats:
self.assertFalse( self.assertFalse(
self.driver._is_cloneable(location, {'disk_format': f})) self.driver._is_cloneable(location, {'disk_format': f}))
self.assertTrue(mock_get_fsid.called) self.assertTrue(mock_get_fsid.called)
def _copy_image(self): def _copy_image(self, volume_busy=False):
with mock.patch.object(tempfile, 'NamedTemporaryFile'): with mock.patch.object(tempfile, 'NamedTemporaryFile'):
with mock.patch.object(os.path, 'exists') as mock_exists: with mock.patch.object(os.path, 'exists') as mock_exists:
mock_exists.return_value = True mock_exists.return_value = True
with mock.patch.object(image_utils, 'fetch_to_raw'): with mock.patch.object(image_utils, 'fetch_to_raw'):
with mock.patch.object(self.driver, 'delete_volume'): with mock.patch.object(self.driver, 'delete_volume') \
as mock_dv:
with mock.patch.object(self.driver, '_resize'): with mock.patch.object(self.driver, '_resize'):
mock_image_service = mock.MagicMock() mock_image_service = mock.MagicMock()
args = [None, self.volume_a, args = [None, self.volume_a,
mock_image_service, None] mock_image_service, None]
self.driver.copy_image_to_volume(*args) if volume_busy:
mock_dv.side_effect = (
exception.VolumeIsBusy("doh"))
self.assertRaises(
exception.VolumeIsBusy,
self.driver.copy_image_to_volume,
*args)
self.assertEqual(
self.cfg.rados_connection_retries,
mock_dv.call_count)
else:
self.driver.copy_image_to_volume(*args)
@mock.patch('cinder.volume.drivers.rbd.fileutils.delete_if_exists') @mock.patch('cinder.volume.drivers.rbd.fileutils.delete_if_exists')
@mock.patch('cinder.volume.volume_utils.check_encryption_provider', @mock.patch('cinder.volume.volume_utils.check_encryption_provider',
return_value={'encryption_key_id': fake.ENCRYPTION_KEY_ID}) return_value={'encryption_key_id': fake.ENCRYPTION_KEY_ID})
@mock.patch('cinder.image.image_utils.convert_image') @mock.patch('cinder.image.image_utils.convert_image')
def _copy_image_encrypted(self, mock_convert, mock_encrypt_key, def _copy_image_encrypted(self, mock_convert, mock_encrypt_key,
mock_temp_delete): mock_temp_delete):
key_mgr = fake_keymgr.fake_api() key_mgr = fake_keymgr.fake_api()
self.mock_object(castellan.key_manager, 'API', return_value=key_mgr) self.mock_object(castellan.key_manager, 'API', return_value=key_mgr)
key_id = key_mgr.store(self.context, KeyObject()) key_id = key_mgr.store(self.context, KeyObject())
skipping to change at line 1358 skipping to change at line 1411
@common_mocks @common_mocks
def test_copy_image_volume_tmp(self): def test_copy_image_volume_tmp(self):
self.cfg.image_conversion_dir = '/var/run/cinder/tmp' self.cfg.image_conversion_dir = '/var/run/cinder/tmp'
self._copy_image() self._copy_image()
@common_mocks @common_mocks
def test_copy_image_volume_tmp_encrypted(self): def test_copy_image_volume_tmp_encrypted(self):
self.cfg.image_conversion_dir = '/var/run/cinder/tmp' self.cfg.image_conversion_dir = '/var/run/cinder/tmp'
self._copy_image_encrypted() self._copy_image_encrypted()
@common_mocks
def test_copy_image_busy_volume(self):
self.cfg.image_conversion_dir = '/var/run/cinder/tmp'
self._copy_image(volume_busy=True)
@ddt.data(True, False) @ddt.data(True, False)
@common_mocks @common_mocks
@mock.patch('cinder.volume.drivers.rbd.RBDDriver._get_usage_info') @mock.patch('cinder.volume.drivers.rbd.RBDDriver._get_usage_info')
@mock.patch('cinder.volume.drivers.rbd.RBDDriver._get_pool_stats') @mock.patch('cinder.volume.drivers.rbd.RBDDriver._get_pool_stats')
def test_update_volume_stats(self, replication_enabled, stats_mock, def test_update_volume_stats(self, replication_enabled, stats_mock,
usage_mock): usage_mock):
stats_mock.return_value = (mock.sentinel.free_capacity_gb, stats_mock.return_value = (mock.sentinel.free_capacity_gb,
mock.sentinel.total_capacity_gb) mock.sentinel.total_capacity_gb)
usage_mock.return_value = mock.sentinel.provisioned_capacity_gb usage_mock.return_value = mock.sentinel.provisioned_capacity_gb
 End of changes. 5 change blocks. 
3 lines changed or deleted 61 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)