# Copyright 2012 Josh Durgin
+# Copyright 2013 Canonical Ltd.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# under the License.
-import contextlib
import mock
-import mox
import os
import tempfile
from cinder.openstack.common import log as logging
from cinder.openstack.common import timeutils
from cinder import test
-from cinder.tests.backup.fake_rados import mock_rados
-from cinder.tests.backup.fake_rados import mock_rbd
from cinder.tests.image import fake as fake_image
from cinder.tests.test_volume import DriverTestCase
from cinder import units
"""
-class FakeImageService:
- def download(self, context, image_id, path):
- pass
-
-
class TestUtil(test.TestCase):
def test_ascii_str(self):
self.assertIsNone(driver.ascii_str(None))
def setUp(self):
super(RBDTestCase, self).setUp()
- def fake_execute(*args, **kwargs):
- return '', ''
- self.configuration = mox.MockObject(conf.Configuration)
- self.configuration.volume_tmp_dir = None
- self.configuration.rbd_pool = 'rbd'
- self.configuration.rbd_ceph_conf = None
- self.configuration.rbd_secret_uuid = None
- self.configuration.rbd_user = None
- self.configuration.append_config_values(mox.IgnoreArg())
- self.configuration.volume_dd_blocksize = '1M'
-
- self.rados = self.mox.CreateMockAnything()
- self.rbd = self.mox.CreateMockAnything()
- self.driver = driver.RBDDriver(execute=fake_execute,
- configuration=self.configuration,
+ self.cfg = mock.Mock(spec=conf.Configuration)
+ self.cfg.volume_tmp_dir = None
+ self.cfg.rbd_pool = 'rbd'
+ self.cfg.rbd_ceph_conf = None
+ self.cfg.rbd_secret_uuid = None
+ self.cfg.rbd_user = None
+ self.cfg.volume_dd_blocksize = '1M'
+
+ # set some top level mocks for these common modules and tests can then
+ # set method/attributes as required.
+ self.rados = mock.Mock()
+ self.rbd = mock.Mock()
+ self.rbd.RBD = mock.Mock
+ self.rbd.Image = mock.Mock
+ self.rbd.ImageSnapshot = mock.Mock
+
+ mock_exec = mock.Mock()
+ mock_exec.return_value = ('', '')
+
+ self.driver = driver.RBDDriver(execute=mock_exec,
+ configuration=self.cfg,
rados=self.rados,
rbd=self.rbd)
self.driver.set_initialized()
+ self.volume_name = u'volume-00000001'
+ self.snapshot_name = u'snapshot-00000001'
+ self.volume_size = 1
+ self.volume = dict(name=self.volume_name, size=self.volume_size)
+ self.snapshot = dict(volume_name=self.volume_name,
+ name=self.snapshot_name)
+
def tearDown(self):
super(RBDTestCase, self).tearDown()
- def test_create_volume(self):
- name = u'volume-00000001'
- size = 1
- volume = dict(name=name, size=size)
- mock_client = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RADOSClient')
-
- driver.RADOSClient(self.driver).AndReturn(mock_client)
- mock_client.__enter__().AndReturn(mock_client)
- self.rbd.RBD_FEATURE_LAYERING = 1
- _mock_rbd = self.mox.CreateMockAnything()
- self.rbd.RBD().AndReturn(_mock_rbd)
- _mock_rbd.create(mox.IgnoreArg(), str(name), size * units.GiB,
- old_format=False,
- features=self.rbd.RBD_FEATURE_LAYERING)
- mock_client.__exit__(None, None, None).AndReturn(None)
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_create_volume(self, mock_client):
+ client = mock_client.return_value
+ client.__enter__.return_value = client
+
+ self.driver._supports_layering = mock.Mock()
+ self.driver._supports_layering.return_value = True
+ self.rbd.RBD.create = mock.Mock()
+
+ self.driver.create_volume(self.volume)
+
+ args = [client.ioctx, str(self.volume_name),
+ self.volume_size * units.GiB]
+ kwargs = {'old_format': False,
+ 'features': self.rbd.RBD_FEATURE_LAYERING}
+
+ self.rbd.RBD.create.assert_called_once()
+ client.__enter__.assert_called_once()
+ client.__exit__.assert_called_once()
+ self.driver._supports_layering.assert_called_once()
+ self.rbd.RBD.create.assert_called_once_with(*args, **kwargs)
+
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_create_volume_no_layering(self, mock_client):
+ client = mock_client.return_value
+ client.__enter__.return_value = client
+
+ self.driver._supports_layering = mock.Mock()
+ self.driver._supports_layering.return_value = False
+ self.rbd.RBD.create = mock.Mock()
+
+ self.driver.create_volume(self.volume)
+
+ args = [client.ioctx, str(self.volume_name),
+ self.volume_size * units.GiB]
+ kwargs = {'old_format': True,
+ 'features': 0}
+
+ self.rbd.RBD.create.assert_called_once()
+ client.__enter__.assert_called_once()
+ client.__exit__.assert_called_once()
+ self.driver._supports_layering.assert_called_once()
+ self.rbd.RBD.create.assert_called_once_with(*args, **kwargs)
+
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_delete_volume(self, mock_client):
+ client = mock_client.return_value
+
+ self.driver.rbd.Image.list_snaps = mock.Mock()
+ self.driver.rbd.Image.list_snaps.return_value = []
+ self.driver.rbd.Image.close = mock.Mock()
+ self.driver.rbd.Image.remove = mock.Mock()
+ self.driver.rbd.Image.unprotect_snap = mock.Mock()
+
+ self.driver._get_clone_info = mock.Mock()
+ self.driver._get_clone_info.return_value = (None, None, None)
+ self.driver._delete_backup_snaps = mock.Mock()
+
+ self.driver.delete_volume(self.volume)
+
+ self.driver._get_clone_info.assert_called_once()
+ self.driver.rbd.Image.list_snaps.assert_called_once()
+ client.__enter__.assert_called_once()
+ client.__exit__.assert_called_once()
+ self.driver._delete_backup_snaps.assert_called_once()
+ self.assertFalse(self.driver.rbd.Image.unprotect_snap.called)
+ self.driver.rbd.RBD.remove.assert_called_once()
- self.mox.ReplayAll()
+ @mock.patch('cinder.volume.drivers.rbd.rbd')
+ def test_delete_volume_not_found(self, mock_rbd):
+ mock_rbd.RBD = mock.Mock
+ mock_rbd.ImageNotFound = Exception
+ mock_rbd.Image.side_effect = mock_rbd.ImageNotFound
- self.driver.create_volume(volume)
+ self.driver.rbd = mock_rbd
- @mock.patch('cinder.volume.drivers.rbd.rados')
- @mock.patch('cinder.volume.drivers.rbd.rbd')
- def test_delete_volume(self, _mock_rbd, _mock_rados):
- name = u'volume-00000001'
- volume = dict(name=name)
-
- _mock_rbd.Image = mock_rbd.Image
- _mock_rbd.Image.list_snaps = mock.Mock()
- _mock_rbd.Image.list_snaps.return_value = []
- _mock_rbd.Image.unprotect_snap = mock.Mock()
-
- _mock_rbd.RBD = mock_rbd.RBD
- _mock_rbd.RBD.remove = mock.Mock()
-
- self.driver.rbd = _mock_rbd
- self.driver.rados = _mock_rados
-
- mpo = mock.patch.object
- with mpo(driver, 'RADOSClient') as mock_rados_client:
- with mpo(self.driver, '_get_clone_info') as mock_get_clone_info:
- mock_get_clone_info.return_value = (None, None, None)
- with mpo(self.driver,
- '_delete_backup_snaps') as mock_del_backup_snaps:
- self.driver.delete_volume(volume)
-
- self.assertTrue(mock_get_clone_info.called)
- self.assertTrue(_mock_rbd.Image.list_snaps.called)
- self.assertTrue(mock_rados_client.called)
- self.assertTrue(mock_del_backup_snaps.called)
- self.assertFalse(mock_rbd.Image.unprotect_snap.called)
- self.assertTrue(_mock_rbd.RBD.remove.called)
+ with mock.patch.object(driver, 'RADOSClient'):
+ self.assertIsNone(self.driver.delete_volume(self.volume))
+ mock_rbd.Image.assert_called_once()
- @mock.patch('cinder.volume.drivers.rbd.rbd')
- def test_delete_volume_not_found(self, _mock_rbd):
- name = u'volume-00000001'
- volume = dict(name=name)
+ def test_delete_busy_volume(self):
+ self.rbd.Image.close = mock.Mock()
+ self.rbd.Image.list_snaps = mock.Mock()
+ self.rbd.Image.list_snaps.return_value = []
+ self.rbd.Image.unprotect_snap = mock.Mock()
- class MyMockException(Exception):
- pass
+ self.rbd.ImageBusy = Exception
+ self.rbd.RBD.remove = mock.Mock()
+ self.rbd.RBD.remove.side_effect = self.rbd.ImageBusy
- _mock_rbd.RBD = mock_rbd.RBD
- _mock_rbd.ImageNotFound = MyMockException
- _mock_rbd.Image.side_effect = _mock_rbd.ImageNotFound
+ self.driver._get_clone_info = mock.Mock()
+ self.driver._get_clone_info.return_value = (None, None, None)
+ self.driver._delete_backup_snaps = mock.Mock()
- mpo = mock.patch.object
- with mpo(self.driver, 'rbd', _mock_rbd):
- with mpo(driver, 'RADOSClient'):
- self.assertIsNone(self.driver.delete_volume(volume))
- _mock_rbd.Image.assert_called_once()
+ with mock.patch.object(driver, 'RADOSClient') as mock_rados_client:
+ self.assertRaises(exception.VolumeIsBusy,
+ self.driver.delete_volume, self.volume)
- @mock.patch('cinder.volume.drivers.rbd.rados')
- @mock.patch('cinder.volume.drivers.rbd.rbd')
- def test_delete_busy_volume(self, _mock_rbd, _mock_rados):
- name = u'volume-00000001'
- volume = dict(name=name)
-
- _mock_rbd.Image = mock_rbd.Image
- _mock_rbd.Image.list_snaps = mock.Mock()
- _mock_rbd.Image.list_snaps.return_value = []
- _mock_rbd.Image.unprotect_snap = mock.Mock()
-
- class MyMockException(Exception):
- pass
-
- _mock_rbd.RBD = mock_rbd.RBD
- _mock_rbd.ImageBusy = MyMockException
- _mock_rbd.RBD.remove = mock.Mock()
- _mock_rbd.RBD.remove.side_effect = _mock_rbd.ImageBusy
-
- self.driver.rbd = _mock_rbd
- self.driver.rados = _mock_rados
-
- mpo = mock.patch.object
- with mpo(driver, 'RADOSClient') as mock_rados_client:
- with mpo(self.driver, '_get_clone_info') as mock_get_clone_info:
- mock_get_clone_info.return_value = (None, None, None)
- with mpo(self.driver,
- '_delete_backup_snaps') as mock_del_backup_snaps:
-
- self.assertRaises(exception.VolumeIsBusy,
- self.driver.delete_volume,
- volume)
-
- self.assertTrue(mock_get_clone_info.called)
- self.assertTrue(_mock_rbd.Image.list_snaps.called)
- self.assertTrue(mock_rados_client.called)
- self.assertTrue(mock_del_backup_snaps.called)
- self.assertFalse(mock_rbd.Image.unprotect_snap.called)
- self.assertTrue(_mock_rbd.RBD.remove.called)
-
- def test_create_snapshot(self):
- vol_name = u'volume-00000001'
- snap_name = u'snapshot-name'
- snapshot = dict(volume_name=vol_name, name=snap_name)
- mock_proxy = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RBDVolumeProxy')
-
- driver.RBDVolumeProxy(self.driver, vol_name) \
- .AndReturn(mock_proxy)
- mock_proxy.__enter__().AndReturn(mock_proxy)
- mock_proxy.create_snap(str(snap_name))
- self.rbd.RBD_FEATURE_LAYERING = 1
- mock_proxy.protect_snap(str(snap_name))
- mock_proxy.__exit__(None, None, None).AndReturn(None)
+ self.driver._get_clone_info.assert_called_once()
+ self.rbd.Image.list_snaps.assert_called_once()
+ mock_rados_client.assert_called_once()
+ self.driver._delete_backup_snaps.assert_called_once()
+ self.assertFalse(self.rbd.Image.unprotect_snap.called)
+ self.rbd.RBD.remove.assert_called_once()
- self.mox.ReplayAll()
+ @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
+ def test_create_snapshot(self, mock_proxy):
+ proxy = mock_proxy.return_value
+ proxy.__enter__.return_value = proxy
- self.driver.create_snapshot(snapshot)
+ self.driver.create_snapshot(self.snapshot)
- def test_delete_snapshot(self):
- vol_name = u'volume-00000001'
- snap_name = u'snapshot-name'
- snapshot = dict(volume_name=vol_name, name=snap_name)
- mock_proxy = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RBDVolumeProxy')
+ args = [str(self.snapshot_name)]
+ proxy.create_snap.assert_called_with(*args)
+ proxy.protect_snap.assert_called_with(*args)
- driver.RBDVolumeProxy(self.driver, vol_name) \
- .AndReturn(mock_proxy)
- mock_proxy.__enter__().AndReturn(mock_proxy)
- self.rbd.RBD_FEATURE_LAYERING = 1
- mock_proxy.unprotect_snap(str(snap_name))
- mock_proxy.remove_snap(str(snap_name))
- mock_proxy.__exit__(None, None, None).AndReturn(None)
+ @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
+ def test_delete_snapshot(self, mock_proxy):
+ proxy = mock_proxy.return_value
+ proxy.__enter__.return_value = proxy
- self.mox.ReplayAll()
+ self.driver.delete_snapshot(self.snapshot)
- self.driver.delete_snapshot(snapshot)
+ args = [str(self.snapshot_name)]
+ proxy.remove_snap.assert_called_with(*args)
+ proxy.unprotect_snap.assert_called_with(*args)
- def test_create_cloned_volume(self):
- src_name = u'volume-00000001'
- dst_name = u'volume-00000002'
+ def test_get_clone_info(self):
+
+ volume = self.rbd.Image()
+ volume.set_snap = mock.Mock()
+ volume.parent_info = mock.Mock()
+ parent_info = ('a', 'b', '%s.clone_snap' % (self.volume_name))
+ volume.parent_info.return_value = parent_info
+
+ info = self.driver._get_clone_info(volume, self.volume_name)
+
+ self.assertEqual(info, parent_info)
+
+ self.assertFalse(volume.set_snap.called)
+ volume.parent_info.assert_called_once()
+
+ def test_get_clone_info_w_snap(self):
+
+ volume = self.rbd.Image()
+ volume.set_snap = mock.Mock()
+ volume.parent_info = mock.Mock()
+ parent_info = ('a', 'b', '%s.clone_snap' % (self.volume_name))
+ volume.parent_info.return_value = parent_info
+
+ snapshot = self.rbd.ImageSnapshot()
+
+ info = self.driver._get_clone_info(volume, self.volume_name,
+ snap=snapshot)
+
+ self.assertEqual(info, parent_info)
+
+ volume.set_snap.assert_called_once()
+ self.assertEqual(volume.set_snap.call_count, 2)
+ volume.parent_info.assert_called_once()
+
+ def test_get_clone_info_w_exception(self):
+
+ self.rbd.ImageNotFound = Exception
+
+ volume = self.rbd.Image()
+ volume.set_snap = mock.Mock()
+ volume.parent_info = mock.Mock()
+ volume.parent_info.side_effect = self.rbd.ImageNotFound
+
+ snapshot = self.rbd.ImageSnapshot()
- # Setup librbd stubs
- self.stubs.Set(self.driver, 'rados', mock_rados)
- self.stubs.Set(self.driver, 'rbd', mock_rbd)
+ info = self.driver._get_clone_info(volume, self.volume_name,
+ snap=snapshot)
- self.driver.rbd.RBD_FEATURE_LAYERING = 1
+ self.assertEqual(info, (None, None, None))
- class mock_client(object):
- def __init__(self, *args, **kwargs):
- self.ioctx = None
+ volume.set_snap.assert_called_once()
+ self.assertEqual(volume.set_snap.call_count, 2)
+ volume.parent_info.assert_called_once()
- def __enter__(self, *args, **kwargs):
- return self
+ def test_get_clone_info_deleted_volume(self):
- def __exit__(self, type_, value, traceback):
- pass
+ volume = self.rbd.Image()
+ volume.set_snap = mock.Mock()
+ volume.parent_info = mock.Mock()
+ parent_info = ('a', 'b', '%s.clone_snap' % (self.volume_name))
+ volume.parent_info.return_value = parent_info
- self.stubs.Set(driver, 'RADOSClient', mock_client)
+ info = self.driver._get_clone_info(volume,
+ "%s.deleted" % (self.volume_name))
- def mock_clone(*args, **kwargs):
- pass
+ self.assertEqual(info, parent_info)
- self.stubs.Set(self.driver.rbd.RBD, 'clone', mock_clone)
- self.stubs.Set(self.driver.rbd.Image, 'list_snaps',
- lambda *args: [{'name': 'snap1'}, {'name': 'snap2'}])
- self.stubs.Set(self.driver.rbd.Image, 'parent_info',
- lambda *args: (None, None, None))
- self.stubs.Set(self.driver.rbd.Image, 'protect_snap',
- lambda *args: None)
+ self.assertFalse(volume.set_snap.called)
+ volume.parent_info.assert_called_once()
+
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_create_cloned_volume(self, mock_client):
+ src_name = u'volume-00000001'
+ dst_name = u'volume-00000002'
+
+ self.cfg.rbd_max_clone_depth = 2
+ self.rbd.RBD.clone = mock.Mock()
+ self.driver._get_clone_depth = mock.Mock()
+ # Try with no flatten required
+ self.driver._get_clone_depth.return_value = 1
+
+ self.rbd.Image.create_snap = mock.Mock()
+ self.rbd.Image.protect_snap = mock.Mock()
+ self.rbd.Image.close = mock.Mock()
self.driver.create_cloned_volume(dict(name=dst_name),
dict(name=src_name))
+ self.rbd.Image.create_snap.assert_called_once()
+ self.rbd.Image.protect_snap.assert_called_once()
+ self.rbd.RBD.clone.assert_called_once()
+ self.rbd.Image.close.assert_called_once()
+
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_create_cloned_volume_w_flatten(self, mock_client):
+ src_name = u'volume-00000001'
+ dst_name = u'volume-00000002'
+
+ self.cfg.rbd_max_clone_depth = 1
+ self.rbd.RBD.Error = Exception
+ self.rbd.RBD.clone = mock.Mock()
+ self.rbd.RBD.clone.side_effect = self.rbd.RBD.Error
+ self.driver._get_clone_depth = mock.Mock()
+ # Try with no flatten required
+ self.driver._get_clone_depth.return_value = 1
+
+ self.rbd.Image.create_snap = mock.Mock()
+ self.rbd.Image.protect_snap = mock.Mock()
+ self.rbd.Image.unprotect_snap = mock.Mock()
+ self.rbd.Image.remove_snap = mock.Mock()
+ self.rbd.Image.close = mock.Mock()
+
+ self.assertRaises(self.rbd.RBD.Error, self.driver.create_cloned_volume,
+ dict(name=dst_name), dict(name=src_name))
+
+ self.rbd.Image.create_snap.assert_called_once()
+ self.rbd.Image.protect_snap.assert_called_once()
+ self.rbd.RBD.clone.assert_called_once()
+ self.rbd.Image.unprotect_snap.assert_called_once()
+ self.rbd.Image.remove_snap.assert_called_once()
+ self.rbd.Image.close.assert_called_once()
+
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_create_cloned_volume_w_clone_exception(self, mock_client):
+ src_name = u'volume-00000001'
+ dst_name = u'volume-00000002'
+
+ self.cfg.rbd_max_clone_depth = 2
+ self.rbd.RBD.Error = Exception
+ self.rbd.RBD.clone = mock.Mock()
+ self.rbd.RBD.clone.side_effect = self.rbd.RBD.Error
+ self.driver._get_clone_depth = mock.Mock()
+ # Try with no flatten required
+ self.driver._get_clone_depth.return_value = 1
+
+ self.rbd.Image.create_snap = mock.Mock()
+ self.rbd.Image.protect_snap = mock.Mock()
+ self.rbd.Image.unprotect_snap = mock.Mock()
+ self.rbd.Image.remove_snap = mock.Mock()
+ self.rbd.Image.close = mock.Mock()
+
+ self.assertRaises(self.rbd.RBD.Error, self.driver.create_cloned_volume,
+ dict(name=dst_name), dict(name=src_name))
+
+ self.rbd.Image.create_snap.assert_called_once()
+ self.rbd.Image.protect_snap.assert_called_once()
+ self.rbd.RBD.clone.assert_called_once()
+ self.rbd.Image.unprotect_snap.assert_called_once()
+ self.rbd.Image.remove_snap.assert_called_once()
+ self.rbd.Image.close.assert_called_once()
+
def test_good_locations(self):
locations = ['rbd://fsid/pool/image/snap',
'rbd://%2F/%2F/%2F/%2F', ]
self.assertFalse(
self.driver._is_cloneable(loc, {'disk_format': 'raw'}))
- def test_cloneable(self):
- self.stubs.Set(self.driver, '_get_fsid', lambda: 'abc')
+ @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
+ def test_cloneable(self, mock_proxy):
+ self.driver._get_fsid = mock.Mock()
+ self.driver._get_fsid.return_value = 'abc'
location = 'rbd://abc/pool/image/snap'
- mock_proxy = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RBDVolumeProxy')
+ info = {'disk_format': 'raw'}
+ self.assertTrue(self.driver._is_cloneable(location, info))
- driver.RBDVolumeProxy(self.driver, 'image',
- pool='pool',
- snapshot='snap',
- read_only=True).AndReturn(mock_proxy)
- mock_proxy.__enter__().AndReturn(mock_proxy)
- mock_proxy.__exit__(None, None, None).AndReturn(None)
-
- self.mox.ReplayAll()
-
- self.assertTrue(
- self.driver._is_cloneable(location, {'disk_format': 'raw'}))
-
- def test_uncloneable_different_fsid(self):
- self.stubs.Set(self.driver, '_get_fsid', lambda: 'abc')
+ @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
+ def test_uncloneable_different_fsid(self, mock_proxy):
+ self.driver._get_fsid = mock.Mock()
+ self.driver._get_fsid.return_value = 'abc'
location = 'rbd://def/pool/image/snap'
self.assertFalse(
self.driver._is_cloneable(location, {'disk_format': 'raw'}))
- def test_uncloneable_unreadable(self):
- self.stubs.Set(self.driver, '_get_fsid', lambda: 'abc')
+ @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
+ def test_uncloneable_unreadable(self, mock_proxy):
+ self.driver._get_fsid = mock.Mock()
+ self.driver._get_fsid.return_value = 'abc'
location = 'rbd://abc/pool/image/snap'
- self.stubs.Set(self.rbd, 'Error', test.TestingException)
- self.mox.StubOutWithMock(driver, 'RBDVolumeProxy')
- driver.RBDVolumeProxy(self.driver, 'image',
- pool='pool',
- snapshot='snap',
- read_only=True).AndRaise(test.TestingException)
+ self.rbd.Error = Exception
+ mock_proxy.side_effect = self.rbd.Error
- self.mox.ReplayAll()
-
- self.assertFalse(
- self.driver._is_cloneable(location, {'disk_format': 'raw'}))
+ args = [location, {'disk_format': 'raw'}]
+ self.assertFalse(self.driver._is_cloneable(*args))
+ mock_proxy.assert_called_once()
def test_uncloneable_bad_format(self):
- self.stubs.Set(self.driver, '_get_fsid', lambda: 'abc')
+ self.driver._get_fsid = mock.Mock()
+ self.driver._get_fsid.return_value = 'abc'
location = 'rbd://abc/pool/image/snap'
formats = ['qcow2', 'vmdk', 'vdi']
for f in formats:
self.driver._is_cloneable(location, {'disk_format': f}))
def _copy_image(self):
- @contextlib.contextmanager
- def fake_temp_file(dir):
- class FakeTmp:
- def __init__(self, name):
- self.name = name
- yield FakeTmp('test')
-
- def fake_fetch_to_raw(ctx, image_service, image_id, path, blocksize,
- size=None):
- pass
-
- self.stubs.Set(tempfile, 'NamedTemporaryFile', fake_temp_file)
- self.stubs.Set(os.path, 'exists', lambda x: True)
- self.stubs.Set(image_utils, 'fetch_to_raw', fake_fetch_to_raw)
- self.stubs.Set(self.driver, 'delete_volume', lambda x: None)
- self.stubs.Set(self.driver, '_resize', lambda x: None)
- self.driver.copy_image_to_volume(None, {'name': 'test',
- 'size': 1},
- FakeImageService(), None)
+ with mock.patch.object(tempfile, 'NamedTemporaryFile'):
+ with mock.patch.object(os.path, 'exists') as mock_exists:
+ mock_exists.return_value = True
+ with mock.patch.object(image_utils, 'fetch_to_raw'):
+ with mock.patch.object(self.driver, 'delete_volume'):
+ with mock.patch.object(self.driver, '_resize'):
+ mock_image_service = mock.MagicMock()
+ args = [None, {'name': 'test', 'size': 1},
+ mock_image_service, None]
+ self.driver.copy_image_to_volume(*args)
def test_copy_image_no_volume_tmp(self):
- self.configuration.volume_tmp_dir = None
+ self.cfg.volume_tmp_dir = None
self._copy_image()
def test_copy_image_volume_tmp(self):
- self.configuration.volume_tmp_dir = '/var/run/cinder/tmp'
+ self.cfg.volume_tmp_dir = '/var/run/cinder/tmp'
self._copy_image()
- def test_update_volume_stats(self):
- self.stubs.Set(self.driver.configuration, 'safe_get', lambda x: 'RBD')
- mock_client = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RADOSClient')
-
- driver.RADOSClient(self.driver).AndReturn(mock_client)
- mock_client.__enter__().AndReturn(mock_client)
- self.mox.StubOutWithMock(mock_client, 'cluster')
- mock_client.cluster.get_cluster_stats().AndReturn(dict(
- kb=1234567890,
- kb_used=4567890,
- kb_avail=1000000000,
- num_objects=4683))
- mock_client.__exit__(None, None, None).AndReturn(None)
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_update_volume_stats(self, mock_client):
+ client = mock_client.return_value
+ client.__enter__.return_value = client
- self.mox.ReplayAll()
+ client.cluster = mock.Mock()
+ client.cluster.get_cluster_stats = mock.Mock()
+ client.cluster.get_cluster_stats.return_value = {'kb': 1024 ** 3,
+ 'kb_avail': 1024 ** 2}
+
+ self.driver.configuration.safe_get = mock.Mock()
+ self.driver.configuration.safe_get.return_value = 'RBD'
expected = dict(
volume_backend_name='RBD',
vendor_name='Open Source',
driver_version=self.driver.VERSION,
storage_protocol='ceph',
- total_capacity_gb=1177,
- free_capacity_gb=953,
+ total_capacity_gb=1024,
+ free_capacity_gb=1,
reserved_percentage=0)
+
actual = self.driver.get_volume_stats(True)
+ client.cluster.get_cluster_stats.assert_called_once()
self.assertDictMatch(expected, actual)
- def test_update_volume_stats_error(self):
- self.stubs.Set(self.driver.configuration, 'safe_get', lambda x: 'RBD')
- mock_client = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RADOSClient')
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_update_volume_stats_error(self, mock_client):
+ client = mock_client.return_value
+ client.__enter__.return_value = client
- driver.RADOSClient(self.driver).AndReturn(mock_client)
- mock_client.__enter__().AndReturn(mock_client)
- self.mox.StubOutWithMock(mock_client, 'cluster')
- self.stubs.Set(self.rados, 'Error', test.TestingException)
- mock_client.cluster.get_cluster_stats().AndRaise(test.TestingException)
- mock_client.__exit__(test.TestingException,
- mox.IgnoreArg(), mox.IgnoreArg()).AndReturn(None)
+ client.cluster = mock.Mock()
+ client.cluster.get_cluster_stats = mock.Mock()
+ client.cluster.get_cluster_stats.side_effect = Exception
- self.mox.ReplayAll()
+ self.driver.configuration.safe_get = mock.Mock()
+ self.driver.configuration.safe_get.return_value = 'RBD'
+
+ self.rados.Error = Exception
+
+ expected = dict(volume_backend_name='RBD',
+ vendor_name='Open Source',
+ driver_version=self.driver.VERSION,
+ storage_protocol='ceph',
+ total_capacity_gb='unknown',
+ free_capacity_gb='unknown',
+ reserved_percentage=0)
- expected = dict(
- volume_backend_name='RBD',
- vendor_name='Open Source',
- driver_version=self.driver.VERSION,
- storage_protocol='ceph',
- total_capacity_gb='unknown',
- free_capacity_gb='unknown',
- reserved_percentage=0)
actual = self.driver.get_volume_stats(True)
+ client.cluster.get_cluster_stats.assert_called_once()
self.assertDictMatch(expected, actual)
def test_get_mon_addrs(self):
- self.stubs.Set(self.driver, '_execute',
- lambda *a: (CEPH_MON_DUMP, ''))
+ self.driver._execute = mock.Mock()
+ self.driver._execute.return_value = (CEPH_MON_DUMP, '')
+
hosts = ['::1', '::1', '::1', '127.0.0.1', 'example.com']
ports = ['6789', '6790', '6791', '6792', '6791']
self.assertEqual((hosts, ports), self.driver._get_mon_addrs())
def test_initialize_connection(self):
- name = 'volume-00000001'
hosts = ['::1', '::1', '::1', '127.0.0.1', 'example.com']
ports = ['6789', '6790', '6791', '6792', '6791']
- self.stubs.Set(self.driver, '_get_mon_addrs', lambda: (hosts, ports))
+
+ self.driver._get_mon_addrs = mock.Mock()
+ self.driver._get_mon_addrs.return_value = (hosts, ports)
+
expected = {
'driver_volume_type': 'rbd',
'data': {
- 'name': '%s/%s' % (self.configuration.rbd_pool,
- name),
+ 'name': '%s/%s' % (self.cfg.rbd_pool,
+ self.volume_name),
'hosts': hosts,
'ports': ports,
'auth_enabled': False,
'secret_type': 'ceph',
'secret_uuid': None, }
}
- actual = self.driver.initialize_connection(dict(name=name), None)
+ actual = self.driver.initialize_connection(dict(name=self.volume_name),
+ None)
self.assertDictMatch(expected, actual)
- def test_clone(self):
- name = u'volume-00000001'
- volume = dict(name=name)
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_clone(self, mock_client):
src_pool = u'images'
src_image = u'image-name'
src_snap = u'snapshot-name'
- mock_src_client = self.mox.CreateMockAnything()
- mock_dst_client = self.mox.CreateMockAnything()
- mock_rbd = self.mox.CreateMockAnything()
- self.mox.StubOutWithMock(driver, 'RADOSClient')
-
- driver.RADOSClient(self.driver, src_pool).AndReturn(mock_src_client)
- mock_src_client.__enter__().AndReturn(mock_src_client)
- driver.RADOSClient(self.driver).AndReturn(mock_dst_client)
- mock_dst_client.__enter__().AndReturn(mock_dst_client)
- self.rbd.RBD_FEATURE_LAYERING = 1
- self.rbd.RBD().AndReturn(mock_rbd)
- mock_rbd.clone(mox.IgnoreArg(),
- str(src_image),
- str(src_snap),
- mox.IgnoreArg(),
- str(name),
- features=self.rbd.RBD_FEATURE_LAYERING)
- mock_dst_client.__exit__(None, None, None).AndReturn(None)
- mock_src_client.__exit__(None, None, None).AndReturn(None)
- self.mox.ReplayAll()
+ client_stack = []
+
+ def mock__enter__(inst):
+ def _inner():
+ client_stack.append(inst)
+ return inst
+ return _inner
- self.driver._clone(volume, src_pool, src_image, src_snap)
+ client = mock_client.return_value
+ # capture both rados client used to perform the clone
+ client.__enter__.side_effect = mock__enter__(client)
+
+ self.rbd.RBD.clone = mock.Mock()
+
+ self.driver._clone(self.volume, src_pool, src_image, src_snap)
+
+ args = [client_stack[0].ioctx, str(src_image), str(src_snap),
+ client_stack[1].ioctx, str(self.volume_name)]
+ kwargs = {'features': self.rbd.RBD_FEATURE_LAYERING}
+ self.rbd.RBD.clone.assert_called_once_with(*args, **kwargs)
+ self.assertEqual(client.__enter__.call_count, 2)
def test_extend_volume(self):
- fake_name = u'volume-00000001'
fake_size = '20'
- fake_vol = {'project_id': 'testprjid', 'name': fake_name,
+ fake_vol = {'project_id': 'testprjid', 'name': self.volume_name,
'size': fake_size,
'id': 'a720b3c0-d1f0-11e1-9b23-0800200c9a66'}
self.mox.VerifyAll()
- def test_rbd_volume_proxy_init(self):
- name = u'volume-00000001'
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_rbd_volume_proxy_init(self, mock_client):
snap = u'snapshot-name'
- self.stubs.Set(self.driver, '_connect_to_rados',
- lambda x: (None, None))
- self.mox.StubOutWithMock(self.driver, '_disconnect_from_rados')
-
- # no snapshot
- self.rbd.Image(None, str(name), snapshot=None, read_only=False) \
- .AndReturn(None)
- # snapshot
- self.rbd.Image(None, str(name), snapshot=str(snap), read_only=True) \
- .AndReturn(None)
- # error causes disconnect
- self.stubs.Set(self.rbd, 'Error', test.TestingException)
- self.rbd.Image(None, str(name), snapshot=None, read_only=False) \
- .AndRaise(test.TestingException)
- self.driver._disconnect_from_rados(None, None)
- self.mox.ReplayAll()
+ client = mock_client.return_value
+ client.__enter__.return_value = client
- driver.RBDVolumeProxy(self.driver, name)
- driver.RBDVolumeProxy(self.driver, name, snapshot=snap, read_only=True)
- self.assertRaises(test.TestingException,
- driver.RBDVolumeProxy, self.driver, name)
+ self.driver._connect_to_rados = mock.Mock()
+ self.driver._connect_to_rados.return_value = (None, None)
+ self.driver._disconnect_from_rados = mock.Mock()
+ self.driver._disconnect_from_rados.return_value = (None, None)
- def test_connect_to_rados(self):
- mock_client = self.mox.CreateMockAnything()
- mock_ioctx = self.mox.CreateMockAnything()
- self.stubs.Set(self.rados, 'Error', test.TestingException)
+ with driver.RBDVolumeProxy(self.driver, self.volume_name):
+ self.driver._connect_to_rados.assert_called_once()
+ self.assertFalse(self.driver._disconnect_from_rados.called)
- # default configured pool
- self.rados.Rados(rados_id=None, conffile=None).AndReturn(mock_client)
- mock_client.connect()
- mock_client.open_ioctx('rbd').AndReturn(mock_ioctx)
+ self.driver._disconnect_from_rados.assert_called_once()
- # different pool
- self.rados.Rados(rados_id=None, conffile=None).AndReturn(mock_client)
- mock_client.connect()
- mock_client.open_ioctx('images').AndReturn(mock_ioctx)
+ self.driver._connect_to_rados.reset_mock()
+ self.driver._disconnect_from_rados.reset_mock()
- # error
- self.rados.Rados(rados_id=None, conffile=None).AndReturn(mock_client)
- mock_client.connect()
- mock_client.open_ioctx('rbd').AndRaise(test.TestingException)
- mock_client.shutdown()
+ with driver.RBDVolumeProxy(self.driver, self.volume_name,
+ snapshot=snap):
+ self.driver._connect_to_rados.assert_called_once()
+ self.assertFalse(self.driver._disconnect_from_rados.called)
- self.mox.ReplayAll()
+ self.driver._disconnect_from_rados.assert_called_once()
+
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ def test_connect_to_rados(self, mock_client):
+ client = mock_client.return_value
+ client.__enter__.return_value = client
+ client.open_ioctx = mock.Mock()
+
+ mock_ioctx = mock.Mock()
+ client.open_ioctx.return_value = mock_ioctx
- self.assertEqual((mock_client, mock_ioctx),
+ self.rados.Error = test.TestingException
+ self.rados.Rados.return_value = client
+
+ # default configured pool
+ self.assertEqual((client, mock_ioctx),
self.driver._connect_to_rados())
- self.assertEqual((mock_client, mock_ioctx),
+ client.open_ioctx.assert_called_with(self.cfg.rbd_pool)
+
+ # different pool
+ self.assertEqual((client, mock_ioctx),
self.driver._connect_to_rados('images'))
+ client.open_ioctx.assert_called_with('images')
+
+ # error
+ client.open_ioctx.reset_mock()
+ client.shutdown.reset_mock()
+ client.open_ioctx.side_effect = self.rados.Error
self.assertRaises(test.TestingException, self.driver._connect_to_rados)
+ client.open_ioctx.assert_called_once()
+ client.shutdown.assert_called_once()
class RBDImageIOWrapperTestCase(test.TestCase):
with mock.patch.object(driver, 'LOG') as mock_logger:
self.meta.image.flush = mock.Mock()
self.rbd_wrapper.flush()
- self.assertTrue(self.meta.image.flush.called)
+ self.meta.image.flush.assert_called_once()
self.meta.image.flush.reset_mock()
# this should be caught and logged silently.
self.meta.image.flush.side_effect = AttributeError
self.rbd_wrapper.flush()
- self.assertTrue(self.meta.image.flush.called)
+ self.meta.image.flush.assert_called_once()
msg = _("flush() not supported in this version of librbd")
mock_logger.warning.assert_called_with(msg)
def setUp(self):
super(ManagedRBDTestCase, self).setUp()
+ # TODO(dosaboy): need to remove dependency on mox stubs here once
+ # image.fake has been converted to mock.
fake_image.stub_out_image_service(self.stubs)
self.volume.driver.set_initialized()
self.volume.stats = {'allocated_capacity_gb': 0}
NOTE: if clone_error is True we force the image type to raw otherwise
clone_image is not called
"""
- def mock_clone_image(volume, image_location, image_id, image_meta):
- self.called.append('clone_image')
- if clone_error:
- raise exception.CinderException()
- else:
- return {'provider_location': None}, True
+ volume_id = 1
# See tests.image.fake for image types.
if raw:
else:
image_id = 'c905cedb-7281-47e4-8a62-f26bc5fc4c77'
- volume_id = 1
-
# creating volume testdata
db.volume_create(self.context,
{'id': volume_id,
'instance_uuid': None,
'host': 'dummy'})
- mpo = mock.patch.object
- with mpo(self.volume.driver, 'create_volume') as mock_create_volume:
- with mpo(self.volume.driver, 'clone_image', mock_clone_image):
- with mpo(create_volume.CreateVolumeFromSpecTask,
- '_copy_image_to_volume') as mock_copy_image_to_volume:
-
- try:
- if not clone_error:
- self.volume.create_volume(self.context,
- volume_id,
- image_id=image_id)
- else:
- self.assertRaises(exception.CinderException,
- self.volume.create_volume,
- self.context,
- volume_id,
- image_id=image_id)
-
- volume = db.volume_get(self.context, volume_id)
- self.assertEqual(volume['status'], expected_status)
- finally:
- # cleanup
- db.volume_destroy(self.context, volume_id)
-
- self.assertEqual(self.called, ['clone_image'])
- mock_create_volume.assert_called()
- mock_copy_image_to_volume.assert_called()
+ try:
+ if not clone_error:
+ self.volume.create_volume(self.context,
+ volume_id,
+ image_id=image_id)
+ else:
+ self.assertRaises(exception.CinderException,
+ self.volume.create_volume,
+ self.context,
+ volume_id,
+ image_id=image_id)
+
+ volume = db.volume_get(self.context, volume_id)
+ self.assertEqual(volume['status'], expected_status)
+ finally:
+ # cleanup
+ db.volume_destroy(self.context, volume_id)
def test_create_vol_from_image_status_available(self):
"""Clone raw image then verify volume is in available state."""
- self._create_volume_from_image('available', raw=True)
+
+ def mock_clone_image(volume, image_location, image_id, image_meta):
+ return {'provider_location': None}, True
+
+ self.volume.driver.clone_image = mock.Mock()
+ self.volume.driver.clone_image.side_effect = mock_clone_image
+ self.volume.driver.create_volume = mock.Mock()
+
+ with mock.patch.object(create_volume.CreateVolumeFromSpecTask,
+ '_copy_image_to_volume') as mock_copy:
+ self._create_volume_from_image('available', raw=True)
+
+ self.volume.driver.clone_image.assert_called_once()
+ self.assertFalse(self.volume.driver.create_volume.called)
+ self.assertFalse(mock_copy.called)
def test_create_vol_from_non_raw_image_status_available(self):
"""Clone non-raw image then verify volume is in available state."""
- self._create_volume_from_image('available', raw=False)
+
+ def mock_clone_image(volume, image_location, image_id, image_meta):
+ return {'provider_location': None}, False
+
+ self.volume.driver.clone_image = mock.Mock()
+ self.volume.driver.clone_image.side_effect = mock_clone_image
+ self.volume.driver.create_volume = mock.Mock()
+ self.volume.driver.create_volume.return_value = None
+
+ with mock.patch.object(create_volume.CreateVolumeFromSpecTask,
+ '_copy_image_to_volume') as mock_copy:
+ self._create_volume_from_image('available', raw=False)
+
+ self.volume.driver.clone_image.assert_called_once()
+ self.volume.driver.create_volume.assert_called_once()
+ mock_copy.assert_called_once()
def test_create_vol_from_image_status_error(self):
"""Fail to clone raw image then verify volume is in error state."""
- self._create_volume_from_image('error', raw=True, clone_error=True)
+
+ self.volume.driver.clone_image = mock.Mock()
+ self.volume.driver.clone_image.side_effect = exception.CinderException
+ self.volume.driver.create_volume = mock.Mock()
+ self.volume.driver.create_volume.return_value = None
+
+ with mock.patch.object(create_volume.CreateVolumeFromSpecTask,
+ '_copy_image_to_volume') as mock_copy:
+ self._create_volume_from_image('error', raw=True, clone_error=True)
+
+ self.volume.driver.clone_image.assert_called_once()
+ self.assertFalse(self.volume.driver.create_volume.called)
+ self.assertFalse(mock_copy.called)
def test_clone_failure(self):
driver = self.volume.driver
def test_clone_success(self):
expected = ({'provider_location': None}, True)
driver = self.volume.driver
- mpo = mock.patch.object
- with mpo(driver, '_is_cloneable', lambda *args: True):
- with mpo(driver, '_parse_location', lambda x: (1, 2, 3, 4)):
- with mpo(driver, '_clone') as mock_clone:
- with mpo(driver, '_resize') as mock_resize:
- image_loc = (mock.Mock(), mock.Mock())
- actual = driver.clone_image(mock.Mock(),
- image_loc,
- mock.Mock(),
- {'disk_format': 'raw'})
- self.assertEqual(expected, actual)
- mock_clone.assert_called()
- mock_resize.assert_called()
+
+ self.volume.driver._is_cloneable = mock.Mock()
+ self.volume.driver._is_cloneable.return_value = True
+ self.volume.driver._clone = mock.Mock()
+ self.volume.driver._resize = mock.Mock()
+
+ image_loc = ('rbd://fee/fi/fo/fum', None)
+ actual = driver.clone_image({'name': 'vol1'},
+ image_loc,
+ 'id.foo',
+ {'disk_format': 'raw'})
+
+ self.assertEqual(expected, actual)
+ self.volume.driver._clone.assert_called_once()
+ self.volume.driver._resize.assert_called_once()
Code Review / openstack-build / cinder-build.git / commitdiff