From 855caef27a6525fc6b200d014dd7fb5de9dfa7ba Mon Sep 17 00:00:00 2001 From: Edward Hope-Morley Date: Mon, 23 Dec 2013 21:10:55 +0000 Subject: [PATCH] Convert RBD unit tests to use mock instead of mox Change-Id: If32e14a8f5ff180b6c0ff46797bcb81b21f805f1 Closes-bug: 1261864 --- cinder/tests/test_rbd.py | 907 ++++++++++++++++++++++----------------- 1 file changed, 508 insertions(+), 399 deletions(-) diff --git a/cinder/tests/test_rbd.py b/cinder/tests/test_rbd.py index efe8959ce..5c435d1c5 100644 --- a/cinder/tests/test_rbd.py +++ b/cinder/tests/test_rbd.py @@ -1,5 +1,6 @@ # Copyright 2012 Josh Durgin +# Copyright 2013 Canonical Ltd. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may @@ -15,9 +16,7 @@ # under the License. -import contextlib import mock -import mox import os import tempfile @@ -27,8 +26,6 @@ from cinder.image import image_utils 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 @@ -68,11 +65,6 @@ CEPH_MON_DUMP = """dumped monmap epoch 1 """ -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)) @@ -87,213 +79,319 @@ class RBDTestCase(test.TestCase): 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', ] @@ -314,48 +412,38 @@ class RBDTestCase(test.TestCase): 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: @@ -363,105 +451,97 @@ class RBDTestCase(test.TestCase): 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, @@ -469,43 +549,41 @@ class RBDTestCase(test.TestCase): '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'} @@ -518,60 +596,63 @@ class RBDTestCase(test.TestCase): 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): @@ -678,12 +759,12 @@ 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) @@ -699,6 +780,8 @@ class ManagedRBDTestCase(DriverTestCase): 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} @@ -712,12 +795,7 @@ class ManagedRBDTestCase(DriverTestCase): 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: @@ -725,8 +803,6 @@ class ManagedRBDTestCase(DriverTestCase): else: image_id = 'c905cedb-7281-47e4-8a62-f26bc5fc4c77' - volume_id = 1 - # creating volume testdata db.volume_create(self.context, {'id': volume_id, @@ -737,45 +813,76 @@ class ManagedRBDTestCase(DriverTestCase): '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 @@ -792,16 +899,18 @@ class ManagedRBDTestCase(DriverTestCase): 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() -- 2.45.2