"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "cinder/tests/unit/volume/drivers/test_rbd.py" between
cinder-17.0.1.tar.gz and cinder-17.1.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 "Victoria" series (latest release).

test_rbd.py  (cinder-17.0.1):test_rbd.py  (cinder-17.1.0)
skipping to change at line 872 skipping to change at line 872
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 = mock.Mock()
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 = mock.Mock()
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 1304 skipping to change at line 1345
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 1364 skipping to change at line 1417
@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
skipping to change at line 1518 skipping to change at line 1576
'"bytes_used":1546440971,"max_avail":%s,"objects":412}},' '"bytes_used":1546440971,"max_avail":%s,"objects":412}},'
'{"name":"volumes","id":3,"stats":{"kb_used":0,"bytes_used":0,' '{"name":"volumes","id":3,"stats":{"kb_used":0,"bytes_used":0,'
'"max_avail":28987613184,"objects":0}}]}\n' % max_avail, ''), '"max_avail":28987613184,"objects":0}}]}\n' % max_avail, ''),
(0, '{"pool_name":"volumes","pool_id":4,"quota_max_objects":0,' (0, '{"pool_name":"volumes","pool_id":4,"quota_max_objects":0,'
'"quota_max_bytes":%s}\n' % quota_max_bytes, ''), '"quota_max_bytes":%s}\n' % quota_max_bytes, ''),
] ]
with mock.patch.object(self.driver.configuration, 'safe_get', with mock.patch.object(self.driver.configuration, 'safe_get',
return_value=dynamic_total): return_value=dynamic_total):
result = self.driver._get_pool_stats() result = self.driver._get_pool_stats()
client.cluster.mon_command.assert_has_calls([ client.cluster.mon_command.assert_has_calls([
mock.call('{"prefix":"df", "format":"json"}', ''), mock.call('{"prefix":"df", "format":"json"}', b''),
mock.call('{"prefix":"osd pool get-quota", "pool": "rbd",' mock.call('{"prefix":"osd pool get-quota", "pool": "rbd",'
' "format":"json"}', ''), ' "format":"json"}', b''),
]) ])
self.assertEqual((free_capacity, total_capacity), result) self.assertEqual((free_capacity, total_capacity), result)
@common_mocks @common_mocks
def test_get_pool_bytes(self): def test_get_pool_bytes(self):
"""Test for mon_commands returning bytes instead of strings.""" """Test for mon_commands returning bytes instead of strings."""
client = self.mock_client.return_value client = self.mock_client.return_value
client.__enter__.return_value = client client.__enter__.return_value = client
client.cluster.mon_command.side_effect = [ client.cluster.mon_command.side_effect = [
(0, b'{"stats":{"total_bytes":64385286144,' (0, b'{"stats":{"total_bytes":64385286144,'
b'"total_used_bytes":3289628672,"total_avail_bytes":61095657472},' b'"total_used_bytes":3289628672,"total_avail_bytes":61095657472},'
b'"pools":[{"name":"rbd","id":2,"stats":{"kb_used":1510197,' b'"pools":[{"name":"rbd","id":2,"stats":{"kb_used":1510197,'
b'"bytes_used":1546440971,"max_avail":2897613184,"objects":412}},' b'"bytes_used":1546440971,"max_avail":2897613184,"objects":412}},'
b'{"name":"volumes","id":3,"stats":{"kb_used":0,"bytes_used":0,' b'{"name":"volumes","id":3,"stats":{"kb_used":0,"bytes_used":0,'
b'"max_avail":28987613184,"objects":0}}]}\n', ''), b'"max_avail":28987613184,"objects":0}}]}\n', ''),
(0, b'{"pool_name":"volumes","pool_id":4,"quota_max_objects":0,' (0, b'{"pool_name":"volumes","pool_id":4,"quota_max_objects":0,'
b'"quota_max_bytes":3221225472}\n', ''), b'"quota_max_bytes":3221225472}\n', ''),
] ]
result = self.driver._get_pool_stats() result = self.driver._get_pool_stats()
client.cluster.mon_command.assert_has_calls([ client.cluster.mon_command.assert_has_calls([
mock.call('{"prefix":"df", "format":"json"}', ''), mock.call('{"prefix":"df", "format":"json"}', b''),
mock.call('{"prefix":"osd pool get-quota", "pool": "rbd",' mock.call('{"prefix":"osd pool get-quota", "pool": "rbd",'
' "format":"json"}', ''), ' "format":"json"}', b''),
]) ])
free_capacity = 1.56 free_capacity = 1.56
total_capacity = 3.0 total_capacity = 3.0
self.assertEqual((free_capacity, total_capacity), result) self.assertEqual((free_capacity, total_capacity), result)
@common_mocks @common_mocks
def test_get_pool_stats_failure(self): def test_get_pool_stats_failure(self):
client = self.mock_client.return_value client = self.mock_client.return_value
client.__enter__.return_value = client client.__enter__.return_value = client
client.cluster.mon_command.return_value = (-1, '', '') client.cluster.mon_command.return_value = (-1, '', '')
skipping to change at line 1631 skipping to change at line 1689
self.driver._set_keyring_attributes() self.driver._set_keyring_attributes()
self.assertIsNone(self.driver.keyring_file) self.assertIsNone(self.driver.keyring_file)
self.assertIsNone(self.driver.keyring_data) self.assertIsNone(self.driver.keyring_data)
def test__set_keyring_attributes_cinderlib(self): def test__set_keyring_attributes_cinderlib(self):
# OpenStack usage doesn't have the rbd_keyring_conf Oslo Config option # OpenStack usage doesn't have the rbd_keyring_conf Oslo Config option
cfg_file = '/etc/ceph/ceph.client.admin.keyring' cfg_file = '/etc/ceph/ceph.client.admin.keyring'
self.driver.configuration.rbd_keyring_conf = cfg_file self.driver.configuration.rbd_keyring_conf = cfg_file
self.driver._set_keyring_attributes() with mock.patch('os.path.isfile', return_value=False):
self.driver._set_keyring_attributes()
self.assertEqual(cfg_file, self.driver.keyring_file) self.assertEqual(cfg_file, self.driver.keyring_file)
self.assertIsNone(self.driver.keyring_data) self.assertIsNone(self.driver.keyring_data)
@mock.patch('os.path.isfile') @mock.patch('os.path.isfile')
@mock.patch.object(driver, 'open') @mock.patch.object(driver, 'open')
def test__set_keyring_attributes_cinderlib_read_file(self, mock_open, def test__set_keyring_attributes_cinderlib_read_file(self, mock_open,
mock_isfile): mock_isfile):
cfg_file = '/etc/ceph/ceph.client.admin.keyring' cfg_file = '/etc/ceph/ceph.client.admin.keyring'
# This is how cinderlib sets the config option # This is how cinderlib sets the config option
setattr(self.driver.configuration, 'rbd_keyring_conf', cfg_file) setattr(self.driver.configuration, 'rbd_keyring_conf', cfg_file)
 End of changes. 10 change blocks. 
8 lines changed or deleted 67 lines changed or added

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