openstack/deb-cinder
Code Issues Proposed changes

Stop using mox in unit/volume/drivers/*.py

Browse Source 
As mox is not compatible with python 3, all test cases
that use mox should be updated to use mock instead.

Replaced stubs.Set() with mock_object() in some classes in
cinder/tests/unit/volume/drivers/.

Change-Id: I8c180d3d466cf4b9bb19e939bafaffd5c9b382de
This commit is contained in:
Dao Cong Tien 2016-08-19 14:39:38 +07:00
parent 175667cf4e
commit 9c191bd316
6 changed files with 114 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
cinder/tests/unit/volume/drivers/test_drbdmanagedrv.py
View File

@ -324,21 +324,21 @@ class DrbdManageIscsiTestCase(test.TestCase):
super(DrbdManageIscsiTestCase, self).setUp()
self.stubs.Set(importutils, 'import_object',
self.fake_import_object)
self.stubs.Set(drv.DrbdManageBaseDriver,
'call_or_reconnect',
self.fake_issue_dbus_call)
self.stubs.Set(drv.DrbdManageBaseDriver,
'dbus_connect',
self.fake_issue_dbus_connect)
self.stubs.Set(drv.DrbdManageBaseDriver,
'_wait_for_node_assignment',
self.fake_wait_node_assignment)
self.mock_object(importutils, 'import_object',
self.fake_import_object)
self.mock_object(drv.DrbdManageBaseDriver,
'call_or_reconnect',
self.fake_issue_dbus_call)
self.mock_object(drv.DrbdManageBaseDriver,
'dbus_connect',
self.fake_issue_dbus_connect)
self.mock_object(drv.DrbdManageBaseDriver,
'_wait_for_node_assignment',
self.fake_wait_node_assignment)
self.configuration.safe_get = self._fake_safe_get
self.stubs.Set(eventlet, 'sleep', self._fake_sleep)
self.mock_object(eventlet, 'sleep', self._fake_sleep)
# Infrastructure
def fake_import_object(self, what, configuration, db, executor):
@ -624,9 +624,9 @@ class DrbdManageDrbdTestCase(DrbdManageIscsiTestCase):
def setUp(self):
super(DrbdManageDrbdTestCase, self).setUp()
self.stubs.Set(drv.DrbdManageDrbdDriver,
'_is_external_node',
self.fake_is_external_node)
self.mock_object(drv.DrbdManageDrbdDriver,
'_is_external_node',
self.fake_is_external_node)
def test_drbd_create_export(self):
volume = {'project_id': 'testprjid',

20
cinder/tests/unit/volume/drivers/test_fujitsu.py
View File

@ -711,12 +711,12 @@ class FJFCDriverTestCase(test.TestCase):
self.configuration = mock.Mock(spec=conf.Configuration)
self.configuration.cinder_eternus_config_file = self.config_file.name
self.stubs.Set(dx_common.FJDXCommon, '_get_eternus_connection',
self.fake_eternus_connection)
self.mock_object(dx_common.FJDXCommon, '_get_eternus_connection',
self.fake_eternus_connection)
instancename = FakeCIMInstanceName()
self.stubs.Set(dx_common.FJDXCommon, '_create_eternus_instance_name',
instancename.fake_create_eternus_instance_name)
self.mock_object(dx_common.FJDXCommon, '_create_eternus_instance_name',
instancename.fake_create_eternus_instance_name)
# Set iscsi driver to self.driver.
driver = dx_fc.FJDXFCDriver(configuration=self.configuration)
@ -824,15 +824,15 @@ class FJISCSIDriverTestCase(test.TestCase):
self.configuration = mock.Mock(spec=conf.Configuration)
self.configuration.cinder_eternus_config_file = self.config_file.name
self.stubs.Set(dx_common.FJDXCommon, '_get_eternus_connection',
self.fake_eternus_connection)
self.mock_object(dx_common.FJDXCommon, '_get_eternus_connection',
self.fake_eternus_connection)
instancename = FakeCIMInstanceName()
self.stubs.Set(dx_common.FJDXCommon, '_create_eternus_instance_name',
instancename.fake_create_eternus_instance_name)
self.mock_object(dx_common.FJDXCommon, '_create_eternus_instance_name',
instancename.fake_create_eternus_instance_name)
self.stubs.Set(dx_common.FJDXCommon, '_get_mapdata_iscsi',
self.fake_get_mapdata)
self.mock_object(dx_common.FJDXCommon, '_get_mapdata_iscsi',
self.fake_get_mapdata)
# Set iscsi driver to self.driver.
driver = dx_iscsi.FJDXISCSIDriver(configuration=self.configuration)

2
cinder/tests/unit/volume/drivers/test_hgst.py
View File

@ -35,7 +35,7 @@ class HGSTTestCase(test.TestCase):
def setUp(self, mock_ghn, mock_grnam, mock_pwnam):
"""Set up UUT and all the flags required for later fake_executes."""
super(HGSTTestCase, self).setUp()
self.stubs.Set(processutils, 'execute', self._fake_execute)
self.mock_object(processutils, 'execute', self._fake_execute)
self._fail_vgc_cluster = False
self._fail_ip = False
self._fail_network_list = False

15
cinder/tests/unit/volume/drivers/test_sheepdog.py
View File

@ -1191,14 +1191,13 @@ class SheepdogDriverTestCase(test.TestCase):
def fake_try_execute(obj, *command, **kwargs):
return True
self.stubs.Set(image_utils, 'temporary_file', fake_temp_file)
self.stubs.Set(image_utils, 'fetch_verify_image',
lambda w, x, y, z: None)
self.stubs.Set(image_utils, 'convert_image',
lambda x, y, z: None)
self.stubs.Set(sheepdog.SheepdogDriver,
'_try_execute',
fake_try_execute)
self.mock_object(image_utils, 'temporary_file', fake_temp_file)
self.mock_object(image_utils, 'fetch_verify_image',
return_value=None)
self.mock_object(image_utils, 'convert_image',
return_value=None)
self.mock_object(sheepdog.SheepdogDriver, '_try_execute',
fake_try_execute)
fake_run_dog.return_value = ('fake_stdout', 'fake_stderr')
self.driver.copy_image_to_volume(None, self.test_data.TEST_VOLUME,
FakeImageService(), None)

128
cinder/tests/unit/volume/drivers/test_solidfire.py
View File

@ -50,12 +50,12 @@ class SolidFireVolumeTestCase(test.TestCase):
self.configuration.replication_device = []
super(SolidFireVolumeTestCase, self).setUp()
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.stubs.Set(solidfire.SolidFireDriver,
'_build_endpoint_info',
self.fake_build_endpoint_info)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_build_endpoint_info',
self.fake_build_endpoint_info)
self.expected_qos_results = {'minIOPS': 1000,
'maxIOPS': 10000,
@ -458,21 +458,21 @@ class SolidFireVolumeTestCase(test.TestCase):
def test_create_volume_fails(self):
# NOTE(JDG) This test just fakes update_cluster_status
# this is inentional for this test
self.stubs.Set(solidfire.SolidFireDriver,
'_update_cluster_status',
self.fake_update_cluster_status)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_update_cluster_status',
self.fake_update_cluster_status)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
testvol = {'project_id': 'testprjid',
'name': 'testvol',
'size': 1,
'id': 'a720b3c0-d1f0-11e1-9b23-0800200c9a66',
'created_at': timeutils.utcnow()}
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
try:
sfv.create_volume(testvol)
self.fail("Should have thrown Error")
@ -481,33 +481,33 @@ class SolidFireVolumeTestCase(test.TestCase):
def test_create_sfaccount(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
account = sfv._create_sfaccount('project-id')
self.assertIsNotNone(account)
def test_create_sfaccount_fails(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.assertRaises(exception.SolidFireAPIException,
sfv._create_sfaccount, 'project-id')
def test_get_sfaccount_by_name(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
account = sfv._get_sfaccount_by_name('some-name')
self.assertIsNotNone(account)
def test_get_sfaccount_by_name_fails(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.assertRaises(exception.SolidFireAPIException,
sfv._get_sfaccount_by_name, 'some-name')
@ -604,9 +604,9 @@ class SolidFireVolumeTestCase(test.TestCase):
sfv.delete_snapshot(testsnap)
def test_extend_volume(self):
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
testvol = {'project_id': 'testprjid',
'name': 'test_volume',
'size': 1,
@ -617,9 +617,9 @@ class SolidFireVolumeTestCase(test.TestCase):
sfv.extend_volume(testvol, 2)
def test_extend_volume_fails_no_volume(self):
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
testvol = {'project_id': 'testprjid',
'name': 'no-name',
'size': 1,
@ -632,12 +632,12 @@ class SolidFireVolumeTestCase(test.TestCase):
def test_extend_volume_fails_account_lookup(self):
# NOTE(JDG) This test just fakes update_cluster_status
# this is intentional for this test
self.stubs.Set(solidfire.SolidFireDriver,
'_update_cluster_status',
self.fake_update_cluster_status)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_update_cluster_status',
self.fake_update_cluster_status)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
testvol = {'project_id': 'testprjid',
'name': 'no-name',
'size': 1,
@ -645,9 +645,9 @@ class SolidFireVolumeTestCase(test.TestCase):
'created_at': timeutils.utcnow()}
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request_fails)
self.assertRaises(exception.SolidFireAPIException,
sfv.extend_volume,
testvol, 2)
@ -697,9 +697,9 @@ class SolidFireVolumeTestCase(test.TestCase):
def test_accept_transfer(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
testvol = {'project_id': 'testprjid',
'name': 'test_volume',
'size': 1,
@ -713,9 +713,9 @@ class SolidFireVolumeTestCase(test.TestCase):
def test_accept_transfer_volume_not_found_raises(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
testvol = {'project_id': 'testprjid',
'name': 'test_volume',
'size': 1,
@ -730,9 +730,9 @@ class SolidFireVolumeTestCase(test.TestCase):
def test_retype(self):
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
type_ref = volume_types.create(self.ctxt,
"type1", {"qos:minIOPS": "500",
"qos:burstIOPS": "2000",
@ -773,13 +773,13 @@ class SolidFireVolumeTestCase(test.TestCase):
def _fake_get_qos_spec(ctxt, spec_id):
return test_qos_spec
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.stubs.Set(volume_types, 'get_volume_type',
_fake_get_volume_type)
self.stubs.Set(qos_specs, 'get_qos_specs',
_fake_get_qos_spec)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(volume_types, 'get_volume_type',
_fake_get_volume_type)
self.mock_object(qos_specs, 'get_qos_specs',
_fake_get_qos_spec)
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
@ -800,9 +800,9 @@ class SolidFireVolumeTestCase(test.TestCase):
test_type, diff, host))
def test_update_cluster_status(self):
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
sfv._update_cluster_status()
self.assertEqual(99.0, sfv.cluster_stats['free_capacity_gb'])
@ -815,9 +815,9 @@ class SolidFireVolumeTestCase(test.TestCase):
'size': 1,
'id': 'a720b3c0-d1f0-11e1-9b23-0800200c9a66',
'created_at': timeutils.utcnow()}
self.stubs.Set(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
self.mock_object(solidfire.SolidFireDriver,
'_issue_api_request',
self.fake_issue_api_request)
sfv = solidfire.SolidFireDriver(configuration=self.configuration)
model_update = sfv.manage_existing(testvol, external_ref)
self.assertIsNotNone(model_update)

34
cinder/tests/unit/volume/drivers/test_tintri.py
View File

@ -65,23 +65,23 @@ class TintriDriverTestCase(test.TestCase):
return configuration
def fake_stubs(self):
self.stubs.Set(TClient, 'login', self.fake_login)
self.stubs.Set(TClient, 'logout', self.fake_logout)
self.stubs.Set(TClient, 'get_snapshot', self.fake_get_snapshot)
self.stubs.Set(TClient, 'get_image_snapshots_to_date',
self.fake_get_image_snapshots_to_date)
self.stubs.Set(TintriDriver, '_move_cloned_volume',
self.fake_move_cloned_volume)
self.stubs.Set(TintriDriver, '_get_provider_location',
self.fake_get_provider_location)
self.stubs.Set(TintriDriver, '_set_rw_permissions',
self.fake_set_rw_permissions)
self.stubs.Set(TintriDriver, '_is_volume_present',
self.fake_is_volume_present)
self.stubs.Set(TintriDriver, '_is_share_vol_compatible',
self.fake_is_share_vol_compatible)
self.stubs.Set(TintriDriver, '_is_file_size_equal',
self.fake_is_file_size_equal)
self.mock_object(TClient, 'login', self.fake_login)
self.mock_object(TClient, 'logout', self.fake_logout)
self.mock_object(TClient, 'get_snapshot', self.fake_get_snapshot)
self.mock_object(TClient, 'get_image_snapshots_to_date',
self.fake_get_image_snapshots_to_date)
self.mock_object(TintriDriver, '_move_cloned_volume',
self.fake_move_cloned_volume)
self.mock_object(TintriDriver, '_get_provider_location',
self.fake_get_provider_location)
self.mock_object(TintriDriver, '_set_rw_permissions',
self.fake_set_rw_permissions)
self.mock_object(TintriDriver, '_is_volume_present',
self.fake_is_volume_present)
self.mock_object(TintriDriver, '_is_share_vol_compatible',
self.fake_is_share_vol_compatible)
self.mock_object(TintriDriver, '_is_file_size_equal',
self.fake_is_file_size_equal)
def fake_login(self, user_name, password):
return 'session-id'