openstack/ironic
Code Issues Proposed changes

Merge "Fix Mock objects as specs"

Browse Source
This commit is contained in:
Zuul 2022-01-21 14:20:30 +00:00 committed by Gerrit Code Review
commit e0d69461de
14 changed files with 233 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
ironic/tests/unit/common/test_glance_service.py
View File

@ -147,10 +147,11 @@ class TestGlanceImageService(base.TestCase):
'os_hash_algo': None,
'os_hash_value': None,
}
with mock.patch.object(self.service, 'call', return_value=image,
autospec=True):
image_meta = self.service.show(image_id)
self.service.call.assert_called_once_with('get', image_id)
if not mock._is_instance_mock(self.service.call):
mock.patch.object(self.service, 'call', autospec=True).start()
self.service.call.return_value = image
image_meta = self.service.show(image_id)
self.service.call.assert_called_with('get', image_id)
self.assertEqual(expected, image_meta)
def test_show_makes_datetimes(self):
@ -175,13 +176,13 @@ class TestGlanceImageService(base.TestCase):
def test_show_raises_when_image_not_active(self):
image_id = uuidutils.generate_uuid()
image = self._make_fixture(name='image1', id=image_id, status="queued")
with mock.patch.object(self.service, 'call', return_value=image,
autospec=True):
self.assertRaises(exception.ImageUnacceptable,
self.service.show, image_id)
if not mock._is_instance_mock(self.service.call):
mock.patch.object(self.service, 'call', autospec=True).start()
self.service.call.return_value = image
self.assertRaises(exception.ImageUnacceptable,
self.service.show, image_id)
@mock.patch.object(tenacity, 'retry', autospec=True)
def test_download_with_retries(self, mock_retry):
def test_download_with_retries(self):
tries = [0]
class MyGlanceStubClient(stubs.StubGlanceClient):
@ -203,19 +204,20 @@ class TestGlanceImageService(base.TestCase):
image_id = uuidutils.generate_uuid()
writer = NullWriter()
# When retries are disabled, we should get an exception
self.config(num_retries=0, group='glance')
self.assertRaises(exception.GlanceConnectionFailed,
stub_service.download, image_id, writer)
with mock.patch.object(tenacity, 'retry', autospec=True) as mock_retry:
# When retries are disabled, we should get an exception
self.config(num_retries=0, group='glance')
self.assertRaises(exception.GlanceConnectionFailed,
stub_service.download, image_id, writer)
# Now lets enable retries. No exception should happen now.
self.config(num_retries=1, group='glance')
importlib.reload(image_service)
stub_service = image_service.GlanceImageService(stub_client,
stub_context)
tries = [0]
stub_service.download(image_id, writer)
mock_retry.assert_called_once()
# Now lets enable retries. No exception should happen now.
self.config(num_retries=1, group='glance')
importlib.reload(image_service)
stub_service = image_service.GlanceImageService(stub_client,
stub_context)
tries = [0]
stub_service.download(image_id, writer)
mock_retry.assert_called_once()
def test_download_no_data(self):
self.client.fake_wrapped = None

55
ironic/tests/unit/common/test_nova.py
View File

@ -73,19 +73,18 @@ class NovaApiTestCase(base.TestCase):
mock_adapter, mock_log):
server_ids = ['server-id-1', 'server-id-2']
nova_adapter = mock.Mock()
with mock.patch.object(nova_adapter, 'post',
autospec=True) as mock_post_event:
post_resp_mock = requests.Response()
mock_post_event = nova_adapter.post
post_resp_mock = requests.Response()
def json_func():
return nova_result
post_resp_mock.json = json_func
post_resp_mock.status_code = resp_status
mock_adapter.return_value = nova_adapter
mock_post_event.return_value = post_resp_mock
for server in server_ids:
result = self.api.power_update(self.ctx, server, 'power on')
self.assertEqual(exp_ret, result)
def json_func():
return nova_result
post_resp_mock.json = json_func
post_resp_mock.status_code = resp_status
mock_adapter.return_value = nova_adapter
mock_post_event.return_value = post_resp_mock
for server in server_ids:
result = self.api.power_update(self.ctx, server, 'power on')
self.assertEqual(exp_ret, result)
mock_adapter.assert_has_calls([mock.call(), mock.call()])
req_url = '/os-server-external-events'
@ -116,12 +115,11 @@ class NovaApiTestCase(base.TestCase):
@mock.patch.object(nova, '_get_nova_adapter', autospec=True)
def test_invalid_power_update(self, mock_adapter, mock_log):
nova_adapter = mock.Mock()
with mock.patch.object(nova_adapter, 'post',
autospec=True) as mock_post_event:
result = self.api.power_update(self.ctx, 'server', None)
self.assertFalse(result)
expected = ('Invalid Power State %s.', None)
mock_log.error.assert_called_once_with(*expected)
mock_post_event = nova_adapter.post
result = self.api.power_update(self.ctx, 'server', None)
self.assertFalse(result)
expected = ('Invalid Power State %s.', None)
mock_log.error.assert_called_once_with(*expected)
mock_adapter.assert_not_called()
mock_post_event.assert_not_called()
@ -162,19 +160,18 @@ class NovaApiTestCase(base.TestCase):
def test_power_update_invalid_reponse_format(self, nova_result,
mock_adapter, mock_log):
nova_adapter = mock.Mock()
with mock.patch.object(nova_adapter, 'post',
autospec=True) as mock_post_event:
post_resp_mock = requests.Response()
mock_post_event = nova_adapter.post
post_resp_mock = requests.Response()
def json_func():
return nova_result
def json_func():
return nova_result
post_resp_mock.json = json_func
post_resp_mock.status_code = 207
mock_adapter.return_value = nova_adapter
mock_post_event.return_value = post_resp_mock
result = self.api.power_update(self.ctx, 'server-id-1', 'power on')
self.assertFalse(result)
post_resp_mock.json = json_func
post_resp_mock.status_code = 207
mock_adapter.return_value = nova_adapter
mock_post_event.return_value = post_resp_mock
result = self.api.power_update(self.ctx, 'server-id-1', 'power on')
self.assertFalse(result)
mock_adapter.assert_has_calls([mock.call()])
req_url = '/os-server-external-events'

7
ironic/tests/unit/drivers/modules/drac/test_common.py
View File

@ -109,8 +109,11 @@ class DracCommonMethodsTestCase(test_utils.BaseDracTest):
self.assertRaises(exception.InvalidParameterValue,
drac_common.parse_driver_info, node)
@mock.patch.object(dracclient.client, 'DRACClient', autospec=True)
def test_get_drac_client(self, mock_dracclient):
def test_get_drac_client(self):
if not mock._is_instance_mock(dracclient.client):
mock.patch.object(dracclient.client, 'DRACClient',
autospec=True).start()
mock_dracclient = dracclient.client.DRACClient
expected_call = mock.call('1.2.3.4', 'admin', 'fake', 443, '/wsman',
'https')
node = obj_utils.create_test_node(self.context,

70
ironic/tests/unit/drivers/modules/ibmc/test_management.py
View File

@ -46,8 +46,10 @@ class IBMCManagementTestCase(base.IBMCTestCase):
task.driver.management.validate(task)
mock_parse_driver_info.assert_called_once_with(task.node)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_get_supported_boot_devices(self, connect_ibmc):
def test_get_supported_boot_devices(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock return value
_supported_boot_devices = list(mappings.GET_BOOT_DEVICE_MAP)
@ -60,12 +62,14 @@ class IBMCManagementTestCase(base.IBMCTestCase):
shared=True) as task:
supported_boot_devices = (
task.driver.management.get_supported_boot_devices(task))
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
expect = sorted(list(mappings.GET_BOOT_DEVICE_MAP.values()))
self.assertEqual(expect, sorted(supported_boot_devices))
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_set_boot_device(self, connect_ibmc):
def test_set_boot_device(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock return value
conn.system.set_boot_source.return_value = None
@ -90,7 +94,7 @@ class IBMCManagementTestCase(base.IBMCTestCase):
for (device, persistent) in data_source:
task.driver.management.set_boot_device(
task, device[0], persistent=persistent[0])
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.set_boot_source.assert_called_once_with(
device[1],
enabled=persistent[1])
@ -98,8 +102,10 @@ class IBMCManagementTestCase(base.IBMCTestCase):
connect_ibmc.reset_mock()
conn.system.set_boot_source.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_set_boot_device_fail(self, connect_ibmc):
def test_set_boot_device_fail(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock return value
conn.system.set_boot_source.side_effect = (
@ -112,13 +118,15 @@ class IBMCManagementTestCase(base.IBMCTestCase):
exception.IBMCError, 'set iBMC boot device',
task.driver.management.set_boot_device, task,
boot_devices.PXE)
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.set_boot_source.assert_called_once_with(
constants.BOOT_SOURCE_TARGET_PXE,
enabled=constants.BOOT_SOURCE_ENABLED_ONCE)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_get_boot_device(self, connect_ibmc):
def test_get_boot_device(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock return value
conn.system.get.return_value = mock.Mock(
@ -144,8 +152,10 @@ class IBMCManagementTestCase(base.IBMCTestCase):
self.assertEqual(list(mappings.SET_BOOT_MODE_MAP),
supported_boot_modes)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_set_boot_mode(self, connect_ibmc):
def test_set_boot_mode(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock system boot source override return value
conn.system.get.return_value = mock.Mock(
@ -168,7 +178,7 @@ class IBMCManagementTestCase(base.IBMCTestCase):
mode=ironic_boot_mode)
conn.system.get.assert_called_once()
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.set_boot_source.assert_called_once_with(
constants.BOOT_SOURCE_TARGET_PXE,
@ -180,8 +190,10 @@ class IBMCManagementTestCase(base.IBMCTestCase):
conn.system.set_boot_source.reset_mock()
conn.system.get.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_set_boot_mode_fail(self, connect_ibmc):
def test_set_boot_mode_fail(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock system boot source override return value
conn.system.get.return_value = mock.Mock(
@ -213,15 +225,17 @@ class IBMCManagementTestCase(base.IBMCTestCase):
mode=ibmc_boot_mode)
conn.system.get.assert_called_once()
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
# Reset
connect_ibmc.reset_mock()
conn.system.set_boot_source.reset_mock()
conn.system.get.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_get_boot_mode(self, connect_ibmc):
def test_get_boot_mode(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock system boot source override return value
conn.system.get.return_value = mock.Mock(
@ -236,7 +250,7 @@ class IBMCManagementTestCase(base.IBMCTestCase):
response = task.driver.management.get_boot_mode(task)
conn.system.get.assert_called_once()
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
expected = boot_modes.LEGACY_BIOS
self.assertEqual(expected, response)
@ -247,8 +261,10 @@ class IBMCManagementTestCase(base.IBMCTestCase):
self.assertRaises(NotImplementedError,
task.driver.management.get_sensors_data, task)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_inject_nmi(self, connect_ibmc):
def test_inject_nmi(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock system boot source override return value
conn.system.reset.return_value = None
@ -256,11 +272,13 @@ class IBMCManagementTestCase(base.IBMCTestCase):
shared=False) as task:
task.driver.management.inject_nmi(task)
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.reset.assert_called_once_with(constants.RESET_NMI)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
def test_inject_nmi_fail(self, connect_ibmc):
def test_inject_nmi_fail(self):
if not mock._is_instance_mock(ibmc_client):
mock.patch.object(ibmc_client, 'connect', autospec=True).start()
connect_ibmc = ibmc_client.connect
conn = self.mock_ibmc_conn(connect_ibmc)
# mock system boot source override return value
conn.system.reset.side_effect = (
@ -272,5 +290,5 @@ class IBMCManagementTestCase(base.IBMCTestCase):
exception.IBMCError, 'inject iBMC NMI',
task.driver.management.inject_nmi, task)
connect_ibmc.assert_called_once_with(**self.ibmc)
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.reset.assert_called_once_with(constants.RESET_NMI)

18
ironic/tests/unit/drivers/modules/ibmc/test_power.py
View File

@ -46,7 +46,7 @@ class IBMCPowerTestCase(base.IBMCTestCase):
task.driver.power.validate(task)
mock_parse_driver_info.assert_called_once_with(task.node)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_get_power_state(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
with task_manager.acquire(self.context, self.node.uuid,
@ -69,7 +69,7 @@ class IBMCPowerTestCase(base.IBMCTestCase):
conn.system.get.reset_mock()
connect_ibmc.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_set_power_state(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
with task_manager.acquire(self.context, self.node.uuid,
@ -102,7 +102,7 @@ class IBMCPowerTestCase(base.IBMCTestCase):
conn.system.get.reset_mock()
conn.system.reset.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_set_power_state_not_reached(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
with task_manager.acquire(self.context, self.node.uuid,
@ -137,7 +137,7 @@ class IBMCPowerTestCase(base.IBMCTestCase):
conn.system.get.reset_mock()
conn.system.reset.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_set_power_state_fail(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
@ -155,7 +155,7 @@ class IBMCPowerTestCase(base.IBMCTestCase):
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.reset.assert_called_once_with(constants.RESET_ON)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_set_power_state_timeout(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
with task_manager.acquire(self.context, self.node.uuid,
@ -189,7 +189,7 @@ class IBMCPowerTestCase(base.IBMCTestCase):
lambda *args, **kwargs: None)
class IBMCPowerRebootTestCase(base.IBMCTestCase):
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_reboot(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
expected_values = [
@ -222,7 +222,7 @@ class IBMCPowerRebootTestCase(base.IBMCTestCase):
conn.system.get.reset_mock()
conn.system.reset.reset_mock()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_reboot_not_reached(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
with task_manager.acquire(self.context, self.node.uuid,
@ -240,7 +240,7 @@ class IBMCPowerRebootTestCase(base.IBMCTestCase):
connect_ibmc.assert_called_with(**self.ibmc)
conn.system.reset.assert_called_once_with(constants.RESET_ON)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_reboot_fail(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
@ -263,7 +263,7 @@ class IBMCPowerRebootTestCase(base.IBMCTestCase):
conn.system.reset.assert_called_once_with(
constants.RESET_FORCE_RESTART)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_reboot_timeout(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)

14
ironic/tests/unit/drivers/modules/ibmc/test_raid.py
View File

@ -50,7 +50,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
self.node.target_raid_config = self.target_raid_config
self.node.save()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_create_configuration_without_delete(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.apply_raid_configuration.return_value = None
@ -66,7 +66,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
self.node.target_raid_config.get('logical_disks')
)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_create_configuration_with_delete(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.delete_all_raid_configuration.return_value = None
@ -84,7 +84,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
self.node.target_raid_config.get('logical_disks')
)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_create_configuration_without_nonroot(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.delete_all_raid_configuration.return_value = None
@ -102,7 +102,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
[{'size_gb': 200, 'raid_level': 0, 'is_root_volume': True}]
)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_create_configuration_without_root(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.delete_all_raid_configuration.return_value = None
@ -120,7 +120,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
[{'size_gb': 'MAX', 'raid_level': 5}]
)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_create_configuration_failed(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.delete_all_raid_configuration.return_value = None
@ -140,7 +140,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
self.node.target_raid_config.get('logical_disks')
)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_delete_configuration_success(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.delete_all_raid_configuration.return_value = None
@ -152,7 +152,7 @@ class IbmcRAIDTestCase(base.IBMCTestCase):
conn.system.storage.delete_all_raid_configuration.assert_called_once()
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_sync_delete_configuration_failed(self, connect_ibmc):
conn = self.mock_ibmc_conn(connect_ibmc)
conn.system.storage.delete_all_raid_configuration.side_effect = (

2
ironic/tests/unit/drivers/modules/ibmc/test_utils.py
View File

@ -132,7 +132,7 @@ class IBMCUtilsTestCase(base.IBMCTestCase):
"value2": "key2"
}, revert)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_handle_ibmc_exception_retry(self, connect_ibmc):
@utils.handle_ibmc_exception('get IBMC system')

4
ironic/tests/unit/drivers/modules/ibmc/test_vendor.py
View File

@ -44,7 +44,7 @@ class IBMCVendorTestCase(base.IBMCTestCase):
task.driver.power.validate(task)
mock_parse_driver_info.assert_called_once_with(task.node)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_list_boot_type_order(self, connect_ibmc):
# Mocks
conn = self.mock_ibmc_conn(connect_ibmc)
@ -61,7 +61,7 @@ class IBMCVendorTestCase(base.IBMCTestCase):
connect_ibmc.assert_called_once_with(**self.ibmc)
self.assertEqual(expected, seq)
@mock.patch.object(ibmc_client, 'connect', autospec=True)
@mock.patch.object(ibmc_client, 'connect', spec=object)
def test_list_raid_controller(self, connect_ibmc):
# Mocks
conn = self.mock_ibmc_conn(connect_ibmc)

14
ironic/tests/unit/drivers/modules/ilo/test_common.py
View File

@ -299,9 +299,10 @@ class IloValidateParametersTestCase(BaseIloTest):
class IloCommonMethodsTestCase(BaseIloTest):
@mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
@mock.patch.object(ilo_client, 'IloClient', spec_set=True,
autospec=True)
def _test_get_ilo_object(self, ilo_client_mock, isFile_mock, ca_file=None):
def _test_get_ilo_object(self, isFile_mock, ca_file=None):
if not mock._is_instance_mock(ilo_client):
mock.patch.object(ilo_client, 'IloClient', autospec=True).start()
ilo_client_mock = ilo_client.IloClient
self.info['client_timeout'] = 600
self.info['client_port'] = 4433
self.info['ilo_verify_ca'] = ca_file
@ -319,9 +320,10 @@ class IloCommonMethodsTestCase(BaseIloTest):
self.assertEqual('ilo_object', returned_ilo_object)
@mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
@mock.patch.object(ilo_client, 'IloClient', spec_set=True,
autospec=True)
def test_get_ilo_object_snmp(self, ilo_client_mock, isFile_mock):
def test_get_ilo_object_snmp(self, isFile_mock):
if not mock._is_instance_mock(ilo_client):
mock.patch.object(ilo_client, 'IloClient', autospec=True).start()
ilo_client_mock = ilo_client.IloClient
info = {'auth_user': 'user',
'auth_prot_pp': '1234',
'auth_priv_pp': '4321',

69
ironic/tests/unit/drivers/modules/ilo/test_firmware_processor.py
View File

@ -19,8 +19,8 @@ import io
from unittest import mock
from urllib import parse as urlparse
from oslo_utils import importutils
import proliantutils
from ironic.common import exception
from ironic.drivers.modules.ilo import common as ilo_common
@ -467,29 +467,33 @@ class FirmwareProcessorTestCase(base.TestCase):
urlparse_mock.return_value, self.fw_processor_fake.parsed_url)
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
def test__extract_fw_from_file_calls_process_firmware_image(
self, utils_mock, ilo_common_mock):
self, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
ilo_object_mock = ilo_common_mock.get_ilo_object.return_value
utils_mock.process_firmware_image.return_value = ('some_location',
True, True)
utils_mock.return_value = ('some_location', True, True)
# | WHEN |
ilo_fw_processor._extract_fw_from_file(node_mock, any_target_file)
# | THEN |
utils_mock.process_firmware_image.assert_called_once_with(
any_target_file, ilo_object_mock)
utils_mock.assert_called_once_with(any_target_file, ilo_object_mock)
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
def test__extract_fw_from_file_doesnt_upload_firmware(
self, utils_mock, ilo_common_mock):
self, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
utils_mock.process_firmware_image.return_value = (
utils_mock.return_value = (
'some_location/some_fw_file', False, True)
# | WHEN |
ilo_fw_processor._extract_fw_from_file(node_mock, any_target_file)
@ -497,15 +501,18 @@ class FirmwareProcessorTestCase(base.TestCase):
ilo_common_mock.copy_image_to_web_server.assert_not_called()
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
@mock.patch.object(ilo_fw_processor, '_remove_file_based_me',
autospec=True)
def test__extract_fw_from_file_sets_loc_obj_remove_to_file_if_no_upload(
self, _remove_mock, utils_mock, ilo_common_mock):
self, _remove_mock, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
utils_mock.process_firmware_image.return_value = (
utils_mock.return_value = (
'some_location/some_fw_file', False, True)
# | WHEN |
location_obj, is_different_file = (
@ -515,13 +522,16 @@ class FirmwareProcessorTestCase(base.TestCase):
_remove_mock.assert_called_once_with(location_obj)
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
def test__extract_fw_from_file_uploads_firmware_to_webserver(
self, utils_mock, ilo_common_mock):
self, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
utils_mock.process_firmware_image.return_value = (
utils_mock.return_value = (
'some_location/some_fw_file', True, True)
self.config(use_web_server_for_images=True, group='ilo')
# | WHEN |
@ -531,15 +541,18 @@ class FirmwareProcessorTestCase(base.TestCase):
'some_location/some_fw_file', 'some_fw_file')
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
@mock.patch.object(ilo_fw_processor, '_remove_webserver_based_me',
autospec=True)
def test__extract_fw_from_file_sets_loc_obj_remove_to_webserver(
self, _remove_mock, utils_mock, ilo_common_mock):
self, _remove_mock, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
utils_mock.process_firmware_image.return_value = (
utils_mock.return_value = (
'some_location/some_fw_file', True, True)
self.config(use_web_server_for_images=True, group='ilo')
# | WHEN |
@ -550,13 +563,16 @@ class FirmwareProcessorTestCase(base.TestCase):
_remove_mock.assert_called_once_with(location_obj)
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
def test__extract_fw_from_file_uploads_firmware_to_swift(
self, utils_mock, ilo_common_mock):
self, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
utils_mock.process_firmware_image.return_value = (
utils_mock.return_value = (
'some_location/some_fw_file', True, True)
self.config(use_web_server_for_images=False, group='ilo')
# | WHEN |
@ -566,15 +582,18 @@ class FirmwareProcessorTestCase(base.TestCase):
'some_location/some_fw_file', 'some_fw_file')
@mock.patch.object(ilo_fw_processor, 'ilo_common', autospec=True)
@mock.patch.object(ilo_fw_processor, 'proliantutils_utils', autospec=True)
@mock.patch.object(ilo_fw_processor, '_remove_swift_based_me',
autospec=True)
def test__extract_fw_from_file_sets_loc_obj_remove_to_swift(
self, _remove_mock, utils_mock, ilo_common_mock):
self, _remove_mock, ilo_common_mock):
# | GIVEN |
if not mock._is_instance_mock(proliantutils.utils):
mock.patch.object(proliantutils.utils, 'process_firmware_image',
autospec=True).start()
utils_mock = proliantutils.utils.process_firmware_image
node_mock = mock.MagicMock(uuid='fake_node_uuid')
any_target_file = 'any_target_file'
utils_mock.process_firmware_image.return_value = (
utils_mock.return_value = (
'some_location/some_fw_file', True, True)
self.config(use_web_server_for_images=False, group='ilo')
# | WHEN |

37
ironic/tests/unit/drivers/modules/irmc/test_bios.py
View File

@ -38,12 +38,12 @@ class IRMCBIOSTestCase(test_common.BaseIRMCTest):
task.driver.bios.validate(task)
parse_driver_info_mock.assert_called_once_with(task.node)
@mock.patch.object(irmc_bios.irmc.elcm, 'set_bios_configuration',
autospec=True)
@mock.patch.object(irmc_bios.irmc.elcm, 'get_bios_settings',
autospec=True)
def test_apply_configuration(self, get_bios_settings_mock,
set_bios_configuration_mock):
def test_apply_configuration(self):
if not mock._is_instance_mock(irmc_bios.irmc.elcm):
mock.patch.object(irmc_bios.irmc, 'elcm', autospec=True).start()
set_bios_configuration_mock = (
irmc_bios.irmc.elcm.set_bios_configuration)
get_bios_settings_mock = irmc_bios.irmc.elcm.get_bios_settings
settings = [{
"name": "launch_csm_enabled",
"value": True
@ -62,9 +62,11 @@ class IRMCBIOSTestCase(test_common.BaseIRMCTest):
set_bios_configuration_mock.assert_called_once_with(irmc_info,
settings)
@mock.patch.object(irmc_bios.irmc.elcm, 'set_bios_configuration',
autospec=True)
def test_apply_configuration_failed(self, set_bios_configuration_mock):
def test_apply_configuration_failed(self):
if not mock._is_instance_mock(irmc_bios.irmc.elcm):
mock.patch.object(irmc_bios.irmc, 'elcm', autospec=True).start()
set_bios_configuration_mock = (
irmc_bios.irmc.elcm.set_bios_configuration)
settings = [{
"name": "launch_csm_enabled",
"value": True
@ -95,11 +97,11 @@ class IRMCBIOSTestCase(test_common.BaseIRMCTest):
autospec=True)
@mock.patch.object(objects.BIOSSettingList, 'delete',
autospec=True)
@mock.patch.object(irmc_bios.irmc.elcm, 'get_bios_settings',
autospec=True)
def test_cache_bios_settings(self, get_bios_settings_mock,
delete_mock, save_mock, create_mock,
def test_cache_bios_settings(self, delete_mock, save_mock, create_mock,
sync_node_setting_mock):
if not mock._is_instance_mock(irmc_bios.irmc.elcm):
mock.patch.object(irmc_bios.irmc, 'elcm', autospec=True).start()
get_bios_settings_mock = irmc_bios.irmc.elcm.get_bios_settings
settings = [{
"name": "launch_csm_enabled",
"value": True
@ -133,7 +135,7 @@ class IRMCBIOSTestCase(test_common.BaseIRMCTest):
[]
)
task.driver.bios.cache_bios_settings(task)
get_bios_settings_mock.assert_called_once_with(irmc_info)
get_bios_settings_mock.assert_called_with(irmc_info)
sync_node_setting_mock.assert_called_once_with(task.context,
task.node.id,
settings)
@ -149,9 +151,10 @@ class IRMCBIOSTestCase(test_common.BaseIRMCTest):
delete_mock.assert_called_once_with(task.context, task.node.id,
delete_names)
@mock.patch.object(irmc_bios.irmc.elcm, 'get_bios_settings',
autospec=True)
def test_cache_bios_settings_failed(self, get_bios_settings_mock):
def test_cache_bios_settings_failed(self):
if not mock._is_instance_mock(irmc_bios.irmc.elcm):
mock.patch.object(irmc_bios.irmc, 'elcm', autospec=True).start()
get_bios_settings_mock = irmc_bios.irmc.elcm.get_bios_settings
irmc_bios.irmc.scci.SCCIError = Exception
get_bios_settings_mock.side_effect = Exception
with task_manager.acquire(self.context, self.node.uuid) as task:

12
ironic/tests/unit/drivers/modules/irmc/test_raid.py
View File

@ -20,6 +20,7 @@ from unittest import mock
from ironic.common import exception
from ironic.conductor import task_manager
from ironic import drivers as ironic_drivers
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules.irmc import raid
from ironic.tests.unit.drivers.modules.irmc import test_common
@ -603,12 +604,15 @@ class IRMCRaidConfigurationInternalMethodsTestCase(test_common.BaseIRMCTest):
disk, self.valid_disk_slots)
@mock.patch('ironic.common.raid.update_raid_info', autospec=True)
@mock.patch('ironic.drivers.modules.irmc.raid.client.elcm.'
'get_raid_adapter', autospec=True)
@mock.patch.object(deploy_utils, 'set_async_step_flags', autospec=True)
def test__commit_raid_config_with_logical_drives(
self, set_async_step_flags_mock,
get_raid_adapter_mock, update_raid_info_mock):
self, set_async_step_flags_mock, update_raid_info_mock):
if not mock._is_instance_mock(
ironic_drivers.modules.irmc.raid.client.elcm):
mock.patch.object(ironic_drivers.modules.irmc.raid.client,
'elcm', autospec=True).start()
get_raid_adapter_mock = (
ironic_drivers.modules.irmc.raid.client.elcm.get_raid_adapter)
get_raid_adapter_mock.return_value = {
"Server": {
"HWConfigurationIrmc": {

57
ironic/tests/unit/drivers/modules/test_agent_client.py
View File

@ -555,19 +555,21 @@ class TestAgentClient(base.TestCase):
verify=True)
def test_get_commands_status(self):
with mock.patch.object(self.client.session, 'get',
autospec=True) as mock_get:
res = mock.MagicMock(spec_set=['json'])
res.json.return_value = {'commands': []}
mock_get.return_value = res
self.assertEqual([], self.client.get_commands_status(self.node))
agent_url = self.node.driver_internal_info.get('agent_url')
mock_get.assert_called_once_with(
'%(agent_url)s/%(api_version)s/commands' % {
'agent_url': agent_url,
'api_version': CONF.agent.agent_api_version},
timeout=CONF.agent.command_timeout,
verify=True)
if not mock._is_instance_mock(self.client.session):
mock.patch.object(self.client.session, 'get',
autospec=True).start()
mock_get = self.client.session.get
res = mock.MagicMock(spec_set=['json'])
res.json.return_value = {'commands': []}
mock_get.return_value = res
self.assertEqual([], self.client.get_commands_status(self.node))
agent_url = self.node.driver_internal_info.get('agent_url')
mock_get.assert_called_once_with(
'%(agent_url)s/%(api_version)s/commands/' % {
'agent_url': agent_url,
'api_version': CONF.agent.agent_api_version},
verify=True, timeout=CONF.agent.command_timeout)
def test_get_commands_status_retries(self):
res = mock.MagicMock(spec_set=['json'])
@ -590,19 +592,22 @@ class TestAgentClient(base.TestCase):
def test_get_commands_status_verify(self, mock_exists):
self.node.driver_info['agent_verify_ca'] = '/path/to/agent.crt'
with mock.patch.object(self.client.session, 'get',
autospec=True) as mock_get:
res = mock.MagicMock(spec_set=['json'])
res.json.return_value = {'commands': []}
mock_get.return_value = res
self.assertEqual([], self.client.get_commands_status(self.node))
agent_url = self.node.driver_internal_info.get('agent_url')
mock_get.assert_called_once_with(
'%(agent_url)s/%(api_version)s/commands' % {
'agent_url': agent_url,
'api_version': CONF.agent.agent_api_version},
timeout=CONF.agent.command_timeout,
verify='/path/to/agent.crt')
if not mock._is_instance_mock(self.client.session):
mock.patch.object(self.client.session, 'get',
autospec=True).start()
mock_get = self.client.session.get
res = mock.MagicMock(spec_set=['json'])
res.json.return_value = {'commands': []}
mock_get.return_value = res
self.assertEqual([], self.client.get_commands_status(self.node))
agent_url = self.node.driver_internal_info.get('agent_url')
mock_get.assert_called_once_with(
'%(agent_url)s/%(api_version)s/commands/' % {
'agent_url': agent_url,
'api_version': CONF.agent.agent_api_version},
verify='/path/to/agent.crt',
timeout=CONF.agent.command_timeout)
def _test_install_bootloader(self, root_uuid, efi_system_part_uuid=None,
prep_boot_part_uuid=None):

6
ironic/tests/unit/drivers/modules/xclarity/test_common.py
View File

@ -95,8 +95,10 @@ class XClarityCommonTestCase(db_base.DbTestCase):
self.assertRaises(exception.InvalidParameterValue,
common.parse_driver_info, self.node)
@mock.patch.object(xclarity_client, 'Client', autospec=True)
def test_get_xclarity_client(self, mock_xclarityclient):
def test_get_xclarity_client(self):
if not mock._is_instance_mock(xclarity_client):
mock.patch.object(xclarity_client, 'Client', autospec=True).start()
mock_xclarityclient = xclarity_client.Client
expected_call = mock.call(ip='1.2.3.4', password='fake', port=443,
username='USERID')
common.get_xclarity_client(self.node)