# under the License.
""" Tests for Ceph backup service."""
-import eventlet
-import fcntl
import hashlib
import mock
import os
-import subprocess
import tempfile
-import time
import uuid
from cinder.backup.drivers import ceph
LOG = logging.getLogger(__name__)
+# This is used to collect raised exceptions so that tests may check what was
+# raised.
+# NOTE: this must be initialised in test setUp().
+RAISED_EXCEPTIONS = []
-class ImageNotFound(Exception):
- _called = False
- def __init__(self, *args, **kwargs):
- self.__class__._called = True
-
- @classmethod
- def called(cls):
- ret = cls._called
- cls._called = False
- return ret
-
-
-class ImageBusy(Exception):
- _called = False
+class MockException(Exception):
def __init__(self, *args, **kwargs):
- self.__class__._called = True
-
- @classmethod
- def called(cls):
- ret = cls._called
- cls._called = False
- return ret
-
-
-def common_backup_mocks(f):
- """Decorator to set mocks common to all backup tests.
- """
- def _common_backup_mocks_inner(inst, *args, **kwargs):
- inst.service.rbd.Image.size = mock.Mock()
- inst.service.rbd.Image.size.return_value = \
- inst.chunk_size * inst.num_chunks
+ RAISED_EXCEPTIONS.append(self.__class__)
- with mock.patch.object(inst.service, '_get_rbd_support') as \
- mock_rbd_support:
- mock_rbd_support.return_value = (True, 3)
- with mock.patch.object(inst.service, 'get_backup_snaps'):
- return f(inst, *args, **kwargs)
- return _common_backup_mocks_inner
+class MockImageNotFoundException(MockException):
+ """Used as mock for rbd.ImageNotFound."""
-def common_restore_mocks(f):
- """Decorator to set mocks common to all restore tests.
- """
- @common_backup_mocks
- def _common_restore_mocks_inner(inst, *args, **kwargs):
- return f(inst, *args, **kwargs)
-
- return _common_restore_mocks_inner
+class MockImageBusyException(MockException):
+ """Used as mock for rbd.ImageBusy."""
def common_mocks(f):
mocks that can't/dont't get unset.
"""
def _common_inner_inner1(inst, *args, **kwargs):
+ # NOTE(dosaboy): mock Popen to, by default, raise Exception in order to
+ # ensure that any test ending up in a subprocess fails
+ # if not properly mocked.
+ @mock.patch('subprocess.Popen')
+ # NOTE(dosaboy): mock out eventlet.sleep() so that it does nothing.
+ @mock.patch('eventlet.sleep')
+ @mock.patch('time.time')
@mock.patch('cinder.backup.drivers.ceph.rbd')
@mock.patch('cinder.backup.drivers.ceph.rados')
- def _common_inner_inner2(mock_rados, mock_rbd):
+ def _common_inner_inner2(mock_rados, mock_rbd, mock_time, mock_sleep,
+ mock_popen):
+ mock_time.side_effect = inst.time_inc
+ mock_popen.side_effect = Exception
+
inst.mock_rados = mock_rados
inst.mock_rbd = mock_rbd
inst.mock_rados.Rados = mock.Mock
inst.mock_rbd.RBD = mock.Mock
inst.mock_rbd.Image = mock.Mock
inst.mock_rbd.Image.close = mock.Mock()
- inst.mock_rbd.ImageBusy = ImageBusy
- inst.mock_rbd.ImageNotFound = ImageNotFound
+ inst.mock_rbd.ImageBusy = MockImageBusyException
+ inst.mock_rbd.ImageNotFound = MockImageNotFoundException
inst.service.rbd = inst.mock_rbd
inst.service.rados = inst.mock_rados
-
- with mock.patch.object(time, 'time') as mock_time:
- mock_time.side_effect = inst.time_inc
- with mock.patch.object(eventlet, 'sleep'):
- # Mock Popen to raise Exception in order to ensure that any
- # test ending up in a subprocess fails if not properly
- # mocked.
- with mock.patch.object(subprocess, 'Popen') as mock_popen:
- mock_popen.side_effect = Exception
- return f(inst, *args, **kwargs)
+ return f(inst, *args, **kwargs)
return _common_inner_inner2()
+
return _common_inner_inner1
mock_popen.side_effect = MockPopen
def setUp(self):
+ global RAISED_EXCEPTIONS
+ RAISED_EXCEPTIONS = []
super(BackupCephTestCase, self).setUp()
self.ctxt = context.get_admin_context()
self.assertEqual(len(snaps), 3)
@common_mocks
- @common_backup_mocks
def test_transfer_data_from_rbd_to_file(self):
self.mock_rbd.Image.read = mock.Mock()
self.mock_rbd.Image.read.return_value = \
self.assertEqual(checksum.digest(), self.checksum.digest())
@common_mocks
- @common_backup_mocks
def test_transfer_data_from_file_to_rbd(self):
def mock_write_data(data, offset):
self.assertEqual(checksum.digest(), self.checksum.digest())
@common_mocks
- @common_backup_mocks
def test_transfer_data_from_file_to_file(self):
with tempfile.NamedTemporaryFile() as test_file:
self.volume_file.seek(0)
self.assertEqual(checksum.digest(), self.checksum.digest())
@common_mocks
- @common_backup_mocks
def test_backup_volume_from_file(self):
+ checksum = hashlib.sha256()
def mock_write_data(data, offset):
checksum.update(data)
with mock.patch.object(self.service, '_discard_bytes'):
with tempfile.NamedTemporaryFile() as test_file:
- checksum = hashlib.sha256()
-
self.service.backup(self.backup, self.volume_file)
# Ensure the files are equal
self.assertEqual(checksum.digest(), self.checksum.digest())
+ self.assertTrue(self.service.rbd.Image.write.called)
+
@common_mocks
def test_get_backup_base_name(self):
name = self.service._get_backup_base_name(self.volume_id,
"volume-%s.backup.%s" % (self.volume_id, '1234'))
@common_mocks
- @common_backup_mocks
+ @mock.patch('fcntl.fcntl')
@mock.patch('subprocess.Popen')
- def test_backup_volume_from_rbd(self, mock_popen):
+ def test_backup_volume_from_rbd(self, mock_popen, mock_fnctl):
backup_name = self.service._get_backup_base_name(self.backup_id,
diff_format=True)
self.mock_rbd.RBD.list = mock.Mock()
self.mock_rbd.RBD.list.return_value = [backup_name]
- with mock.patch.object(fcntl, 'fcntl'):
- with mock.patch.object(self.service, '_discard_bytes'):
+ with mock.patch.object(self.service, 'get_backup_snaps') as \
+ mock_get_backup_snaps:
+ with mock.patch.object(self.service, '_full_backup') as \
+ mock_full_backup:
with mock.patch.object(self.service, '_try_delete_base_image'):
with tempfile.NamedTemporaryFile() as test_file:
checksum = hashlib.sha256()
'pool_foo',
'user_foo',
'conf_foo')
- rbd_io = rbddriver.RBDImageIOWrapper(meta)
-
- self.service.backup(self.backup, rbd_io)
+ self.service.backup(self.backup,
+ rbddriver.RBDImageIOWrapper(meta))
self.assertEqual(self.callstack, ['popen_init',
'read',
'stdout_close',
'communicate'])
+ self.assertFalse(mock_full_backup.called)
+ self.assertTrue(mock_get_backup_snaps.called)
+
# Ensure the files are equal
self.assertEqual(checksum.digest(),
self.checksum.digest())
@common_mocks
- @common_backup_mocks
def test_backup_vol_length_0(self):
volume_id = str(uuid.uuid4())
self._create_volume_db_entry(volume_id, 0)
backup, self.volume_file)
@common_mocks
- @common_restore_mocks
def test_restore(self):
backup_name = self.service._get_backup_base_name(self.backup_id,
diff_format=True)
self.mock_rbd.Image.read = mock.Mock()
self.mock_rbd.Image.read.side_effect = mock_read_data
- with mock.patch.object(self.service, '_discard_bytes'):
+ with mock.patch.object(self.service, '_discard_bytes') as \
+ mock_discard_bytes:
with tempfile.NamedTemporaryFile() as test_file:
self.volume_file.seek(0)
# Ensure the files are equal
self.assertEqual(checksum.digest(), self.checksum.digest())
+ self.assertTrue(mock_discard_bytes.called)
+
+ self.assertTrue(self.service.rbd.Image.read.called)
+
@common_mocks
def test_discard_bytes(self):
self.mock_rbd.Image.discard = mock.Mock()
self.mock_rbd.Image.write = mock.Mock()
self.mock_rbd.Image.flush = mock.Mock()
+ # Test discard with no remainder
with mock.patch.object(self.service, '_file_is_rbd') as \
mock_file_is_rbd:
mock_file_is_rbd.return_value = False
self.mock_rbd.Image.write.reset_mock()
self.mock_rbd.Image.flush.reset_mock()
+ self.mock_rbd.Image.discard.reset_mock()
+ # Now test with a remainder.
with mock.patch.object(self.service, '_file_is_rbd') as \
mock_file_is_rbd:
mock_file_is_rbd.return_value = False
self.mock_rbd.RBD.remove = mock.Mock()
self.mock_rbd.RBD.remove.side_effect = self.mock_rbd.ImageBusy
- with mock.patch.object(self.service, 'get_backup_snaps'):
+ with mock.patch.object(self.service, 'get_backup_snaps') as \
+ mock_get_backup_snaps:
self.assertRaises(self.mock_rbd.ImageBusy,
self.service._try_delete_base_image,
self.backup['id'], self.backup['volume_id'])
+ self.assertTrue(mock_get_backup_snaps.called)
- self.assertTrue(self.mock_rbd.RBD.list.called)
- self.assertTrue(self.mock_rbd.RBD.remove.called)
- self.assertTrue(self.mock_rbd.ImageBusy.called())
+ self.assertTrue(self.mock_rbd.RBD.list.called)
+ self.assertTrue(self.mock_rbd.RBD.remove.called)
+ self.assertTrue(MockImageBusyException in RAISED_EXCEPTIONS)
@common_mocks
def test_delete(self):
with mock.patch.object(self.service, '_try_delete_base_image'):
self.service.delete(self.backup)
- self.assertFalse(self.mock_rbd.ImageNotFound.called())
+ self.assertEqual(RAISED_EXCEPTIONS, [])
@common_mocks
def test_delete_image_not_found(self):
mock_del_base.side_effect = self.mock_rbd.ImageNotFound
# ImageNotFound exception is caught so that db entry can be cleared
self.service.delete(self.backup)
- self.assertTrue(self.mock_rbd.ImageNotFound.called())
+ self.assertEqual(RAISED_EXCEPTIONS, [MockImageNotFoundException])
@common_mocks
def test_diff_restore_allowed_true(self):
self.mock_rbd.Image.size = mock.Mock()
self.mock_rbd.Image.size.return_value = self.volume_size * units.GiB
- mpo = mock.patch.object
- with mpo(self.service, '_get_restore_point') as mock_restore_point:
+ with mock.patch.object(self.service, '_get_restore_point') as \
+ mock_restore_point:
mock_restore_point.return_value = restore_point
- with mpo(self.service, '_rbd_has_extents') as mock_rbd_has_extents:
+ with mock.patch.object(self.service, '_rbd_has_extents') as \
+ mock_rbd_has_extents:
mock_rbd_has_extents.return_value = False
- with mpo(self.service, '_rbd_image_exists') as \
+ with mock.patch.object(self.service, '_rbd_image_exists') as \
mock_rbd_image_exists:
mock_rbd_image_exists.return_value = (True, 'foo')
- with mpo(self.service, '_file_is_rbd') as \
+ with mock.patch.object(self.service, '_file_is_rbd') as \
mock_file_is_rbd:
mock_file_is_rbd.return_value = True
self.assertTrue(mock_rbd_has_extents.called)
@common_mocks
+ @mock.patch('fcntl.fcntl')
@mock.patch('subprocess.Popen')
- def test_piped_execute(self, mock_popen):
- with mock.patch.object(fcntl, 'fcntl') as mock_fcntl:
- mock_fcntl.return_value = 0
- self._setup_mock_popen(mock_popen, ['out', 'err'])
- self.service._piped_execute(['foo'], ['bar'])
- self.assertEqual(self.callstack, ['popen_init', 'popen_init',
- 'stdout_close', 'communicate'])
+ def test_piped_execute(self, mock_popen, mock_fcntl):
+ mock_fcntl.return_value = 0
+ self._setup_mock_popen(mock_popen, ['out', 'err'])
+ self.service._piped_execute(['foo'], ['bar'])
+ self.assertEqual(self.callstack, ['popen_init', 'popen_init',
+ 'stdout_close', 'communicate'])
LOG = logging.getLogger(__name__)
+# This is used to collect raised exceptions so that tests may check what was
+# raised.
+# NOTE: this must be initialised in test setUp().
+RAISED_EXCEPTIONS = []
+
+
+class MockException(Exception):
+
+ def __init__(self, *args, **kwargs):
+ RAISED_EXCEPTIONS.append(self.__class__)
+
+
+class MockImageNotFoundException(MockException):
+ """Used as mock for rbd.ImageNotFound."""
+
+
+class MockImageBusyException(MockException):
+ """Used as mock for rbd.ImageBusy."""
+
+
+def common_mocks(f):
+ """Decorator to set mocks common to all tests.
+
+ The point of doing these mocks here is so that we don't accidentally set
+ mocks that can't/dont't get unset.
+ """
+ def _common_inner_inner1(inst, *args, **kwargs):
+ @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
+ @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
+ @mock.patch('cinder.backup.drivers.ceph.rbd')
+ @mock.patch('cinder.backup.drivers.ceph.rados')
+ def _common_inner_inner2(mock_rados, mock_rbd, mock_client,
+ mock_proxy):
+ inst.mock_rbd = mock_rbd
+ inst.mock_rados = mock_rados
+ inst.mock_client = mock_client
+ inst.mock_proxy = mock_proxy
+ inst.mock_rados.Rados = mock.Mock
+ inst.mock_rados.Rados.ioctx = mock.Mock()
+ inst.mock_rbd.RBD = mock.Mock
+ inst.mock_rbd.Image = mock.Mock
+ inst.mock_rbd.Image.close = mock.Mock()
+ inst.mock_rbd.RBD.Error = Exception
+ inst.mock_rados.Error = Exception
+ inst.mock_rbd.ImageBusy = MockImageBusyException
+ inst.mock_rbd.ImageNotFound = MockImageNotFoundException
+
+ inst.driver.rbd = inst.mock_rbd
+ inst.driver.rados = inst.mock_rados
+ return f(inst, *args, **kwargs)
+
+ return _common_inner_inner2()
+
+ return _common_inner_inner1
+
+
CEPH_MON_DUMP = """dumped monmap epoch 1
{ "epoch": 1,
"fsid": "33630410-6d93-4d66-8e42-3b953cf194aa",
class RBDTestCase(test.TestCase):
def setUp(self):
+ global RAISED_EXCEPTIONS
+ RAISED_EXCEPTIONS = []
super(RBDTestCase, self).setUp()
self.cfg = mock.Mock(spec=conf.Configuration)
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)
+ configuration=self.cfg)
self.driver.set_initialized()
self.volume_name = u'volume-00000001'
def tearDown(self):
super(RBDTestCase, self).tearDown()
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_create_volume(self, mock_client):
- client = mock_client.return_value
+ @common_mocks
+ def test_create_volume(self):
+ client = self.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
+ with mock.patch.object(self.driver, '_supports_layering') as \
+ mock_supports_layering:
+ mock_supports_layering.return_value = True
+ self.mock_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.mock_rbd.RBD_FEATURE_LAYERING}
+ self.mock_rbd.RBD.create.assert_called_once_with(*args, **kwargs)
+ client.__enter__.assert_called_once()
+ client.__exit__.assert_called_once()
+ mock_supports_layering.assert_called_once()
+
+ @common_mocks
+ def test_create_volume_no_layering(self):
+ client = self.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)
+ with mock.patch.object(self.driver, '_supports_layering') as \
+ mock_supports_layering:
+ mock_supports_layering.return_value = False
+ self.mock_rbd.RBD.create = mock.Mock()
- args = [client.ioctx, str(self.volume_name),
- self.volume_size * units.GiB]
- kwargs = {'old_format': True,
- 'features': 0}
+ self.driver.create_volume(self.volume)
- 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)
+ args = [client.ioctx, str(self.volume_name),
+ self.volume_size * units.GiB]
+ kwargs = {'old_format': True,
+ 'features': 0}
+ self.mock_rbd.RBD.create.assert_called_once_with(*args, **kwargs)
+ client.__enter__.assert_called_once()
+ client.__exit__.assert_called_once()
+ mock_supports_layering.assert_called_once()
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_delete_volume(self, mock_client):
- client = mock_client.return_value
+ @common_mocks
+ def test_delete_volume(self):
+ client = self.mock_client.return_value
self.driver.rbd.Image.list_snaps = mock.Mock()
self.driver.rbd.Image.list_snaps.return_value = []
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()
-
- @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.rbd = mock_rbd
-
- with mock.patch.object(driver, 'RADOSClient'):
- self.assertIsNone(self.driver.delete_volume(self.volume))
- mock_rbd.Image.assert_called_once()
-
+ with mock.patch.object(self.driver, '_get_clone_info') as \
+ mock_get_clone_info:
+ with mock.patch.object(self.driver, '_delete_backup_snaps') as \
+ mock_delete_backup_snaps:
+ mock_get_clone_info.return_value = (None, None, None)
+
+ self.driver.delete_volume(self.volume)
+
+ mock_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()
+ mock_delete_backup_snaps.assert_called_once()
+ self.assertFalse(self.driver.rbd.Image.unprotect_snap.called)
+ self.driver.rbd.RBD.remove.assert_called_once()
+
+ @common_mocks
+ def delete_volume_not_found(self):
+ self.mock_rbd.Image.side_effect = self.mock_rbd.ImageNotFound
+ self.assertIsNone(self.driver.delete_volume(self.volume))
+ self.mock_rbd.Image.assert_called_once()
+ # Make sure the exception was raised
+ self.assertEqual(RAISED_EXCEPTIONS, [self.mock_rbd.ImageNotFound])
+
+ @common_mocks
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()
-
- self.rbd.ImageBusy = Exception
- self.rbd.RBD.remove = mock.Mock()
- self.rbd.RBD.remove.side_effect = self.rbd.ImageBusy
-
- self.driver._get_clone_info = mock.Mock()
- self.driver._get_clone_info.return_value = (None, None, None)
- self.driver._delete_backup_snaps = mock.Mock()
-
- with mock.patch.object(driver, 'RADOSClient') as mock_rados_client:
- self.assertRaises(exception.VolumeIsBusy,
- self.driver.delete_volume, self.volume)
-
- 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()
-
- @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
- def test_create_snapshot(self, mock_proxy):
- proxy = mock_proxy.return_value
+ self.mock_rbd.Image.list_snaps = mock.Mock()
+ self.mock_rbd.Image.list_snaps.return_value = []
+ self.mock_rbd.Image.unprotect_snap = mock.Mock()
+
+ self.mock_rbd.RBD.remove = mock.Mock()
+ self.mock_rbd.RBD.remove.side_effect = self.mock_rbd.ImageBusy
+
+ with mock.patch.object(self.driver, '_get_clone_info') as \
+ mock_get_clone_info:
+ mock_get_clone_info.return_value = (None, None, None)
+ with mock.patch.object(self.driver, '_delete_backup_snaps') as \
+ mock_delete_backup_snaps:
+ with mock.patch.object(driver, 'RADOSClient') as \
+ mock_rados_client:
+ self.assertRaises(exception.VolumeIsBusy,
+ self.driver.delete_volume, self.volume)
+
+ mock_get_clone_info.assert_called_once()
+ self.mock_rbd.Image.list_snaps.assert_called_once()
+ mock_rados_client.assert_called_once()
+ mock_delete_backup_snaps.assert_called_once()
+ self.assertFalse(self.mock_rbd.Image.unprotect_snap.called)
+ self.mock_rbd.RBD.remove.assert_called_once()
+ # Make sure the exception was raised
+ self.assertEqual(RAISED_EXCEPTIONS,
+ [self.mock_rbd.ImageBusy])
+
+ @common_mocks
+ def test_create_snapshot(self):
+ proxy = self.mock_proxy.return_value
proxy.__enter__.return_value = proxy
self.driver.create_snapshot(self.snapshot)
proxy.create_snap.assert_called_with(*args)
proxy.protect_snap.assert_called_with(*args)
- @mock.patch('cinder.volume.drivers.rbd.RBDVolumeProxy')
- def test_delete_snapshot(self, mock_proxy):
- proxy = mock_proxy.return_value
+ @common_mocks
+ def test_delete_snapshot(self):
+ proxy = self.mock_proxy.return_value
proxy.__enter__.return_value = proxy
self.driver.delete_snapshot(self.snapshot)
proxy.remove_snap.assert_called_with(*args)
proxy.unprotect_snap.assert_called_with(*args)
+ @common_mocks
def test_get_clone_info(self):
-
- volume = self.rbd.Image()
+ volume = self.mock_rbd.Image()
volume.set_snap = mock.Mock()
volume.parent_info = mock.Mock()
parent_info = ('a', 'b', '%s.clone_snap' % (self.volume_name))
self.assertFalse(volume.set_snap.called)
volume.parent_info.assert_called_once()
+ @common_mocks
def test_get_clone_info_w_snap(self):
-
- volume = self.rbd.Image()
+ volume = self.mock_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()
+ snapshot = self.mock_rbd.ImageSnapshot()
info = self.driver._get_clone_info(volume, self.volume_name,
snap=snapshot)
self.assertEqual(volume.set_snap.call_count, 2)
volume.parent_info.assert_called_once()
+ @common_mocks
def test_get_clone_info_w_exception(self):
-
- self.rbd.ImageNotFound = Exception
-
- volume = self.rbd.Image()
+ volume = self.mock_rbd.Image()
volume.set_snap = mock.Mock()
volume.parent_info = mock.Mock()
- volume.parent_info.side_effect = self.rbd.ImageNotFound
+ volume.parent_info.side_effect = self.mock_rbd.ImageNotFound
- snapshot = self.rbd.ImageSnapshot()
+ snapshot = self.mock_rbd.ImageSnapshot()
info = self.driver._get_clone_info(volume, self.volume_name,
snap=snapshot)
volume.set_snap.assert_called_once()
self.assertEqual(volume.set_snap.call_count, 2)
volume.parent_info.assert_called_once()
+ # Make sure the exception was raised
+ self.assertEqual(RAISED_EXCEPTIONS, [self.mock_rbd.ImageNotFound])
+ @common_mocks
def test_get_clone_info_deleted_volume(self):
-
- volume = self.rbd.Image()
+ volume = self.mock_rbd.Image()
volume.set_snap = mock.Mock()
volume.parent_info = mock.Mock()
parent_info = ('a', 'b', '%s.clone_snap' % (self.volume_name))
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):
+ @common_mocks
+ def test_create_cloned_volume(self):
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.mock_rbd.RBD.clone = mock.Mock()
- self.rbd.Image.create_snap = mock.Mock()
- self.rbd.Image.protect_snap = mock.Mock()
- self.rbd.Image.close = mock.Mock()
+ with mock.patch.object(self.driver, '_get_clone_depth') as \
+ mock_get_clone_depth:
+ # Try with no flatten required
+ mock_get_clone_depth.return_value = 1
- self.driver.create_cloned_volume(dict(name=dst_name),
- dict(name=src_name))
+ self.mock_rbd.Image.create_snap = mock.Mock()
+ self.mock_rbd.Image.protect_snap = mock.Mock()
+ self.mock_rbd.Image.close = mock.Mock()
- 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()
+ self.driver.create_cloned_volume(dict(name=dst_name),
+ dict(name=src_name))
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_create_cloned_volume_w_flatten(self, mock_client):
+ self.mock_rbd.Image.create_snap.assert_called_once()
+ self.mock_rbd.Image.protect_snap.assert_called_once()
+ self.mock_rbd.RBD.clone.assert_called_once()
+ self.mock_rbd.Image.close.assert_called_once()
+ self.assertTrue(mock_get_clone_depth.called)
+
+ @common_mocks
+ def test_create_cloned_volume_w_flatten(self):
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):
+ self.mock_rbd.RBD.clone = mock.Mock()
+ self.mock_rbd.RBD.clone.side_effect = self.mock_rbd.RBD.Error
+
+ with mock.patch.object(self.driver, '_get_clone_depth') as \
+ mock_get_clone_depth:
+ # Try with no flatten required
+ mock_get_clone_depth.return_value = 1
+
+ self.mock_rbd.Image.create_snap = mock.Mock()
+ self.mock_rbd.Image.protect_snap = mock.Mock()
+ self.mock_rbd.Image.unprotect_snap = mock.Mock()
+ self.mock_rbd.Image.remove_snap = mock.Mock()
+ self.mock_rbd.Image.close = mock.Mock()
+
+ self.assertRaises(self.mock_rbd.RBD.Error,
+ self.driver.create_cloned_volume,
+ dict(name=dst_name), dict(name=src_name))
+
+ self.mock_rbd.Image.create_snap.assert_called_once()
+ self.mock_rbd.Image.protect_snap.assert_called_once()
+ self.mock_rbd.RBD.clone.assert_called_once()
+ self.mock_rbd.Image.unprotect_snap.assert_called_once()
+ self.mock_rbd.Image.remove_snap.assert_called_once()
+ self.mock_rbd.Image.close.assert_called_once()
+ self.assertTrue(mock_get_clone_depth.called)
+
+ @common_mocks
+ def test_create_cloned_volume_w_clone_exception(self):
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()
-
+ self.mock_rbd.RBD.clone = mock.Mock()
+ self.mock_rbd.RBD.clone.side_effect = self.mock_rbd.RBD.Error
+ with mock.patch.object(self.driver, '_get_clone_depth') as \
+ mock_get_clone_depth:
+ # Try with no flatten required
+ mock_get_clone_depth.return_value = 1
+
+ self.mock_rbd.Image.create_snap = mock.Mock()
+ self.mock_rbd.Image.protect_snap = mock.Mock()
+ self.mock_rbd.Image.unprotect_snap = mock.Mock()
+ self.mock_rbd.Image.remove_snap = mock.Mock()
+ self.mock_rbd.Image.close = mock.Mock()
+
+ self.assertRaises(self.mock_rbd.RBD.Error,
+ self.driver.create_cloned_volume,
+ dict(name=dst_name), dict(name=src_name))
+
+ self.mock_rbd.Image.create_snap.assert_called_once()
+ self.mock_rbd.Image.protect_snap.assert_called_once()
+ self.mock_rbd.RBD.clone.assert_called_once()
+ self.mock_rbd.Image.unprotect_snap.assert_called_once()
+ self.mock_rbd.Image.remove_snap.assert_called_once()
+ self.mock_rbd.Image.close.assert_called_once()
+
+ @common_mocks
def test_good_locations(self):
locations = ['rbd://fsid/pool/image/snap',
'rbd://%2F/%2F/%2F/%2F', ]
map(self.driver._parse_location, locations)
+ @common_mocks
def test_bad_locations(self):
locations = ['rbd://image',
'http://path/to/somewhere/else',
self.assertFalse(
self.driver._is_cloneable(loc, {'disk_format': 'raw'}))
- @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'
- info = {'disk_format': 'raw'}
- self.assertTrue(self.driver._is_cloneable(location, info))
-
- @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'}))
-
- @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.rbd.Error = Exception
- mock_proxy.side_effect = self.rbd.Error
-
- args = [location, {'disk_format': 'raw'}]
- self.assertFalse(self.driver._is_cloneable(*args))
- mock_proxy.assert_called_once()
+ @common_mocks
+ def test_cloneable(self):
+ with mock.patch.object(self.driver, '_get_fsid') as mock_get_fsid:
+ mock_get_fsid.return_value = 'abc'
+ location = 'rbd://abc/pool/image/snap'
+ info = {'disk_format': 'raw'}
+ self.assertTrue(self.driver._is_cloneable(location, info))
+ self.assertTrue(mock_get_fsid.called)
+
+ @common_mocks
+ def test_uncloneable_different_fsid(self):
+ with mock.patch.object(self.driver, '_get_fsid') as mock_get_fsid:
+ mock_get_fsid.return_value = 'abc'
+ location = 'rbd://def/pool/image/snap'
+ self.assertFalse(
+ self.driver._is_cloneable(location, {'disk_format': 'raw'}))
+ self.assertTrue(mock_get_fsid.called)
+
+ @common_mocks
+ def test_uncloneable_unreadable(self):
+ with mock.patch.object(self.driver, '_get_fsid') as mock_get_fsid:
+ mock_get_fsid.return_value = 'abc'
+ location = 'rbd://abc/pool/image/snap'
+
+ self.mock_proxy.side_effect = self.mock_rbd.Error
+ args = [location, {'disk_format': 'raw'}]
+ self.assertFalse(self.driver._is_cloneable(*args))
+ self.mock_proxy.assert_called_once()
+ self.assertTrue(mock_get_fsid.called)
+
+ @common_mocks
def test_uncloneable_bad_format(self):
- 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.assertFalse(
- self.driver._is_cloneable(location, {'disk_format': f}))
+ with mock.patch.object(self.driver, '_get_fsid') as mock_get_fsid:
+ mock_get_fsid.return_value = 'abc'
+ location = 'rbd://abc/pool/image/snap'
+ formats = ['qcow2', 'vmdk', 'vdi']
+ for f in formats:
+ self.assertFalse(
+ self.driver._is_cloneable(location, {'disk_format': f}))
+ self.assertTrue(mock_get_fsid.called)
def _copy_image(self):
with mock.patch.object(tempfile, 'NamedTemporaryFile'):
mock_image_service, None]
self.driver.copy_image_to_volume(*args)
+ @common_mocks
def test_copy_image_no_volume_tmp(self):
self.cfg.volume_tmp_dir = None
self._copy_image()
+ @common_mocks
def test_copy_image_volume_tmp(self):
self.cfg.volume_tmp_dir = '/var/run/cinder/tmp'
self._copy_image()
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_update_volume_stats(self, mock_client):
- client = mock_client.return_value
+ @common_mocks
+ def test_update_volume_stats(self):
+ client = self.mock_client.return_value
client.__enter__.return_value = client
client.cluster = mock.Mock()
client.cluster.get_cluster_stats.assert_called_once()
self.assertDictMatch(expected, actual)
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_update_volume_stats_error(self, mock_client):
- client = mock_client.return_value
+ @common_mocks
+ def test_update_volume_stats_error(self):
+ client = self.mock_client.return_value
client.__enter__.return_value = client
client.cluster = mock.Mock()
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,
client.cluster.get_cluster_stats.assert_called_once()
self.assertDictMatch(expected, actual)
+ @common_mocks
def test_get_mon_addrs(self):
- 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())
+ with mock.patch.object(self.driver, '_execute') as mock_execute:
+ mock_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())
+ @common_mocks
def test_initialize_connection(self):
hosts = ['::1', '::1', '::1', '127.0.0.1', 'example.com']
ports = ['6789', '6790', '6791', '6792', '6791']
- 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.cfg.rbd_pool,
- self.volume_name),
- 'hosts': hosts,
- 'ports': ports,
- 'auth_enabled': False,
- 'auth_username': None,
- 'secret_type': 'ceph',
- 'secret_uuid': None, }
- }
- actual = self.driver.initialize_connection(dict(name=self.volume_name),
- None)
- self.assertDictMatch(expected, actual)
-
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_clone(self, mock_client):
+ with mock.patch.object(self.driver, '_get_mon_addrs') as \
+ mock_get_mon_addrs:
+ mock_get_mon_addrs.return_value = (hosts, ports)
+
+ expected = {
+ 'driver_volume_type': 'rbd',
+ 'data': {
+ 'name': '%s/%s' % (self.cfg.rbd_pool,
+ self.volume_name),
+ 'hosts': hosts,
+ 'ports': ports,
+ 'auth_enabled': False,
+ 'auth_username': None,
+ 'secret_type': 'ceph',
+ 'secret_uuid': None, }
+ }
+ volume = dict(name=self.volume_name)
+ actual = self.driver.initialize_connection(volume, None)
+ self.assertDictMatch(expected, actual)
+ self.assertTrue(mock_get_mon_addrs.called)
+
+ @common_mocks
+ def test_clone(self):
src_pool = u'images'
src_image = u'image-name'
src_snap = u'snapshot-name'
return inst
return _inner
- client = mock_client.return_value
+ client = self.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.mock_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)
+ kwargs = {'features': self.mock_rbd.RBD_FEATURE_LAYERING}
+ self.mock_rbd.RBD.clone.assert_called_once_with(*args, **kwargs)
self.assertEqual(client.__enter__.call_count, 2)
+ @common_mocks
def test_extend_volume(self):
fake_size = '20'
fake_vol = {'project_id': 'testprjid', 'name': self.volume_name,
self.mox.VerifyAll()
- @mock.patch('cinder.volume.drivers.rbd.RADOSClient')
- def test_rbd_volume_proxy_init(self, mock_client):
+ @common_mocks
+ def test_rbd_volume_proxy_init(self):
snap = u'snapshot-name'
- client = mock_client.return_value
+ client = self.mock_client.return_value
client.__enter__.return_value = client
- 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)
+ with mock.patch.object(self.driver, '_connect_to_rados') as \
+ mock_connect_from_rados:
+ with mock.patch.object(self.driver, '_disconnect_from_rados') as \
+ mock_disconnect_from_rados:
+ mock_connect_from_rados.return_value = (None, None)
+ mock_disconnect_from_rados.return_value = (None, None)
- with driver.RBDVolumeProxy(self.driver, self.volume_name):
- self.driver._connect_to_rados.assert_called_once()
- self.assertFalse(self.driver._disconnect_from_rados.called)
+ with driver.RBDVolumeProxy(self.driver, self.volume_name):
+ mock_connect_from_rados.assert_called_once()
+ self.assertFalse(mock_disconnect_from_rados.called)
- self.driver._disconnect_from_rados.assert_called_once()
+ mock_disconnect_from_rados.assert_called_once()
- self.driver._connect_to_rados.reset_mock()
- self.driver._disconnect_from_rados.reset_mock()
+ mock_connect_from_rados.reset_mock()
+ mock_disconnect_from_rados.reset_mock()
- 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)
+ with driver.RBDVolumeProxy(self.driver, self.volume_name,
+ snapshot=snap):
+ mock_connect_from_rados.assert_called_once()
+ self.assertFalse(mock_disconnect_from_rados.called)
- self.driver._disconnect_from_rados.assert_called_once()
+ mock_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.rados.Error = test.TestingException
- self.rados.Rados.return_value = client
+ @common_mocks
+ def test_connect_to_rados(self):
+ self.mock_rados.Rados.connect = mock.Mock()
+ self.mock_rados.Rados.shutdown = mock.Mock()
+ self.mock_rados.Rados.open_ioctx = mock.Mock()
+ self.mock_rados.Rados.open_ioctx.return_value = \
+ self.mock_rados.Rados.ioctx
# default configured pool
- self.assertEqual((client, mock_ioctx),
- self.driver._connect_to_rados())
- client.open_ioctx.assert_called_with(self.cfg.rbd_pool)
+ ret = self.driver._connect_to_rados()
+ self.assertTrue(self.mock_rados.Rados.connect.called)
+ self.assertTrue(self.mock_rados.Rados.open_ioctx.called)
+ self.assertIsInstance(ret[0], self.mock_rados.Rados)
+ self.assertEqual(ret[1], self.mock_rados.Rados.ioctx)
+ self.mock_rados.Rados.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')
+ ret = self.driver._connect_to_rados('alt_pool')
+ self.assertTrue(self.mock_rados.Rados.connect.called)
+ self.assertTrue(self.mock_rados.Rados.open_ioctx.called)
+ self.assertIsInstance(ret[0], self.mock_rados.Rados)
+ self.assertEqual(ret[1], self.mock_rados.Rados.ioctx)
+ self.mock_rados.Rados.open_ioctx.assert_called_with('alt_pool')
# 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()
+ self.mock_rados.Rados.open_ioctx.reset_mock()
+ self.mock_rados.Rados.shutdown.reset_mock()
+ self.mock_rados.Rados.open_ioctx.side_effect = self.mock_rados.Error
+ self.assertRaises(self.mock_rados.Error, self.driver._connect_to_rados)
+ self.mock_rados.Rados.open_ioctx.assert_called_once()
+ self.mock_rados.Rados.shutdown.assert_called_once()
class RBDImageIOWrapperTestCase(test.TestCase):
self.meta.image.read = mock.Mock()
self.meta.image.write = mock.Mock()
self.meta.image.size = mock.Mock()
- self.rbd_wrapper = driver.RBDImageIOWrapper(self.meta)
+ self.mock_rbd_wrapper = driver.RBDImageIOWrapper(self.meta)
self.data_length = 1024
self.full_data = 'abcd' * 256
super(RBDImageIOWrapperTestCase, self).tearDown()
def test_init(self):
- self.assertEqual(self.rbd_wrapper._rbd_meta, self.meta)
- self.assertEqual(self.rbd_wrapper._offset, 0)
+ self.assertEqual(self.mock_rbd_wrapper._rbd_meta, self.meta)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 0)
def test_inc_offset(self):
- self.rbd_wrapper._inc_offset(10)
- self.rbd_wrapper._inc_offset(10)
- self.assertEqual(self.rbd_wrapper._offset, 20)
+ self.mock_rbd_wrapper._inc_offset(10)
+ self.mock_rbd_wrapper._inc_offset(10)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 20)
def test_rbd_image(self):
- self.assertEqual(self.rbd_wrapper.rbd_image, self.meta.image)
+ self.assertEqual(self.mock_rbd_wrapper.rbd_image, self.meta.image)
def test_rbd_user(self):
- self.assertEqual(self.rbd_wrapper.rbd_user, self.meta.user)
+ self.assertEqual(self.mock_rbd_wrapper.rbd_user, self.meta.user)
def test_rbd_pool(self):
- self.assertEqual(self.rbd_wrapper.rbd_conf, self.meta.conf)
+ self.assertEqual(self.mock_rbd_wrapper.rbd_conf, self.meta.conf)
def test_rbd_conf(self):
- self.assertEqual(self.rbd_wrapper.rbd_pool, self.meta.pool)
+ self.assertEqual(self.mock_rbd_wrapper.rbd_pool, self.meta.pool)
def test_read(self):
self.meta.image.read.side_effect = mock_read
self.meta.image.size.return_value = self.data_length
- data = self.rbd_wrapper.read()
+ data = self.mock_rbd_wrapper.read()
self.assertEqual(data, self.full_data)
- data = self.rbd_wrapper.read()
+ data = self.mock_rbd_wrapper.read()
self.assertEqual(data, '')
- self.rbd_wrapper.seek(0)
- data = self.rbd_wrapper.read()
+ self.mock_rbd_wrapper.seek(0)
+ data = self.mock_rbd_wrapper.read()
self.assertEqual(data, self.full_data)
- self.rbd_wrapper.seek(0)
- data = self.rbd_wrapper.read(10)
+ self.mock_rbd_wrapper.seek(0)
+ data = self.mock_rbd_wrapper.read(10)
self.assertEqual(data, self.full_data[:10])
def test_write(self):
- self.rbd_wrapper.write(self.full_data)
- self.assertEqual(self.rbd_wrapper._offset, 1024)
+ self.mock_rbd_wrapper.write(self.full_data)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 1024)
def test_seekable(self):
- self.assertTrue(self.rbd_wrapper.seekable)
+ self.assertTrue(self.mock_rbd_wrapper.seekable)
def test_seek(self):
- self.assertEqual(self.rbd_wrapper._offset, 0)
- self.rbd_wrapper.seek(10)
- self.assertEqual(self.rbd_wrapper._offset, 10)
- self.rbd_wrapper.seek(10)
- self.assertEqual(self.rbd_wrapper._offset, 10)
- self.rbd_wrapper.seek(10, 1)
- self.assertEqual(self.rbd_wrapper._offset, 20)
-
- self.rbd_wrapper.seek(0)
- self.rbd_wrapper.write(self.full_data)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 0)
+ self.mock_rbd_wrapper.seek(10)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 10)
+ self.mock_rbd_wrapper.seek(10)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 10)
+ self.mock_rbd_wrapper.seek(10, 1)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 20)
+
+ self.mock_rbd_wrapper.seek(0)
+ self.mock_rbd_wrapper.write(self.full_data)
self.meta.image.size.return_value = self.data_length
- self.rbd_wrapper.seek(0)
- self.assertEqual(self.rbd_wrapper._offset, 0)
+ self.mock_rbd_wrapper.seek(0)
+ self.assertEqual(self.mock_rbd_wrapper._offset, 0)
- self.rbd_wrapper.seek(10, 2)
- self.assertEqual(self.rbd_wrapper._offset, self.data_length + 10)
- self.rbd_wrapper.seek(-10, 2)
- self.assertEqual(self.rbd_wrapper._offset, self.data_length - 10)
+ self.mock_rbd_wrapper.seek(10, 2)
+ self.assertEqual(self.mock_rbd_wrapper._offset, self.data_length + 10)
+ self.mock_rbd_wrapper.seek(-10, 2)
+ self.assertEqual(self.mock_rbd_wrapper._offset, self.data_length - 10)
# test exceptions.
- self.assertRaises(IOError, self.rbd_wrapper.seek, 0, 3)
- self.assertRaises(IOError, self.rbd_wrapper.seek, -1)
+ self.assertRaises(IOError, self.mock_rbd_wrapper.seek, 0, 3)
+ self.assertRaises(IOError, self.mock_rbd_wrapper.seek, -1)
# offset should not have been changed by any of the previous
# operations.
- self.assertEqual(self.rbd_wrapper._offset, self.data_length - 10)
+ self.assertEqual(self.mock_rbd_wrapper._offset, self.data_length - 10)
def test_tell(self):
- self.assertEqual(self.rbd_wrapper.tell(), 0)
- self.rbd_wrapper._inc_offset(10)
- self.assertEqual(self.rbd_wrapper.tell(), 10)
+ self.assertEqual(self.mock_rbd_wrapper.tell(), 0)
+ self.mock_rbd_wrapper._inc_offset(10)
+ self.assertEqual(self.mock_rbd_wrapper.tell(), 10)
def test_flush(self):
with mock.patch.object(driver, 'LOG') as mock_logger:
self.meta.image.flush = mock.Mock()
- self.rbd_wrapper.flush()
+ self.mock_rbd_wrapper.flush()
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.mock_rbd_wrapper.flush()
self.meta.image.flush.assert_called_once()
msg = _("flush() not supported in this version of librbd")
mock_logger.warning.assert_called_with(msg)
def test_fileno(self):
- self.assertRaises(IOError, self.rbd_wrapper.fileno)
+ self.assertRaises(IOError, self.mock_rbd_wrapper.fileno)
def test_close(self):
- self.rbd_wrapper.close()
+ self.mock_rbd_wrapper.close()
class ManagedRBDTestCase(DriverTestCase):
def test_create_vol_from_image_status_available(self):
"""Clone raw image then verify volume is in available state."""
- def mock_clone_image(volume, image_location, image_id, image_meta):
+ 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)
+ with mock.patch.object(self.volume.driver, 'clone_image') as \
+ mock_clone_image:
+ mock_clone_image.side_effect = _mock_clone_image
+ with mock.patch.object(self.volume.driver, 'create_volume') as \
+ mock_create:
+ with mock.patch.object(create_volume.CreateVolumeFromSpecTask,
+ '_copy_image_to_volume') as mock_copy:
+ self._create_volume_from_image('available', raw=True)
+ self.assertFalse(mock_copy.called)
- self.volume.driver.clone_image.assert_called_once()
- self.assertFalse(self.volume.driver.create_volume.called)
- self.assertFalse(mock_copy.called)
+ mock_clone_image.assert_called_once()
+ self.assertFalse(mock_create.called)
def test_create_vol_from_non_raw_image_status_available(self):
"""Clone non-raw image then verify volume is in available state."""
- def mock_clone_image(volume, image_location, image_id, image_meta):
+ 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(self.volume.driver, 'clone_image') as \
+ mock_clone_image:
+ mock_clone_image.side_effect = _mock_clone_image
+ with mock.patch.object(self.volume.driver, 'create_volume') as \
+ mock_create:
+ with mock.patch.object(create_volume.CreateVolumeFromSpecTask,
+ '_copy_image_to_volume') as mock_copy:
+ self._create_volume_from_image('available', raw=False)
+ mock_copy.assert_called_once()
- 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()
+ mock_clone_image.assert_called_once()
+ mock_create.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.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)
+ with mock.patch.object(self.volume.driver, 'clone_image') as \
+ mock_clone_image:
+ mock_clone_image.side_effect = exception.CinderException
+ with mock.patch.object(self.volume.driver, 'create_volume') as \
+ mock_create:
+ 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.assertFalse(mock_copy.called)
+
+ mock_clone_image.assert_called_once()
+ self.assertFalse(self.volume.driver.create_volume.called)
def test_clone_failure(self):
driver = self.volume.driver
expected = ({'provider_location': None}, True)
driver = self.volume.driver
- 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()
+ with mock.patch.object(self.volume.driver, '_is_cloneable') as \
+ mock_is_cloneable:
+ mock_is_cloneable.return_value = True
+ with mock.patch.object(self.volume.driver, '_clone') as \
+ mock_clone:
+ with mock.patch.object(self.volume.driver, '_resize') as \
+ mock_resize:
+ 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)
+ mock_clone.assert_called_once()
+ mock_resize.assert_called_once()
Code Review / openstack-build / cinder-build.git / commitdiff