"Fossies" - the Fresh Open Source Software Archive

Member "ironic-16.0.3/ironic/tests/unit/drivers/modules/ilo/test_management.py" (18 Jan 2021, 85812 Bytes) of package /linux/misc/openstack/ironic-16.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_management.py": 16.0.2_vs_16.0.3.

    1 # Copyright 2014 Hewlett-Packard Development Company, L.P.
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 # not use this file except in compliance with the License. You may obtain
    5 # a copy of the License at
    6 #
    7 #      http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 # License for the specific language governing permissions and limitations
   13 # under the License.
   14 
   15 """Test class for Management Interface used by iLO modules."""
   16 
   17 from unittest import mock
   18 
   19 import ddt
   20 from oslo_utils import importutils
   21 from oslo_utils import uuidutils
   22 
   23 from ironic.common import boot_devices
   24 from ironic.common import boot_modes
   25 from ironic.common import exception
   26 from ironic.common import states
   27 from ironic.conductor import task_manager
   28 from ironic.conductor import utils as manager_utils
   29 from ironic.drivers.modules import agent_base
   30 from ironic.drivers.modules import deploy_utils
   31 from ironic.drivers.modules.ilo import boot as ilo_boot
   32 from ironic.drivers.modules.ilo import common as ilo_common
   33 from ironic.drivers.modules.ilo import management as ilo_management
   34 from ironic.drivers.modules import ipmitool
   35 from ironic.drivers import utils as driver_utils
   36 from ironic.tests.unit.db import base as db_base
   37 from ironic.tests.unit.db import utils as db_utils
   38 from ironic.tests.unit.drivers.modules.ilo import test_common
   39 from ironic.tests.unit.objects import utils as obj_utils
   40 
   41 ilo_error = importutils.try_import('proliantutils.exception')
   42 
   43 INFO_DICT = db_utils.get_test_ilo_info()
   44 
   45 
   46 @ddt.ddt
   47 class IloManagementTestCase(test_common.BaseIloTest):
   48 
   49     def setUp(self):
   50         super(IloManagementTestCase, self).setUp()
   51         port_1 = obj_utils.create_test_port(
   52             self.context, node_id=self.node.id,
   53             address='11:22:33:44:55:66', uuid=uuidutils.generate_uuid())
   54         port_2 = obj_utils.create_test_port(
   55             self.context, node_id=self.node.id,
   56             address='11:22:33:44:55:67', uuid=uuidutils.generate_uuid())
   57         self.ports = [port_1, port_2]
   58 
   59     def test_get_properties(self):
   60         with task_manager.acquire(self.context, self.node.uuid,
   61                                   shared=False) as task:
   62             expected = ilo_management.MANAGEMENT_PROPERTIES
   63             self.assertEqual(expected,
   64                              task.driver.management.get_properties())
   65 
   66     @mock.patch.object(ilo_common, 'parse_driver_info', spec_set=True,
   67                        autospec=True)
   68     def test_validate(self, driver_info_mock):
   69         with task_manager.acquire(self.context, self.node.uuid,
   70                                   shared=False) as task:
   71             task.driver.management.validate(task)
   72             driver_info_mock.assert_called_once_with(task.node)
   73 
   74     def test_get_supported_boot_devices(self):
   75         with task_manager.acquire(self.context, self.node.uuid,
   76                                   shared=False) as task:
   77             expected = [boot_devices.PXE, boot_devices.DISK,
   78                         boot_devices.CDROM, boot_devices.ISCSIBOOT]
   79             self.assertEqual(
   80                 sorted(expected),
   81                 sorted(task.driver.management.
   82                        get_supported_boot_devices(task)))
   83 
   84     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
   85                        autospec=True)
   86     def test_get_boot_device_next_boot(self, get_ilo_object_mock):
   87         ilo_object_mock = get_ilo_object_mock.return_value
   88         ilo_object_mock.get_one_time_boot.return_value = 'CDROM'
   89         with task_manager.acquire(self.context, self.node.uuid,
   90                                   shared=False) as task:
   91             expected_device = boot_devices.CDROM
   92             expected_response = {'boot_device': expected_device,
   93                                  'persistent': False}
   94             self.assertEqual(expected_response,
   95                              task.driver.management.get_boot_device(task))
   96             ilo_object_mock.get_one_time_boot.assert_called_once_with()
   97 
   98     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
   99                        autospec=True)
  100     def test_get_boot_device_persistent(self, get_ilo_object_mock):
  101         ilo_mock = get_ilo_object_mock.return_value
  102         ilo_mock.get_one_time_boot.return_value = 'Normal'
  103         ilo_mock.get_persistent_boot_device.return_value = 'NETWORK'
  104 
  105         with task_manager.acquire(self.context, self.node.uuid,
  106                                   shared=False) as task:
  107             expected_device = boot_devices.PXE
  108             expected_response = {'boot_device': expected_device,
  109                                  'persistent': True}
  110             self.assertEqual(expected_response,
  111                              task.driver.management.get_boot_device(task))
  112             ilo_mock.get_one_time_boot.assert_called_once_with()
  113             ilo_mock.get_persistent_boot_device.assert_called_once_with()
  114 
  115     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  116                        autospec=True)
  117     def test_get_boot_device_fail(self, get_ilo_object_mock):
  118         ilo_mock_object = get_ilo_object_mock.return_value
  119         exc = ilo_error.IloError('error')
  120         ilo_mock_object.get_one_time_boot.side_effect = exc
  121 
  122         with task_manager.acquire(self.context, self.node.uuid,
  123                                   shared=False) as task:
  124             self.assertRaises(exception.IloOperationError,
  125                               task.driver.management.get_boot_device,
  126                               task)
  127         ilo_mock_object.get_one_time_boot.assert_called_once_with()
  128 
  129     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  130                        autospec=True)
  131     def test_get_boot_device_persistent_fail(self, get_ilo_object_mock):
  132         ilo_mock_object = get_ilo_object_mock.return_value
  133         ilo_mock_object.get_one_time_boot.return_value = 'Normal'
  134         exc = ilo_error.IloError('error')
  135         ilo_mock_object.get_persistent_boot_device.side_effect = exc
  136 
  137         with task_manager.acquire(self.context, self.node.uuid,
  138                                   shared=False) as task:
  139             self.assertRaises(exception.IloOperationError,
  140                               task.driver.management.get_boot_device,
  141                               task)
  142         ilo_mock_object.get_one_time_boot.assert_called_once_with()
  143         ilo_mock_object.get_persistent_boot_device.assert_called_once_with()
  144 
  145     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  146                        autospec=True)
  147     def test_set_boot_device_ok(self, get_ilo_object_mock):
  148         ilo_object_mock = get_ilo_object_mock.return_value
  149         with task_manager.acquire(self.context, self.node.uuid,
  150                                   shared=False) as task:
  151             task.driver.management.set_boot_device(task, boot_devices.CDROM,
  152                                                    False)
  153             get_ilo_object_mock.assert_called_once_with(task.node)
  154             ilo_object_mock.set_one_time_boot.assert_called_once_with('CDROM')
  155 
  156     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  157                        autospec=True)
  158     def test_set_boot_device_persistent_true(self, get_ilo_object_mock):
  159         ilo_mock = get_ilo_object_mock.return_value
  160         with task_manager.acquire(self.context, self.node.uuid,
  161                                   shared=False) as task:
  162             task.driver.management.set_boot_device(task, boot_devices.PXE,
  163                                                    True)
  164             get_ilo_object_mock.assert_called_once_with(task.node)
  165             ilo_mock.update_persistent_boot.assert_called_once_with(
  166                 ['NETWORK'])
  167 
  168     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  169                        autospec=True)
  170     def test_set_boot_device_fail(self, get_ilo_object_mock):
  171         ilo_mock_object = get_ilo_object_mock.return_value
  172         exc = ilo_error.IloError('error')
  173         ilo_mock_object.set_one_time_boot.side_effect = exc
  174 
  175         with task_manager.acquire(self.context, self.node.uuid,
  176                                   shared=False) as task:
  177             self.assertRaises(exception.IloOperationError,
  178                               task.driver.management.set_boot_device,
  179                               task, boot_devices.PXE)
  180         ilo_mock_object.set_one_time_boot.assert_called_once_with('NETWORK')
  181 
  182     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  183                        autospec=True)
  184     def test_set_boot_device_persistent_fail(self, get_ilo_object_mock):
  185         ilo_mock_object = get_ilo_object_mock.return_value
  186         exc = ilo_error.IloError('error')
  187         ilo_mock_object.update_persistent_boot.side_effect = exc
  188 
  189         with task_manager.acquire(self.context, self.node.uuid,
  190                                   shared=False) as task:
  191             self.assertRaises(exception.IloOperationError,
  192                               task.driver.management.set_boot_device,
  193                               task, boot_devices.PXE, True)
  194         ilo_mock_object.update_persistent_boot.assert_called_once_with(
  195             ['NETWORK'])
  196 
  197     def test_set_boot_device_invalid_device(self):
  198         with task_manager.acquire(self.context, self.node.uuid,
  199                                   shared=False) as task:
  200             self.assertRaises(exception.InvalidParameterValue,
  201                               task.driver.management.set_boot_device,
  202                               task, 'fake-device')
  203 
  204     @mock.patch.object(ilo_common, 'update_ipmi_properties', spec_set=True,
  205                        autospec=True)
  206     @mock.patch.object(ipmitool.IPMIManagement, 'get_sensors_data',
  207                        spec_set=True, autospec=True)
  208     def test_get_sensor_data(self, get_sensors_data_mock, update_ipmi_mock):
  209         with task_manager.acquire(self.context, self.node.uuid,
  210                                   shared=False) as task:
  211             task.driver.management.get_sensors_data(task)
  212             update_ipmi_mock.assert_called_once_with(task)
  213             get_sensors_data_mock.assert_called_once_with(mock.ANY, task)
  214 
  215     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  216                        autospec=True)
  217     def test__execute_ilo_step_ok(self, get_ilo_object_mock):
  218         ilo_mock = get_ilo_object_mock.return_value
  219         step_mock = getattr(ilo_mock, 'fake-step')
  220         ilo_management._execute_ilo_step(
  221             self.node, 'fake-step', 'args', kwarg='kwarg')
  222         step_mock.assert_called_once_with('args', kwarg='kwarg')
  223 
  224     @mock.patch.object(ilo_management, 'LOG', spec_set=True, autospec=True)
  225     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  226                        autospec=True)
  227     def test__execute_ilo_step_not_supported(self, get_ilo_object_mock,
  228                                              log_mock):
  229         ilo_mock = get_ilo_object_mock.return_value
  230         exc = ilo_error.IloCommandNotSupportedError("error")
  231         step_mock = getattr(ilo_mock, 'fake-step')
  232         step_mock.side_effect = exc
  233         ilo_management._execute_ilo_step(
  234             self.node, 'fake-step', 'args', kwarg='kwarg')
  235         step_mock.assert_called_once_with('args', kwarg='kwarg')
  236         self.assertTrue(log_mock.warning.called)
  237 
  238     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
  239                        autospec=True)
  240     def _test__execute_ilo_step_fail(self, get_ilo_object_mock):
  241         if self.node.clean_step:
  242             step = self.node.clean_step
  243             step_name = step['step']
  244             exept = exception.NodeCleaningFailure
  245         else:
  246             step = self.node.deploy_step
  247             step_name = step['step']
  248             exept = exception.InstanceDeployFailure
  249         ilo_mock = get_ilo_object_mock.return_value
  250         exc = ilo_error.IloError("error")
  251         step_mock = getattr(ilo_mock, step_name)
  252         step_mock.side_effect = exc
  253         self.assertRaises(exept,
  254                           ilo_management._execute_ilo_step,
  255                           self.node, step_name, 'args', kwarg='kwarg')
  256         step_mock.assert_called_once_with('args', kwarg='kwarg')
  257 
  258     def test__execute_ilo_step_fail_clean(self):
  259         self.node.clean_step = {'priority': 100, 'interface': 'management',
  260                                 'step': 'fake-step',
  261                                 'argsinfo': {}}
  262         self.node.save()
  263         self._test__execute_ilo_step_fail()
  264 
  265     def test__execute_ilo_step_fail_deploy(self):
  266         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  267                                  'step': 'fake-step',
  268                                  'argsinfo': {}}
  269         self.node.save()
  270         self._test__execute_ilo_step_fail()
  271 
  272     @mock.patch.object(deploy_utils, 'build_agent_options',
  273                        spec_set=True, autospec=True)
  274     @mock.patch.object(ilo_boot.IloVirtualMediaBoot, 'clean_up_ramdisk',
  275                        spec_set=True, autospec=True)
  276     @mock.patch.object(ilo_boot.IloVirtualMediaBoot, 'prepare_ramdisk',
  277                        spec_set=True, autospec=True)
  278     @mock.patch.object(ilo_management, '_execute_ilo_step',
  279                        spec_set=True, autospec=True)
  280     def test_reset_ilo(
  281             self, execute_step_mock, prepare_mock, cleanup_mock, build_mock):
  282         build_mock.return_value = {'a': 'b'}
  283         with task_manager.acquire(self.context, self.node.uuid,
  284                                   shared=False) as task:
  285             task.driver.management.reset_ilo(task)
  286             execute_step_mock.assert_called_once_with(task.node, 'reset_ilo')
  287             cleanup_mock.assert_called_once_with(mock.ANY, task)
  288             build_mock.assert_called_once_with(task.node)
  289             prepare_mock.assert_called_once_with(
  290                 mock.ANY, mock.ANY, {'a': 'b'})
  291 
  292     @mock.patch.object(ilo_management, '_execute_ilo_step',
  293                        spec_set=True, autospec=True)
  294     def test_reset_ilo_credential_ok(self, step_mock):
  295         info = self.node.driver_info
  296         info['ilo_change_password'] = "fake-password"
  297         self.node.driver_info = info
  298         self.node.save()
  299         with task_manager.acquire(self.context, self.node.uuid,
  300                                   shared=False) as task:
  301             task.driver.management.reset_ilo_credential(task)
  302             step_mock.assert_called_once_with(
  303                 task.node, 'reset_ilo_credential', 'fake-password')
  304             self.assertNotIn('ilo_change_password', task.node.driver_info)
  305             self.assertEqual('fake-password',
  306                              task.node.driver_info['ilo_password'])
  307 
  308     @mock.patch.object(ilo_management, '_execute_ilo_step',
  309                        spec_set=True, autospec=True)
  310     def test_reset_ilo_credential_pass_as_arg_ok(self, step_mock):
  311         with task_manager.acquire(self.context, self.node.uuid,
  312                                   shared=False) as task:
  313             task.driver.management.reset_ilo_credential(
  314                 task, change_password='fake-password')
  315             step_mock.assert_called_once_with(
  316                 task.node, 'reset_ilo_credential', 'fake-password')
  317             self.assertNotIn('ilo_change_password', task.node.driver_info)
  318             self.assertEqual('fake-password',
  319                              task.node.driver_info['ilo_password'])
  320 
  321     @mock.patch.object(ilo_management, 'LOG', spec_set=True, autospec=True)
  322     @mock.patch.object(ilo_management, '_execute_ilo_step',
  323                        spec_set=True, autospec=True)
  324     def test_reset_ilo_credential_no_password(self, step_mock,
  325                                               log_mock):
  326         with task_manager.acquire(self.context, self.node.uuid,
  327                                   shared=False) as task:
  328             task.driver.management.reset_ilo_credential(task)
  329             self.assertFalse(step_mock.called)
  330             self.assertTrue(log_mock.info.called)
  331 
  332     @mock.patch.object(ilo_management, '_execute_ilo_step',
  333                        spec_set=True, autospec=True)
  334     def test_reset_bios_to_default(self, step_mock):
  335         with task_manager.acquire(self.context, self.node.uuid,
  336                                   shared=False) as task:
  337             task.driver.management.reset_bios_to_default(task)
  338             step_mock.assert_called_once_with(task.node,
  339                                               'reset_bios_to_default')
  340 
  341     @mock.patch.object(ilo_management, '_execute_ilo_step',
  342                        spec_set=True, autospec=True)
  343     def test_reset_secure_boot_keys_to_default(self, step_mock):
  344         with task_manager.acquire(self.context, self.node.uuid,
  345                                   shared=False) as task:
  346             task.driver.management.reset_secure_boot_keys_to_default(task)
  347             step_mock.assert_called_once_with(task.node,
  348                                               'reset_secure_boot_keys')
  349 
  350     @mock.patch.object(ilo_management, '_execute_ilo_step',
  351                        spec_set=True, autospec=True)
  352     def test_clear_secure_boot_keys(self, step_mock):
  353         with task_manager.acquire(self.context, self.node.uuid,
  354                                   shared=False) as task:
  355             task.driver.management.clear_secure_boot_keys(task)
  356             step_mock.assert_called_once_with(task.node,
  357                                               'clear_secure_boot_keys')
  358 
  359     @mock.patch.object(ilo_management, '_execute_ilo_step',
  360                        spec_set=True, autospec=True)
  361     def test_activate_license(self, step_mock):
  362         with task_manager.acquire(self.context, self.node.uuid,
  363                                   shared=False) as task:
  364             activate_license_args = {
  365                 'ilo_license_key': 'XXXXX-YYYYY-ZZZZZ-XYZZZ-XXYYZ'}
  366             task.driver.management.activate_license(task,
  367                                                     **activate_license_args)
  368             step_mock.assert_called_once_with(
  369                 task.node, 'activate_license', 'XXXXX-YYYYY-ZZZZZ-XYZZZ-XXYYZ')
  370 
  371     @mock.patch.object(ilo_management, '_execute_ilo_step',
  372                        spec_set=True, autospec=True)
  373     def test_activate_license_no_or_invalid_format_license_key(
  374             self, step_mock):
  375         with task_manager.acquire(self.context, self.node.uuid,
  376                                   shared=False) as task:
  377             for license_key_value in (None, [], {}):
  378                 activate_license_args = {'ilo_license_key': license_key_value}
  379                 self.assertRaises(exception.InvalidParameterValue,
  380                                   task.driver.management.activate_license,
  381                                   task,
  382                                   **activate_license_args)
  383                 self.assertFalse(step_mock.called)
  384 
  385     @mock.patch.object(deploy_utils, 'build_agent_options',
  386                        spec_set=True, autospec=True)
  387     @mock.patch.object(ilo_boot.IloVirtualMediaBoot, 'clean_up_ramdisk',
  388                        spec_set=True, autospec=True)
  389     @mock.patch.object(ilo_boot.IloVirtualMediaBoot, 'prepare_ramdisk',
  390                        spec_set=True, autospec=True)
  391     @mock.patch.object(ilo_management, 'LOG', autospec=True)
  392     @mock.patch.object(ilo_management, '_execute_ilo_step',
  393                        spec_set=True, autospec=True)
  394     @mock.patch.object(ilo_management.firmware_processor, 'FirmwareProcessor',
  395                        spec_set=True, autospec=True)
  396     @mock.patch.object(ilo_common, 'remove_single_or_list_of_files',
  397                        spec_set=True, autospec=True)
  398     def _test_update_firmware_calls_step_foreach_url(
  399             self, remove_file_mock, FirmwareProcessor_mock, execute_step_mock,
  400             LOG_mock, prepare_mock, cleanup_mock, build_mock):
  401         if self.node.clean_step:
  402             step = self.node.clean_step
  403         else:
  404             step = self.node.deploy_step
  405         build_mock.return_value = {'a': 'b'}
  406         with task_manager.acquire(self.context, self.node.uuid,
  407                                   shared=False) as task:
  408             firmware_update_args = step['argsinfo']
  409             FirmwareProcessor_mock.return_value.process_fw_on.side_effect = [
  410                 ilo_management.firmware_processor.FirmwareImageLocation(
  411                     'fw_location_for_filepath', 'filepath'),
  412                 ilo_management.firmware_processor.FirmwareImageLocation(
  413                     'fw_location_for_httppath', 'httppath'),
  414                 ilo_management.firmware_processor.FirmwareImageLocation(
  415                     'fw_location_for_httpspath', 'httpspath'),
  416                 ilo_management.firmware_processor.FirmwareImageLocation(
  417                     'fw_location_for_swiftpath', 'swiftpath'),
  418                 ilo_management.firmware_processor.FirmwareImageLocation(
  419                     'fw_location_for_another_filepath', 'filepath2')
  420             ]
  421             task.driver.management.update_firmware(task,
  422                                                    **firmware_update_args)
  423             calls = [mock.call(task.node, 'update_firmware',
  424                                'fw_location_for_filepath', 'ilo'),
  425                      mock.call(task.node, 'update_firmware',
  426                                'fw_location_for_httppath', 'cpld'),
  427                      mock.call(task.node, 'update_firmware',
  428                                'fw_location_for_httpspath', 'power_pic'),
  429                      mock.call(task.node, 'update_firmware',
  430                                'fw_location_for_swiftpath', 'bios'),
  431                      mock.call(task.node, 'update_firmware',
  432                                'fw_location_for_another_filepath', 'chassis'),
  433                      ]
  434             execute_step_mock.assert_has_calls(calls)
  435             self.assertEqual(5, execute_step_mock.call_count)
  436             cleanup_mock.assert_called_once_with(mock.ANY, task)
  437             build_mock.assert_called_once_with(task.node)
  438             prepare_mock.assert_called_once_with(
  439                 mock.ANY, mock.ANY, {'a': 'b'})
  440 
  441     def test_update_firmware_calls_step_foreach_url_clean(self):
  442         firmware_images = [
  443             {
  444                 'url': 'file:///any_path',
  445                 'checksum': 'xxxx',
  446                 'component': 'ilo'
  447             },
  448             {
  449                 'url': 'http://any_url',
  450                 'checksum': 'xxxx',
  451                 'component': 'cpld'
  452             },
  453             {
  454                 'url': 'https://any_url',
  455                 'checksum': 'xxxx',
  456                 'component': 'power_pic'
  457             },
  458             {
  459                 'url': 'swift://container/object',
  460                 'checksum': 'xxxx',
  461                 'component': 'bios'
  462             },
  463             {
  464                 'url': 'file:///any_path',
  465                 'checksum': 'xxxx',
  466                 'component': 'chassis'
  467             }
  468         ]
  469         firmware_update_args = {'firmware_update_mode': 'ilo',
  470                                 'firmware_images': firmware_images}
  471         self.node.clean_step = {'priority': 100, 'interface': 'management',
  472                                 'step': 'update_firmware',
  473                                 'argsinfo': firmware_update_args}
  474         self.node.save()
  475         self._test_update_firmware_calls_step_foreach_url()
  476 
  477     def test_update_firmware_calls_step_foreach_url_deploy(self):
  478         firmware_images = [
  479             {
  480                 'url': 'file:///any_path',
  481                 'checksum': 'xxxx',
  482                 'component': 'ilo'
  483             },
  484             {
  485                 'url': 'http://any_url',
  486                 'checksum': 'xxxx',
  487                 'component': 'cpld'
  488             },
  489             {
  490                 'url': 'https://any_url',
  491                 'checksum': 'xxxx',
  492                 'component': 'power_pic'
  493             },
  494             {
  495                 'url': 'swift://container/object',
  496                 'checksum': 'xxxx',
  497                 'component': 'bios'
  498             },
  499             {
  500                 'url': 'file:///any_path',
  501                 'checksum': 'xxxx',
  502                 'component': 'chassis'
  503             }
  504         ]
  505         firmware_update_args = {'firmware_update_mode': 'ilo',
  506                                 'firmware_images': firmware_images}
  507         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  508                                  'step': 'update_firmware',
  509                                  'argsinfo': firmware_update_args}
  510         self.node.save()
  511         self._test_update_firmware_calls_step_foreach_url()
  512 
  513     def _test_update_firmware_invalid_update_mode_provided(self):
  514         if self.node.clean_step:
  515             step = self.node.clean_step
  516         else:
  517             step = self.node.deploy_step
  518         with task_manager.acquire(self.context, self.node.uuid,
  519                                   shared=False) as task:
  520             firmware_update_args = step['argsinfo']
  521             firmware_update_args = {'firmware_update_mode': 'invalid_mode',
  522                                     'firmware_images': None}
  523             self.assertRaises(exception.InvalidParameterValue,
  524                               task.driver.management.update_firmware,
  525                               task,
  526                               **firmware_update_args)
  527 
  528     def test_update_firmware_invalid_update_mode_provided_clean(self):
  529         firmware_update_args = {'firmware_update_mode': 'invalid_mode',
  530                                 'firmware_images': None}
  531         self.node.clean_step = {'priority': 100, 'interface': 'management',
  532                                 'step': 'update_firmware',
  533                                 'argsinfo': firmware_update_args}
  534         self.node.save()
  535         self._test_update_firmware_invalid_update_mode_provided()
  536 
  537     def test_update_firmware_invalid_update_mode_provided_deploy(self):
  538         firmware_update_args = {'firmware_update_mode': 'invalid_mode',
  539                                 'firmware_images': None}
  540         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  541                                  'step': 'update_firmware',
  542                                  'argsinfo': firmware_update_args}
  543         self.node.save()
  544         self._test_update_firmware_invalid_update_mode_provided()
  545 
  546     def _test_update_firmware_error_for_no_firmware_url(self):
  547         if self.node.clean_step:
  548             step = self.node.clean_step
  549         else:
  550             step = self.node.deploy_step
  551         with task_manager.acquire(self.context, self.node.uuid,
  552                                   shared=False) as task:
  553             firmware_update_args = step['argsinfo']
  554             firmware_update_args = {'firmware_update_mode': 'ilo',
  555                                     'firmware_images': []}
  556             self.assertRaises(exception.InvalidParameterValue,
  557                               task.driver.management.update_firmware,
  558                               task,
  559                               **firmware_update_args)
  560 
  561     def test_update_firmware_error_for_no_firmware_url_clean(self):
  562         firmware_update_args = {'firmware_update_mode': 'ilo',
  563                                 'firmware_images': []}
  564         self.node.clean_step = {'priority': 100, 'interface': 'management',
  565                                 'step': 'update_firmware',
  566                                 'argsinfo': firmware_update_args}
  567         self.node.save()
  568         self._test_update_firmware_error_for_no_firmware_url()
  569 
  570     def test_update_firmware_error_for_no_firmware_url_deploy(self):
  571         firmware_update_args = {'firmware_update_mode': 'ilo',
  572                                 'firmware_images': []}
  573         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  574                                  'step': 'update_firmware',
  575                                  'argsinfo': firmware_update_args}
  576         self.node.save()
  577         self._test_update_firmware_error_for_no_firmware_url()
  578 
  579     def _test_update_firmware_throws_error_for_invalid_component_type(self):
  580         if self.node.clean_step:
  581             step = self.node.clean_step
  582             exept = exception.NodeCleaningFailure
  583         else:
  584             step = self.node.deploy_step
  585             exept = exception.InstanceDeployFailure
  586         with task_manager.acquire(self.context, self.node.uuid,
  587                                   shared=False) as task:
  588             firmware_update_args = step['argsinfo']
  589             self.assertRaises(exept,
  590                               task.driver.management.update_firmware,
  591                               task,
  592                               **firmware_update_args)
  593 
  594     def test_update_firmware_error_for_invalid_component_type_clean(self):
  595         firmware_update_args = {'firmware_update_mode': 'ilo',
  596                                 'firmware_images': [
  597                                     {
  598                                         'url': 'any_valid_url',
  599                                         'checksum': 'xxxx',
  600                                         'component': 'xyz'
  601                                     }
  602                                 ]}
  603         self.node.clean_step = {'priority': 100, 'interface': 'management',
  604                                 'step': 'update_firmware',
  605                                 'argsinfo': firmware_update_args}
  606         self.node.save()
  607         self._test_update_firmware_throws_error_for_invalid_component_type()
  608 
  609     def test_update_firmware_error_for_invalid_component_type_deploy(self):
  610         firmware_update_args = {'firmware_update_mode': 'ilo',
  611                                 'firmware_images': [
  612                                     {
  613                                         'url': 'any_valid_url',
  614                                         'checksum': 'xxxx',
  615                                         'component': 'xyz'
  616                                     }
  617                                 ]}
  618         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  619                                  'step': 'update_firmware',
  620                                  'argsinfo': firmware_update_args}
  621         self.node.save()
  622         self._test_update_firmware_throws_error_for_invalid_component_type()
  623 
  624     @mock.patch.object(ilo_management, 'LOG', autospec=True)
  625     @mock.patch.object(ilo_management.firmware_processor.FirmwareProcessor,
  626                        'process_fw_on', spec_set=True, autospec=True)
  627     def _test_update_firmware_throws_error_for_checksum_validation_error(
  628             self, process_fw_on_mock, LOG_mock):
  629         if self.node.clean_step:
  630             step = self.node.clean_step
  631             exept = exception.NodeCleaningFailure
  632         else:
  633             step = self.node.deploy_step
  634             exept = exception.InstanceDeployFailure
  635         with task_manager.acquire(self.context, self.node.uuid,
  636                                   shared=False) as task:
  637             # | GIVEN |
  638             firmware_update_args = step['argsinfo']
  639             process_fw_on_mock.side_effect = exception.ImageRefValidationFailed
  640             # | WHEN & THEN |
  641             self.assertRaises(exept,
  642                               task.driver.management.update_firmware,
  643                               task,
  644                               **firmware_update_args)
  645 
  646     def test_update_firmware_error_for_checksum_validation_error_clean(self):
  647         firmware_update_args = {'firmware_update_mode': 'ilo',
  648                                 'firmware_images': [
  649                                     {
  650                                         'url': 'any_valid_url',
  651                                         'checksum': 'invalid_checksum',
  652                                         'component': 'bios'
  653                                     }
  654                                 ]}
  655         self.node.clean_step = {'priority': 100, 'interface': 'management',
  656                                 'step': 'update_firmware',
  657                                 'argsinfo': firmware_update_args}
  658         self.node.save()
  659         self._test_update_firmware_throws_error_for_checksum_validation_error()
  660 
  661     def test_update_firmware_error_for_checksum_validation_error_deploy(self):
  662         firmware_update_args = {'firmware_update_mode': 'ilo',
  663                                 'firmware_images': [
  664                                     {
  665                                         'url': 'any_valid_url',
  666                                         'checksum': 'invalid_checksum',
  667                                         'component': 'bios'
  668                                     }
  669                                 ]}
  670         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  671                                  'step': 'update_firmware',
  672                                  'argsinfo': firmware_update_args}
  673         self.node.save()
  674         self._test_update_firmware_throws_error_for_checksum_validation_error()
  675 
  676     @mock.patch.object(ilo_management, '_execute_ilo_step',
  677                        spec_set=True, autospec=True)
  678     @mock.patch.object(ilo_management.firmware_processor, 'FirmwareProcessor',
  679                        spec_set=True, autospec=True)
  680     def _test_update_firmware_doesnt_update_any_if_any_url_fails(
  681             self, FirmwareProcessor_mock, clean_step_mock):
  682         """update_firmware throws error for failure in processing any url
  683 
  684         update_firmware doesn't invoke firmware update of proliantutils
  685         for any url if processing on any firmware url fails.
  686         """
  687         if self.node.clean_step:
  688             step = self.node.clean_step
  689             exept = exception.NodeCleaningFailure
  690         else:
  691             step = self.node.deploy_step
  692             exept = exception.InstanceDeployFailure
  693         with task_manager.acquire(self.context, self.node.uuid,
  694                                   shared=False) as task:
  695             firmware_update_args = step['argsinfo']
  696             FirmwareProcessor_mock.return_value.process_fw_on.side_effect = [
  697                 ilo_management.firmware_processor.FirmwareImageLocation(
  698                     'extracted_firmware_url_of_any_valid_url', 'filename'),
  699                 exception.IronicException
  700             ]
  701             self.assertRaises(exept,
  702                               task.driver.management.update_firmware,
  703                               task,
  704                               **firmware_update_args)
  705             self.assertFalse(clean_step_mock.called)
  706 
  707     def test_update_firmware_doesnt_update_any_if_any_url_fails_clean(self):
  708         firmware_update_args = {'firmware_update_mode': 'ilo',
  709                                 'firmware_images': [
  710                                     {
  711                                         'url': 'any_valid_url',
  712                                         'checksum': 'xxxx',
  713                                         'component': 'ilo'
  714                                     },
  715                                     {
  716                                         'url': 'any_invalid_url',
  717                                         'checksum': 'xxxx',
  718                                         'component': 'bios'
  719                                     }]
  720                                 }
  721         self.node.clean_step = {'priority': 100, 'interface': 'management',
  722                                 'step': 'update_firmware',
  723                                 'argsinfo': firmware_update_args}
  724         self.node.save()
  725         self._test_update_firmware_doesnt_update_any_if_any_url_fails()
  726 
  727     def test_update_firmware_doesnt_update_any_if_any_url_fails_deploy(self):
  728         firmware_update_args = {'firmware_update_mode': 'ilo',
  729                                 'firmware_images': [
  730                                     {
  731                                         'url': 'any_valid_url',
  732                                         'checksum': 'xxxx',
  733                                         'component': 'ilo'
  734                                     },
  735                                     {
  736                                         'url': 'any_invalid_url',
  737                                         'checksum': 'xxxx',
  738                                         'component': 'bios'
  739                                     }]
  740                                 }
  741         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  742                                  'step': 'update_firmware',
  743                                  'argsinfo': firmware_update_args}
  744         self.node.save()
  745         self._test_update_firmware_doesnt_update_any_if_any_url_fails()
  746 
  747     @mock.patch.object(ilo_management, 'LOG', autospec=True)
  748     @mock.patch.object(ilo_management, '_execute_ilo_step',
  749                        spec_set=True, autospec=True)
  750     @mock.patch.object(ilo_management.firmware_processor, 'FirmwareProcessor',
  751                        spec_set=True, autospec=True)
  752     @mock.patch.object(ilo_management.firmware_processor.FirmwareImageLocation,
  753                        'remove', spec_set=True, autospec=True)
  754     def _test_update_firmware_cleans_all_files_if_exc_thrown(
  755             self, remove_mock, FirmwareProcessor_mock, clean_step_mock,
  756             LOG_mock):
  757         if self.node.clean_step:
  758             step = self.node.clean_step
  759             exept = exception.NodeCleaningFailure
  760         else:
  761             step = self.node.deploy_step
  762             exept = exception.InstanceDeployFailure
  763         with task_manager.acquire(self.context, self.node.uuid,
  764                                   shared=False) as task:
  765             firmware_update_args = step['argsinfo']
  766             fw_loc_obj_1 = (ilo_management.firmware_processor.
  767                             FirmwareImageLocation('extracted_firmware_url_1',
  768                                                   'filename_1'))
  769             fw_loc_obj_2 = (ilo_management.firmware_processor.
  770                             FirmwareImageLocation('extracted_firmware_url_2',
  771                                                   'filename_2'))
  772             FirmwareProcessor_mock.return_value.process_fw_on.side_effect = [
  773                 fw_loc_obj_1, fw_loc_obj_2
  774             ]
  775             clean_step_mock.side_effect = exept(
  776                 node=self.node.uuid, reason='ilo_exc')
  777             self.assertRaises(exept,
  778                               task.driver.management.update_firmware,
  779                               task,
  780                               **firmware_update_args)
  781             clean_step_mock.assert_called_once_with(
  782                 task.node, 'update_firmware',
  783                 'extracted_firmware_url_1', 'ilo')
  784             self.assertTrue(LOG_mock.error.called)
  785             remove_mock.assert_has_calls([mock.call(fw_loc_obj_1),
  786                                           mock.call(fw_loc_obj_2)])
  787 
  788     def test_update_firmware_cleans_all_files_if_exc_thrown_clean(self):
  789         firmware_update_args = {'firmware_update_mode': 'ilo',
  790                                 'firmware_images': [
  791                                     {
  792                                         'url': 'any_valid_url',
  793                                         'checksum': 'xxxx',
  794                                         'component': 'ilo'
  795                                     },
  796                                     {
  797                                         'url': 'any_invalid_url',
  798                                         'checksum': 'xxxx',
  799                                         'component': 'bios'
  800                                     }]
  801                                 }
  802         self.node.clean_step = {'priority': 100, 'interface': 'management',
  803                                 'step': 'update_firmware',
  804                                 'argsinfo': firmware_update_args}
  805         self.node.save()
  806         self._test_update_firmware_cleans_all_files_if_exc_thrown()
  807 
  808     def test_update_firmware_cleans_all_files_if_exc_thrown_deploy(self):
  809         firmware_update_args = {'firmware_update_mode': 'ilo',
  810                                 'firmware_images': [
  811                                     {
  812                                         'url': 'any_valid_url',
  813                                         'checksum': 'xxxx',
  814                                         'component': 'ilo'
  815                                     },
  816                                     {
  817                                         'url': 'any_invalid_url',
  818                                         'checksum': 'xxxx',
  819                                         'component': 'bios'
  820                                     }]
  821                                 }
  822         self.node.deploy_step = {'priority': 100, 'interface': 'management',
  823                                  'step': 'update_firmware',
  824                                  'argsinfo': firmware_update_args}
  825         self.node.save()
  826         self._test_update_firmware_cleans_all_files_if_exc_thrown()
  827 
  828     @mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
  829                        autospec=True)
  830     @mock.patch.object(agent_base, 'execute_step', autospec=True)
  831     def _test_write_firmware_sum_mode_with_component(
  832             self, execute_mock, attach_vmedia_mock, step_type='clean'):
  833         with task_manager.acquire(self.context, self.node.uuid,
  834                                   shared=False) as task:
  835             # | GIVEN |
  836             firmware_update_args = {
  837                 'url': 'http://any_url',
  838                 'checksum': 'xxxx',
  839                 'component': ['CP02345.scexe', 'CP02567.exe']}
  840             step = {'interface': 'management',
  841                     'args': firmware_update_args}
  842             if step_type == 'clean':
  843                 step['step'] = 'update_firmware_sum'
  844                 task.node.provision_state = states.CLEANING
  845                 execute_mock.return_value = states.CLEANWAIT
  846                 task.node.clean_step = step
  847                 func = task.driver.management.update_firmware_sum
  848                 exp_ret_state = states.CLEANWAIT
  849             else:
  850                 step['step'] = 'flash_firmware_sum'
  851                 task.node.provision_state = states.DEPLOYING
  852                 execute_mock.return_value = states.DEPLOYWAIT
  853                 task.node.deploy_step = step
  854                 func = task.driver.management.flash_firmware_sum
  855                 exp_ret_state = states.DEPLOYWAIT
  856             # | WHEN |
  857             return_value = func(task, **firmware_update_args)
  858             # | THEN |
  859             attach_vmedia_mock.assert_any_call(
  860                 task.node, 'CDROM', 'http://any_url')
  861             self.assertEqual(exp_ret_state, return_value)
  862             execute_mock.assert_called_once_with(task, step, step_type)
  863 
  864     def test_update_firmware_sum_mode_with_component(self):
  865         self._test_write_firmware_sum_mode_with_component(step_type='clean')
  866 
  867     def test_flash_firmware_sum_mode_with_component(self):
  868         self._test_write_firmware_sum_mode_with_component(step_type='deploy')
  869 
  870     @mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
  871                        autospec=True)
  872     @mock.patch.object(ilo_management.firmware_processor,
  873                        'get_swift_url', autospec=True)
  874     @mock.patch.object(agent_base, 'execute_step', autospec=True)
  875     def _test_write_firmware_sum_mode_swift_url(
  876             self, execute_mock, swift_url_mock, attach_vmedia_mock,
  877             step_type='clean'):
  878         with task_manager.acquire(self.context, self.node.uuid,
  879                                   shared=False) as task:
  880             # | GIVEN |
  881             swift_url_mock.return_value = "http://path-to-file"
  882             firmware_update_args = {
  883                 'url': 'swift://container/object',
  884                 'checksum': 'xxxx',
  885                 'components': ['CP02345.scexe', 'CP02567.exe']}
  886             step = {'interface': 'management',
  887                     'args': firmware_update_args}
  888             if step_type == 'clean':
  889                 task.node.provision_state = states.CLEANING
  890                 execute_mock.return_value = states.CLEANWAIT
  891                 step['step'] = 'update_firmware_sum',
  892                 task.node.clean_step = step
  893                 func = task.driver.management.update_firmware_sum
  894                 exp_ret_state = states.CLEANWAIT
  895                 args_data = task.node.clean_step['args']
  896             else:
  897                 task.node.provision_state = states.DEPLOYING
  898                 execute_mock.return_value = states.DEPLOYWAIT
  899                 step['step'] = 'flash_firmware_sum',
  900                 task.node.deploy_step = step
  901                 func = task.driver.management.flash_firmware_sum
  902                 exp_ret_state = states.DEPLOYWAIT
  903                 args_data = task.node.deploy_step['args']
  904             # | WHEN |
  905             return_value = func(task, **firmware_update_args)
  906             # | THEN |
  907             attach_vmedia_mock.assert_any_call(
  908                 task.node, 'CDROM', 'http://path-to-file')
  909             self.assertEqual(exp_ret_state, return_value)
  910             self.assertEqual(args_data['url'], "http://path-to-file")
  911 
  912     def test_write_firmware_sum_mode_swift_url_clean(self):
  913         self._test_write_firmware_sum_mode_swift_url(step_type='clean')
  914 
  915     def test_write_firmware_sum_mode_swift_url_deploy(self):
  916         self._test_write_firmware_sum_mode_swift_url(step_type='deploy')
  917 
  918     @mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
  919                        autospec=True)
  920     @mock.patch.object(agent_base, 'execute_step', autospec=True)
  921     def _test_write_firmware_sum_mode_without_component(
  922             self, execute_mock, attach_vmedia_mock, step_type='clean'):
  923         with task_manager.acquire(self.context, self.node.uuid,
  924                                   shared=False) as task:
  925             # | GIVEN |
  926             firmware_update_args = {
  927                 'url': 'any_valid_url',
  928                 'checksum': 'xxxx'}
  929             step = {'interface': 'management',
  930                     'args': firmware_update_args}
  931             if step_type == 'clean':
  932                 task.node.provision_state = states.CLEANING
  933                 execute_mock.return_value = states.CLEANWAIT
  934                 step['step'] = 'update_firmware_sum'
  935                 task.node.clean_step = step
  936                 func = task.driver.management.update_firmware_sum
  937                 exp_ret_state = states.CLEANWAIT
  938             else:
  939                 task.node.provision_state = states.DEPLOYING
  940                 execute_mock.return_value = states.DEPLOYWAIT
  941                 step['step'] = 'flash_firmware_sum'
  942                 task.node.deploy_step = step
  943                 func = task.driver.management.flash_firmware_sum
  944                 exp_ret_state = states.DEPLOYWAIT
  945             # | WHEN |
  946             return_value = func(task, **firmware_update_args)
  947             # | THEN |
  948             attach_vmedia_mock.assert_any_call(
  949                 task.node, 'CDROM', 'any_valid_url')
  950             self.assertEqual(exp_ret_state, return_value)
  951             execute_mock.assert_called_once_with(task, step, step_type)
  952 
  953     def test_write_firmware_sum_mode_without_component_clean(self):
  954         self._test_write_firmware_sum_mode_without_component(
  955             step_type='clean')
  956 
  957     def test_write_firmware_sum_mode_without_component_deploy(self):
  958         self._test_write_firmware_sum_mode_without_component(
  959             step_type='deploy')
  960 
  961     def _test_write_firmware_sum_mode_invalid_component(self,
  962                                                         step_type='clean'):
  963         # | GIVEN |
  964         firmware_update_args = {
  965             'url': 'any_valid_url',
  966             'checksum': 'xxxx',
  967             'components': ['CP02345']}
  968         with task_manager.acquire(self.context, self.node.uuid,
  969                                   shared=False) as task:
  970             # | WHEN & THEN |
  971             if step_type == 'clean':
  972                 func = task.driver.management.update_firmware_sum
  973             else:
  974                 func = task.driver.management.flash_firmware_sum
  975             self.assertRaises(exception.InvalidParameterValue,
  976                               func, task, **firmware_update_args)
  977 
  978     def test_write_firmware_sum_mode_invalid_component_clean(self):
  979         self._test_write_firmware_sum_mode_invalid_component(
  980             step_type='clean')
  981 
  982     def test_write_firmware_sum_mode_invalid_component_deploy(self):
  983         self._test_write_firmware_sum_mode_invalid_component(
  984             step_type='deploy')
  985 
  986     @mock.patch.object(driver_utils, 'store_ramdisk_logs', autospec=True)
  987     def _test__write_firmware_sum_final_with_logs(self, store_mock,
  988                                                   step_type='clean'):
  989         self.config(deploy_logs_collect='always', group='agent')
  990         firmware_update_args = {
  991             'url': 'any_valid_url',
  992             'checksum': 'xxxx'}
  993         step = {'interface': 'management',
  994                 'args': firmware_update_args}
  995         if step_type == 'clean':
  996             step['step'] = 'update_firmware_sum'
  997             node_state = states.CLEANWAIT
  998             command = {
  999                 'command_status': 'SUCCEEDED',
 1000                 'command_result': {
 1001                     'clean_result': {'Log Data': 'aaaabbbbcccdddd'},
 1002                     'clean_step': step,
 1003                 }
 1004             }
 1005             exp_label = 'update_firmware_sum'
 1006         else:
 1007             step['step'] = 'flash_firmware_sum'
 1008             node_state = states.DEPLOYWAIT
 1009             command = {
 1010                 'command_status': 'SUCCEEDED',
 1011                 'command_result': {
 1012                     'deploy_result': {'Log Data': 'aaaabbbbcccdddd'},
 1013                     'deploy_step': step,
 1014                 }
 1015             }
 1016             exp_label = 'flash_firmware_sum'
 1017         with task_manager.acquire(self.context, self.node.uuid,
 1018                                   shared=False) as task:
 1019             task.node.provision_state = node_state
 1020             task.driver.management._update_firmware_sum_final(
 1021                 task, command)
 1022             store_mock.assert_called_once_with(task.node, 'aaaabbbbcccdddd',
 1023                                                label=exp_label)
 1024 
 1025     def test__write_firmware_sum_final_with_logs_clean(self):
 1026         self._test__write_firmware_sum_final_with_logs(step_type='clean')
 1027 
 1028     def test__write_firmware_sum_final_with_logs_deploy(self):
 1029         self._test__write_firmware_sum_final_with_logs(step_type='deploy')
 1030 
 1031     @mock.patch.object(driver_utils, 'store_ramdisk_logs', autospec=True)
 1032     def _test__write_firmware_sum_final_without_logs(self, store_mock,
 1033                                                      step_type='clean'):
 1034         self.config(deploy_logs_collect='on_failure', group='agent')
 1035         firmware_update_args = {
 1036             'url': 'any_valid_url',
 1037             'checksum': 'xxxx'}
 1038         step = {'interface': 'management',
 1039                 'args': firmware_update_args}
 1040         if step_type == 'clean':
 1041             step['step'] = 'update_firmware_sum'
 1042             command = {
 1043                 'command_status': 'SUCCEEDED',
 1044                 'command_result': {
 1045                     'clean_result': {'Log Data': 'aaaabbbbcccdddd'},
 1046                     'clean_step': step,
 1047                 }
 1048             }
 1049         else:
 1050             step['step'] = 'flash_firmware_sum'
 1051             command = {
 1052                 'command_status': 'SUCCEEDED',
 1053                 'command_result': {
 1054                     'deploy_result': {'Log Data': 'aaaabbbbcccdddd'},
 1055                     'deploy_step': step,
 1056                 }
 1057             }
 1058         with task_manager.acquire(self.context, self.node.uuid,
 1059                                   shared=False) as task:
 1060             task.driver.management._update_firmware_sum_final(
 1061                 task, command)
 1062         self.assertFalse(store_mock.called)
 1063 
 1064     def test__write_firmware_sum_final_without_logs_clean(self):
 1065         self._test__write_firmware_sum_final_without_logs(step_type='clean')
 1066 
 1067     def test__write_firmware_sum_final_without_logs_deploy(self):
 1068         self._test__write_firmware_sum_final_without_logs(step_type='deploy')
 1069 
 1070     @mock.patch.object(ilo_management, 'LOG', spec_set=True, autospec=True)
 1071     @mock.patch.object(driver_utils, 'store_ramdisk_logs', autospec=True)
 1072     def _test__write_firmware_sum_final_swift_error(self, store_mock,
 1073                                                     log_mock,
 1074                                                     step_type='clean'):
 1075         self.config(deploy_logs_collect='always', group='agent')
 1076         firmware_update_args = {
 1077             'url': 'any_valid_url',
 1078             'checksum': 'xxxx'}
 1079         step = {'interface': 'management',
 1080                 'args': firmware_update_args}
 1081         if step_type == 'clean':
 1082             step['step'] = 'update_firmware_sum'
 1083             node_state = states.CLEANWAIT
 1084             command = {
 1085                 'command_status': 'SUCCEEDED',
 1086                 'command_result': {
 1087                     'clean_result': {'Log Data': 'aaaabbbbcccdddd'},
 1088                     'clean_step': step,
 1089                 }
 1090             }
 1091         else:
 1092             step['step'] = 'flash_firmware_sum'
 1093             node_state = states.DEPLOYWAIT
 1094             command = {
 1095                 'command_status': 'SUCCEEDED',
 1096                 'command_result': {
 1097                     'deploy_result': {'Log Data': 'aaaabbbbcccdddd'},
 1098                     'deploy_step': step,
 1099                 }
 1100             }
 1101         store_mock.side_effect = exception.SwiftOperationError('Error')
 1102 
 1103         with task_manager.acquire(self.context, self.node.uuid,
 1104                                   shared=False) as task:
 1105             task.node.provision_state = node_state
 1106             task.driver.management._update_firmware_sum_final(
 1107                 task, command)
 1108         self.assertTrue(log_mock.error.called)
 1109 
 1110     def test__write_firmware_sum_final_swift_error_clean(self):
 1111         self._test__write_firmware_sum_final_swift_error(step_type='clean')
 1112 
 1113     def test__write_firmware_sum_final_swift_error_deploy(self):
 1114         self._test__write_firmware_sum_final_swift_error(step_type='deploy')
 1115 
 1116     @mock.patch.object(ilo_management, 'LOG', spec_set=True, autospec=True)
 1117     @mock.patch.object(driver_utils, 'store_ramdisk_logs', autospec=True)
 1118     def _test__write_firmware_sum_final_environment_error(self, store_mock,
 1119                                                           log_mock,
 1120                                                           step_type='clean'):
 1121         self.config(deploy_logs_collect='always', group='agent')
 1122         firmware_update_args = {
 1123             'url': 'any_valid_url',
 1124             'checksum': 'xxxx'}
 1125         step = {'interface': 'management',
 1126                 'args': firmware_update_args}
 1127         if step_type == 'clean':
 1128             step['step'] = 'update_firmware_sum'
 1129             node_state = states.CLEANWAIT
 1130             command = {
 1131                 'command_status': 'SUCCEEDED',
 1132                 'command_result': {
 1133                     'clean_result': {'Log Data': 'aaaabbbbcccdddd'},
 1134                     'clean_step': step,
 1135                 }
 1136             }
 1137         else:
 1138             step['step'] = 'flash_firmware_sum'
 1139             node_state = states.DEPLOYWAIT
 1140             command = {
 1141                 'command_status': 'SUCCEEDED',
 1142                 'command_result': {
 1143                     'deploy_result': {'Log Data': 'aaaabbbbcccdddd'},
 1144                     'deploy_step': step,
 1145                 }
 1146             }
 1147         store_mock.side_effect = EnvironmentError('Error')
 1148 
 1149         with task_manager.acquire(self.context, self.node.uuid,
 1150                                   shared=False) as task:
 1151             task.node.provision_state = node_state
 1152             task.driver.management._update_firmware_sum_final(
 1153                 task, command)
 1154         self.assertTrue(log_mock.exception.called)
 1155 
 1156     def test__write_firmware_sum_final_environment_error_clean(self):
 1157         self._test__write_firmware_sum_final_environment_error(
 1158             step_type='clean')
 1159 
 1160     def test__write_firmware_sum_final_environment_error_deploy(self):
 1161         self._test__write_firmware_sum_final_environment_error(
 1162             step_type='deploy')
 1163 
 1164     @mock.patch.object(ilo_management, 'LOG', spec_set=True, autospec=True)
 1165     @mock.patch.object(driver_utils, 'store_ramdisk_logs', autospec=True)
 1166     def _test__write_firmware_sum_final_unknown_exception(self, store_mock,
 1167                                                           log_mock,
 1168                                                           step_type='clean'):
 1169         self.config(deploy_logs_collect='always', group='agent')
 1170         firmware_update_args = {
 1171             'url': 'any_valid_url',
 1172             'checksum': 'xxxx'}
 1173         step = {'interface': 'management',
 1174                 'args': firmware_update_args}
 1175         if step_type == 'clean':
 1176             step['step'] = 'update_firmware_sum'
 1177             node_state = states.CLEANWAIT
 1178             command = {
 1179                 'command_status': 'SUCCEEDED',
 1180                 'command_result': {
 1181                     'clean_result': {'Log Data': 'aaaabbbbcccdddd'},
 1182                     'clean_step': step,
 1183                 }
 1184             }
 1185         else:
 1186             step['step'] = 'flash_firmware_sum'
 1187             node_state = states.DEPLOYWAIT
 1188             command = {
 1189                 'command_status': 'SUCCEEDED',
 1190                 'command_result': {
 1191                     'deploy_result': {'Log Data': 'aaaabbbbcccdddd'},
 1192                     'deploy_step': step,
 1193                 }
 1194             }
 1195         store_mock.side_effect = Exception('Error')
 1196 
 1197         with task_manager.acquire(self.context, self.node.uuid,
 1198                                   shared=False) as task:
 1199             task.node.provision_state = node_state
 1200             task.driver.management._update_firmware_sum_final(
 1201                 task, command)
 1202         self.assertTrue(log_mock.exception.called)
 1203 
 1204     def test__write_firmware_sum_final_unknown_exception_clean(self):
 1205         self._test__write_firmware_sum_final_unknown_exception(
 1206             step_type='clean')
 1207 
 1208     def test__write_firmware_sum_final_unknown_exception_deploy(self):
 1209         self._test__write_firmware_sum_final_unknown_exception(
 1210             step_type='deploy')
 1211 
 1212     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1213                        autospec=True)
 1214     def test_set_iscsi_boot_target_with_auth(self, get_ilo_object_mock):
 1215         with task_manager.acquire(self.context, self.node.uuid,
 1216                                   shared=False) as task:
 1217             vol_id = uuidutils.generate_uuid()
 1218             obj_utils.create_test_volume_target(
 1219                 self.context, node_id=self.node.id, volume_type='iscsi',
 1220                 boot_index=0, volume_id='1234', uuid=vol_id,
 1221                 properties={'target_lun': 0,
 1222                             'target_portal': 'fake_host:3260',
 1223                             'target_iqn': 'fake_iqn',
 1224                             'auth_username': 'fake_username',
 1225                             'auth_password': 'fake_password'})
 1226             driver_internal_info = task.node.driver_internal_info
 1227             driver_internal_info['boot_from_volume'] = vol_id
 1228             task.node.driver_internal_info = driver_internal_info
 1229             task.node.save()
 1230             ilo_object_mock = get_ilo_object_mock.return_value
 1231             task.driver.management.set_iscsi_boot_target(task)
 1232             ilo_object_mock.set_iscsi_info.assert_called_once_with(
 1233                 'fake_iqn', 0, 'fake_host', '3260',
 1234                 auth_method='CHAP', username='fake_username',
 1235                 password='fake_password',
 1236                 macs=['11:22:33:44:55:66', '11:22:33:44:55:67'])
 1237 
 1238     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1239                        autospec=True)
 1240     def test_set_iscsi_boot_target_without_auth(self, get_ilo_object_mock):
 1241         with task_manager.acquire(self.context, self.node.uuid,
 1242                                   shared=False) as task:
 1243             vol_id = uuidutils.generate_uuid()
 1244             obj_utils.create_test_volume_target(
 1245                 self.context, node_id=self.node.id, volume_type='iscsi',
 1246                 boot_index=0, volume_id='1234', uuid=vol_id,
 1247                 properties={'target_lun': 0,
 1248                             'target_portal': 'fake_host:3260',
 1249                             'target_iqn': 'fake_iqn'})
 1250             driver_internal_info = task.node.driver_internal_info
 1251             driver_internal_info['boot_from_volume'] = vol_id
 1252             task.node.driver_internal_info = driver_internal_info
 1253             task.node.save()
 1254             ilo_object_mock = get_ilo_object_mock.return_value
 1255             task.driver.management.set_iscsi_boot_target(task)
 1256             ilo_object_mock.set_iscsi_info.assert_called_once_with(
 1257                 'fake_iqn', 0, 'fake_host', '3260', auth_method=None,
 1258                 password=None, username=None,
 1259                 macs=['11:22:33:44:55:66', '11:22:33:44:55:67'])
 1260 
 1261     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1262                        autospec=True)
 1263     def test_set_iscsi_boot_target_failed(self, get_ilo_object_mock):
 1264         with task_manager.acquire(self.context, self.node.uuid,
 1265                                   shared=False) as task:
 1266             vol_id = uuidutils.generate_uuid()
 1267             obj_utils.create_test_volume_target(
 1268                 self.context, node_id=self.node.id, volume_type='iscsi',
 1269                 boot_index=0, volume_id='1234', uuid=vol_id,
 1270                 properties={'target_lun': 0,
 1271                             'target_portal': 'fake_host:3260',
 1272                             'target_iqn': 'fake_iqn',
 1273                             'auth_username': 'fake_username',
 1274                             'auth_password': 'fake_password'})
 1275             driver_internal_info = task.node.driver_internal_info
 1276             driver_internal_info['boot_from_volume'] = vol_id
 1277             task.node.driver_internal_info = driver_internal_info
 1278             task.node.save()
 1279             ilo_object_mock = get_ilo_object_mock.return_value
 1280             ilo_object_mock.set_iscsi_info.side_effect = (
 1281                 ilo_error.IloError('error'))
 1282             self.assertRaises(exception.IloOperationError,
 1283                               task.driver.management.set_iscsi_boot_target,
 1284                               task)
 1285 
 1286     def test_set_iscsi_boot_target_missed_properties(self):
 1287         with task_manager.acquire(self.context, self.node.uuid,
 1288                                   shared=False) as task:
 1289             vol_id = uuidutils.generate_uuid()
 1290             obj_utils.create_test_volume_target(
 1291                 self.context, node_id=self.node.id, volume_type='iscsi',
 1292                 boot_index=0, volume_id='1234', uuid=vol_id,
 1293                 properties={'target_iqn': 'fake_iqn',
 1294                             'auth_username': 'fake_username',
 1295                             'auth_password': 'fake_password'})
 1296             driver_internal_info = task.node.driver_internal_info
 1297             driver_internal_info['boot_from_volume'] = vol_id
 1298             task.node.driver_internal_info = driver_internal_info
 1299             task.node.save()
 1300             self.assertRaises(exception.MissingParameterValue,
 1301                               task.driver.management.set_iscsi_boot_target,
 1302                               task)
 1303 
 1304     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1305                        autospec=True)
 1306     def test_set_iscsi_boot_target_in_bios(self, get_ilo_object_mock):
 1307         with task_manager.acquire(self.context, self.node.uuid,
 1308                                   shared=False) as task:
 1309             vol_id = uuidutils.generate_uuid()
 1310             obj_utils.create_test_volume_target(
 1311                 self.context, node_id=self.node.id, volume_type='iscsi',
 1312                 boot_index=0, volume_id='1234', uuid=vol_id,
 1313                 properties={'target_lun': 0,
 1314                             'target_portal': 'fake_host:3260',
 1315                             'target_iqn': 'fake_iqn',
 1316                             'auth_username': 'fake_username',
 1317                             'auth_password': 'fake_password'})
 1318             driver_internal_info = task.node.driver_internal_info
 1319             driver_internal_info['boot_from_volume'] = vol_id
 1320             task.node.driver_internal_info = driver_internal_info
 1321             task.node.save()
 1322             ilo_object_mock = get_ilo_object_mock.return_value
 1323             ilo_object_mock.set_iscsi_info.side_effect = (
 1324                 ilo_error.IloCommandNotSupportedInBiosError('error'))
 1325             self.assertRaises(exception.IloOperationNotSupported,
 1326                               task.driver.management.set_iscsi_boot_target,
 1327                               task)
 1328 
 1329     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1330                        autospec=True)
 1331     def test_clear_iscsi_boot_target(self, get_ilo_object_mock):
 1332         with task_manager.acquire(self.context, self.node.uuid,
 1333                                   shared=False) as task:
 1334             ilo_object_mock = get_ilo_object_mock.return_value
 1335 
 1336             task.driver.management.clear_iscsi_boot_target(task)
 1337             ilo_object_mock.unset_iscsi_info.assert_called_once_with(
 1338                 macs=['11:22:33:44:55:66', '11:22:33:44:55:67'])
 1339 
 1340     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1341                        autospec=True)
 1342     def test_clear_iscsi_boot_target_failed(self, get_ilo_object_mock):
 1343         with task_manager.acquire(self.context, self.node.uuid,
 1344                                   shared=False) as task:
 1345             ilo_object_mock = get_ilo_object_mock.return_value
 1346             ilo_object_mock.unset_iscsi_info.side_effect = (
 1347                 ilo_error.IloError('error'))
 1348             self.assertRaises(exception.IloOperationError,
 1349                               task.driver.management.clear_iscsi_boot_target,
 1350                               task)
 1351 
 1352     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1353                        autospec=True)
 1354     def test_clear_iscsi_boot_target_in_bios(self, get_ilo_object_mock):
 1355         with task_manager.acquire(self.context, self.node.uuid,
 1356                                   shared=False) as task:
 1357             ilo_object_mock = get_ilo_object_mock.return_value
 1358             ilo_object_mock.unset_iscsi_info.side_effect = (
 1359                 ilo_error.IloCommandNotSupportedInBiosError('error'))
 1360             self.assertRaises(exception.IloOperationNotSupported,
 1361                               task.driver.management.clear_iscsi_boot_target,
 1362                               task)
 1363 
 1364     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1365                        autospec=True)
 1366     def test_inject_nmi(self, get_ilo_object_mock):
 1367         with task_manager.acquire(self.context, self.node.uuid,
 1368                                   shared=False) as task:
 1369             ilo_object_mock = get_ilo_object_mock.return_value
 1370 
 1371             task.driver.management.inject_nmi(task)
 1372             ilo_object_mock.inject_nmi.assert_called_once()
 1373 
 1374     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1375                        autospec=True)
 1376     def test_inject_nmi_failed(self, get_ilo_object_mock):
 1377         with task_manager.acquire(self.context, self.node.uuid,
 1378                                   shared=False) as task:
 1379             ilo_object_mock = get_ilo_object_mock.return_value
 1380             ilo_object_mock.inject_nmi.side_effect = (
 1381                 ilo_error.IloError('error'))
 1382             self.assertRaises(exception.IloOperationError,
 1383                               task.driver.management.inject_nmi,
 1384                               task)
 1385 
 1386     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1387                        autospec=True)
 1388     def test_inject_nmi_not_supported(self, get_ilo_object_mock):
 1389         with task_manager.acquire(self.context, self.node.uuid,
 1390                                   shared=False) as task:
 1391             ilo_object_mock = get_ilo_object_mock.return_value
 1392             ilo_object_mock.inject_nmi.side_effect = (
 1393                 ilo_error.IloCommandNotSupportedError('error'))
 1394             self.assertRaises(exception.IloOperationNotSupported,
 1395                               task.driver.management.inject_nmi,
 1396                               task)
 1397 
 1398     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1399                        autospec=True)
 1400     @ddt.data((ilo_common.SUPPORTED_BOOT_MODE_LEGACY_BIOS_ONLY,
 1401                ['bios']),
 1402               (ilo_common.SUPPORTED_BOOT_MODE_UEFI_ONLY,
 1403                ['uefi']),
 1404               (ilo_common.SUPPORTED_BOOT_MODE_LEGACY_BIOS_AND_UEFI,
 1405                ['uefi', 'bios']))
 1406     @ddt.unpack
 1407     def test_get_supported_boot_modes(self, boot_modes_val,
 1408                                       exp_boot_modes,
 1409                                       get_ilo_object_mock):
 1410         ilo_object_mock = get_ilo_object_mock.return_value
 1411         ilo_object_mock.get_supported_boot_mode.return_value = boot_modes_val
 1412         with task_manager.acquire(self.context, self.node.uuid,
 1413                                   shared=True) as task:
 1414             supported_boot_modes = (
 1415                 task.driver.management.get_supported_boot_modes(task))
 1416             self.assertEqual(exp_boot_modes, supported_boot_modes)
 1417 
 1418     @mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
 1419                        autospec=True)
 1420     @mock.patch.object(ilo_management.IloManagement,
 1421                        'get_supported_boot_modes',
 1422                        spec_set=True, autospec=True)
 1423     def test_set_boot_mode(self, supp_boot_modes_mock,
 1424                            set_boot_mode_mock):
 1425         with task_manager.acquire(self.context, self.node.uuid,
 1426                                   shared=False) as task:
 1427             exp_boot_modes = [boot_modes.UEFI, boot_modes.LEGACY_BIOS]
 1428             supp_boot_modes_mock.return_value = exp_boot_modes
 1429 
 1430             for mode in exp_boot_modes:
 1431                 task.driver.management.set_boot_mode(task, mode=mode)
 1432                 supp_boot_modes_mock.assert_called_once_with(mock.ANY, task)
 1433                 set_boot_mode_mock.assert_called_once_with(task.node, mode)
 1434                 set_boot_mode_mock.reset_mock()
 1435                 supp_boot_modes_mock.reset_mock()
 1436 
 1437     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1438                        autospec=True)
 1439     @mock.patch.object(ilo_management.IloManagement,
 1440                        'get_supported_boot_modes',
 1441                        spec_set=True, autospec=True)
 1442     def test_set_boot_mode_fail(self, supp_boot_modes_mock,
 1443                                 get_ilo_object_mock):
 1444         ilo_mock_obj = get_ilo_object_mock.return_value
 1445         ilo_mock_obj.get_pending_boot_mode.return_value = 'legacy'
 1446         exc = ilo_error.IloError('error')
 1447         ilo_mock_obj.set_pending_boot_mode.side_effect = exc
 1448         exp_boot_modes = [boot_modes.UEFI, boot_modes.LEGACY_BIOS]
 1449         supp_boot_modes_mock.return_value = exp_boot_modes
 1450         with task_manager.acquire(self.context, self.node.uuid,
 1451                                   shared=False) as task:
 1452             self.assertRaisesRegex(
 1453                 exception.IloOperationError, 'uefi as boot mode failed',
 1454                 task.driver.management.set_boot_mode, task, boot_modes.UEFI)
 1455             supp_boot_modes_mock.assert_called_once_with(mock.ANY, task)
 1456 
 1457     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1458                        autospec=True)
 1459     def test_get_boot_mode(self, get_ilo_object_mock):
 1460         expected = 'bios'
 1461         ilo_mock_obj = get_ilo_object_mock.return_value
 1462         ilo_mock_obj.get_current_boot_mode.return_value = 'LEGACY'
 1463         with task_manager.acquire(self.context, self.node.uuid,
 1464                                   shared=True) as task:
 1465             response = task.driver.management.get_boot_mode(task)
 1466             self.assertEqual(expected, response)
 1467 
 1468     @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
 1469                        autospec=True)
 1470     def test_get_boot_mode_fail(self, get_ilo_object_mock):
 1471         ilo_mock_obj = get_ilo_object_mock.return_value
 1472         exc = ilo_error.IloError('error')
 1473         ilo_mock_obj.get_current_boot_mode.side_effect = exc
 1474         with task_manager.acquire(self.context, self.node.uuid,
 1475                                   shared=True) as task:
 1476             self.assertRaisesRegex(
 1477                 exception.IloOperationError, 'Get current boot mode',
 1478                 task.driver.management.get_boot_mode, task)
 1479 
 1480 
 1481 class Ilo5ManagementTestCase(db_base.DbTestCase):
 1482 
 1483     def setUp(self):
 1484         super(Ilo5ManagementTestCase, self).setUp()
 1485         self.driver = mock.Mock(management=ilo_management.Ilo5Management())
 1486         self.clean_step = {'step': 'erase_devices',
 1487                            'interface': 'management'}
 1488         n = {
 1489             'driver': 'ilo5',
 1490             'driver_info': INFO_DICT,
 1491             'clean_step': self.clean_step,
 1492         }
 1493         self.config(enabled_hardware_types=['ilo5'],
 1494                     enabled_boot_interfaces=['ilo-virtual-media'],
 1495                     enabled_console_interfaces=['ilo'],
 1496                     enabled_deploy_interfaces=['iscsi'],
 1497                     enabled_inspect_interfaces=['ilo'],
 1498                     enabled_management_interfaces=['ilo5'],
 1499                     enabled_power_interfaces=['ilo'],
 1500                     enabled_raid_interfaces=['ilo5'])
 1501         self.node = obj_utils.create_test_node(self.context, **n)
 1502 
 1503     @mock.patch.object(deploy_utils, 'build_agent_options',
 1504                        autospec=True)
 1505     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1506     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
 1507     def test_erase_devices_hdd(self, mock_power, ilo_mock, build_agent_mock):
 1508         ilo_mock_object = ilo_mock.return_value
 1509         ilo_mock_object.get_available_disk_types.return_value = ['HDD']
 1510         build_agent_mock.return_value = []
 1511         with task_manager.acquire(self.context, self.node.uuid,
 1512                                   shared=False) as task:
 1513             result = task.driver.management.erase_devices(task)
 1514             self.assertTrue(
 1515                 task.node.driver_internal_info.get(
 1516                     'ilo_disk_erase_hdd_check'))
 1517             self.assertTrue(
 1518                 task.node.driver_internal_info.get(
 1519                     'cleaning_reboot'))
 1520             self.assertFalse(
 1521                 task.node.driver_internal_info.get(
 1522                     'skip_current_clean_step'))
 1523             ilo_mock_object.do_disk_erase.assert_called_once_with(
 1524                 'HDD', 'overwrite')
 1525             self.assertEqual(states.CLEANWAIT, result)
 1526             mock_power.assert_called_once_with(task, states.REBOOT)
 1527 
 1528     @mock.patch.object(deploy_utils, 'build_agent_options',
 1529                        autospec=True)
 1530     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1531     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
 1532     def test_erase_devices_ssd(self, mock_power, ilo_mock, build_agent_mock):
 1533         ilo_mock_object = ilo_mock.return_value
 1534         ilo_mock_object.get_available_disk_types.return_value = ['SSD']
 1535         build_agent_mock.return_value = []
 1536         with task_manager.acquire(self.context, self.node.uuid,
 1537                                   shared=False) as task:
 1538             result = task.driver.management.erase_devices(task)
 1539             self.assertTrue(
 1540                 task.node.driver_internal_info.get(
 1541                     'ilo_disk_erase_ssd_check'))
 1542             self.assertTrue(
 1543                 task.node.driver_internal_info.get(
 1544                     'ilo_disk_erase_hdd_check'))
 1545             self.assertTrue(
 1546                 task.node.driver_internal_info.get(
 1547                     'cleaning_reboot'))
 1548             self.assertFalse(
 1549                 task.node.driver_internal_info.get(
 1550                     'skip_current_clean_step'))
 1551             ilo_mock_object.do_disk_erase.assert_called_once_with(
 1552                 'SSD', 'block')
 1553             self.assertEqual(states.CLEANWAIT, result)
 1554             mock_power.assert_called_once_with(task, states.REBOOT)
 1555 
 1556     @mock.patch.object(deploy_utils, 'build_agent_options',
 1557                        autospec=True)
 1558     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1559     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
 1560     def test_erase_devices_ssd_when_hdd_done(self, mock_power, ilo_mock,
 1561                                              build_agent_mock):
 1562         build_agent_mock.return_value = []
 1563         ilo_mock_object = ilo_mock.return_value
 1564         ilo_mock_object.get_available_disk_types.return_value = ['HDD', 'SSD']
 1565         self.node.driver_internal_info = {'ilo_disk_erase_hdd_check': True}
 1566         self.node.save()
 1567         with task_manager.acquire(self.context, self.node.uuid,
 1568                                   shared=False) as task:
 1569             result = task.driver.management.erase_devices(task)
 1570             self.assertTrue(
 1571                 task.node.driver_internal_info.get(
 1572                     'ilo_disk_erase_hdd_check'))
 1573             self.assertTrue(
 1574                 task.node.driver_internal_info.get(
 1575                     'ilo_disk_erase_ssd_check'))
 1576             self.assertTrue(
 1577                 task.node.driver_internal_info.get(
 1578                     'cleaning_reboot'))
 1579             self.assertFalse(
 1580                 task.node.driver_internal_info.get(
 1581                     'skip_current_clean_step'))
 1582             ilo_mock_object.do_disk_erase.assert_called_once_with(
 1583                 'SSD', 'block')
 1584             self.assertEqual(states.CLEANWAIT, result)
 1585             mock_power.assert_called_once_with(task, states.REBOOT)
 1586 
 1587     @mock.patch.object(ilo_management.LOG, 'info', autospec=True)
 1588     @mock.patch.object(ilo_management.Ilo5Management,
 1589                        '_wait_for_disk_erase_status', autospec=True)
 1590     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1591     def test_erase_devices_completed(self, ilo_mock, disk_status_mock,
 1592                                      log_mock):
 1593         ilo_mock_object = ilo_mock.return_value
 1594         ilo_mock_object.get_available_disk_types.return_value = ['HDD', 'SSD']
 1595         disk_status_mock.return_value = True
 1596         self.node.driver_internal_info = {'ilo_disk_erase_hdd_check': True,
 1597                                           'ilo_disk_erase_ssd_check': True}
 1598         self.node.save()
 1599         with task_manager.acquire(self.context, self.node.uuid,
 1600                                   shared=False) as task:
 1601             task.driver.management.erase_devices(task)
 1602             self.assertFalse(
 1603                 task.node.driver_internal_info.get(
 1604                     'ilo_disk_erase_hdd_check'))
 1605             self.assertFalse(
 1606                 task.node.driver_internal_info.get(
 1607                     'ilo_disk_erase_hdd_check'))
 1608             self.assertTrue(log_mock.called)
 1609 
 1610     @mock.patch.object(deploy_utils, 'build_agent_options',
 1611                        autospec=True)
 1612     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1613     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
 1614     def test_erase_devices_hdd_with_erase_pattern_zero(
 1615             self, mock_power, ilo_mock, build_agent_mock):
 1616         ilo_mock_object = ilo_mock.return_value
 1617         ilo_mock_object.get_available_disk_types.return_value = ['HDD']
 1618         build_agent_mock.return_value = []
 1619         with task_manager.acquire(self.context, self.node.uuid,
 1620                                   shared=False) as task:
 1621             result = task.driver.management.erase_devices(
 1622                 task, erase_pattern={'hdd': 'zero', 'ssd': 'zero'})
 1623             self.assertTrue(
 1624                 task.node.driver_internal_info.get(
 1625                     'ilo_disk_erase_hdd_check'))
 1626             self.assertTrue(
 1627                 task.node.driver_internal_info.get(
 1628                     'cleaning_reboot'))
 1629             self.assertFalse(
 1630                 task.node.driver_internal_info.get(
 1631                     'skip_current_clean_step'))
 1632             ilo_mock_object.do_disk_erase.assert_called_once_with(
 1633                 'HDD', 'zero')
 1634             self.assertEqual(states.CLEANWAIT, result)
 1635             mock_power.assert_called_once_with(task, states.REBOOT)
 1636 
 1637     @mock.patch.object(ilo_management.LOG, 'info', autospec=True)
 1638     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1639     def test_erase_devices_when_no_drive_available(
 1640             self, ilo_mock, log_mock):
 1641         ilo_mock_object = ilo_mock.return_value
 1642         ilo_mock_object.get_available_disk_types.return_value = []
 1643         with task_manager.acquire(self.context, self.node.uuid,
 1644                                   shared=False) as task:
 1645             task.driver.management.erase_devices(task)
 1646             self.assertTrue(log_mock.called)
 1647 
 1648     def test_erase_devices_hdd_with_invalid_format_erase_pattern(
 1649             self):
 1650         with task_manager.acquire(self.context, self.node.uuid,
 1651                                   shared=False) as task:
 1652             self.assertRaises(exception.InvalidParameterValue,
 1653                               task.driver.management.erase_devices,
 1654                               task, erase_pattern=123)
 1655 
 1656     def test_erase_devices_hdd_with_invalid_device_type_erase_pattern(
 1657             self):
 1658         with task_manager.acquire(self.context, self.node.uuid,
 1659                                   shared=False) as task:
 1660             self.assertRaises(exception.InvalidParameterValue,
 1661                               task.driver.management.erase_devices,
 1662                               task, erase_pattern={'xyz': 'block'})
 1663 
 1664     def test_erase_devices_hdd_with_invalid_erase_pattern(
 1665             self):
 1666         with task_manager.acquire(self.context, self.node.uuid,
 1667                                   shared=False) as task:
 1668             self.assertRaises(exception.InvalidParameterValue,
 1669                               task.driver.management.erase_devices,
 1670                               task, erase_pattern={'ssd': 'xyz'})
 1671 
 1672     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1673     @mock.patch.object(manager_utils, 'cleaning_error_handler',
 1674                        autospec=True)
 1675     def test_erase_devices_hdd_ilo_error(self, clean_err_handler_mock,
 1676                                          ilo_mock):
 1677         ilo_mock_object = ilo_mock.return_value
 1678         ilo_mock_object.get_available_disk_types.return_value = ['HDD']
 1679         exc = ilo_error.IloError('error')
 1680         ilo_mock_object.do_disk_erase.side_effect = exc
 1681         with task_manager.acquire(self.context, self.node.uuid,
 1682                                   shared=False) as task:
 1683             task.driver.management.erase_devices(task)
 1684             ilo_mock_object.do_disk_erase.assert_called_once_with(
 1685                 'HDD', 'overwrite')
 1686             self.assertNotIn('ilo_disk_erase_hdd_check',
 1687                              task.node.driver_internal_info)
 1688             self.assertNotIn('ilo_disk_erase_ssd_check',
 1689                              task.node.driver_internal_info)
 1690             self.assertNotIn('cleaning_reboot',
 1691                              task.node.driver_internal_info)
 1692             self.assertNotIn('skip_current_clean_step',
 1693                              task.node.driver_internal_info)
 1694             clean_err_handler_mock.assert_called_once_with(
 1695                 task,
 1696                 ("Out-of-band sanitize disk erase job failed for node %s. "
 1697                  "Message: 'error'." % task.node.uuid),
 1698                 errmsg=exc)
 1699 
 1700     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
 1701     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1702     def test_one_button_secure_erase(self, ilo_mock, mock_power):
 1703         ilo_mock_object = ilo_mock.return_value
 1704         self.node.clean_step = {'step': 'one_button_secure_erase',
 1705                                 'interface': 'management'}
 1706         self.node.save()
 1707         with task_manager.acquire(self.context, self.node.uuid,
 1708                                   shared=False) as task:
 1709             result = task.driver.management.one_button_secure_erase(task)
 1710             self.assertTrue(
 1711                 ilo_mock_object.do_one_button_secure_erase.called)
 1712             self.assertEqual(states.CLEANWAIT, result)
 1713             mock_power.assert_called_once_with(task, states.REBOOT)
 1714             self.assertEqual(task.node.maintenance, True)
 1715 
 1716     @mock.patch.object(ilo_common, 'get_ilo_object', autospec=True)
 1717     @mock.patch.object(manager_utils, 'cleaning_error_handler',
 1718                        autospec=True)
 1719     def test_one_button_secure_erase_ilo_error(
 1720             self, clean_err_handler_mock, ilo_mock):
 1721         ilo_mock_object = ilo_mock.return_value
 1722         self.node.clean_step = {'step': 'one_button_secure_erase',
 1723                                 'interface': 'management'}
 1724         self.node.save()
 1725         exc = ilo_error.IloError('error')
 1726         ilo_mock_object.do_one_button_secure_erase.side_effect = exc
 1727         with task_manager.acquire(self.context, self.node.uuid,
 1728                                   shared=False) as task:
 1729             task.driver.management.one_button_secure_erase(task)
 1730             clean_err_handler_mock.assert_called_once_with(
 1731                 task,
 1732                 ("One button secure erase job failed for node %s. "
 1733                  "Message: 'error'." % task.node.uuid),
 1734                 errmsg=exc)
 1735             self.assertTrue(
 1736                 ilo_mock_object.do_one_button_secure_erase.called)