path: root/ironic/tests/unit/drivers/modules/ilo/test_common.py

# Copyright 2014 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

"""Test class for common methods used by iLO modules."""

import builtins
import hashlib
import io
import os
import shutil
import tempfile
from unittest import mock

from ironic_lib import utils as ironic_utils
from oslo_config import cfg
from oslo_utils import importutils
from oslo_utils import uuidutils

from ironic.common import boot_devices
from ironic.common import exception
from ironic.common import images
from ironic.common import swift
from ironic.conductor import task_manager
from ironic.conductor import utils as manager_utils
from ironic.drivers.modules.ilo import common as ilo_common
from ironic.tests.unit.db import base as db_base
from ironic.tests.unit.db import utils as db_utils
from ironic.tests.unit.objects import utils as obj_utils

INFO_DICT = db_utils.get_test_ilo_info()

ilo_client = importutils.try_import('proliantutils.ilo.client')
ilo_error = importutils.try_import('proliantutils.exception')

CONF = cfg.CONF


class BaseIloTest(db_base.DbTestCase):

    boot_interface = None

    def setUp(self):
        super(BaseIloTest, self).setUp()
        self.config(enabled_hardware_types=['ilo', 'fake-hardware'],
                    enabled_boot_interfaces=['ilo-pxe', 'ilo-virtual-media',
                                             'fake'],
                    enabled_bios_interfaces=['ilo', 'no-bios'],
                    enabled_power_interfaces=['ilo', 'fake'],
                    enabled_management_interfaces=['ilo', 'fake'],
                    enabled_inspect_interfaces=['ilo', 'fake', 'no-inspect'],
                    enabled_console_interfaces=['ilo', 'fake', 'no-console'],
                    enabled_vendor_interfaces=['ilo', 'fake', 'no-vendor'])
        self.info = INFO_DICT.copy()
        self.node = obj_utils.create_test_node(
            self.context, uuid=uuidutils.generate_uuid(),
            driver='ilo', boot_interface=self.boot_interface,
            bios_interface='ilo',
            driver_info=self.info)


class IloValidateParametersTestCase(BaseIloTest):

    @mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
    def _test_parse_driver_info(self, isFile_mock):

        info = ilo_common.parse_driver_info(self.node)

        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual('/home/user/cafile.pem', info['verify_ca'])
        self.assertEqual('user', info['snmp_auth_user'])
        self.assertEqual('1234', info['snmp_auth_prot_password'])
        self.assertEqual('4321', info['snmp_auth_priv_password'])
        self.assertEqual('SHA', info['snmp_auth_protocol'])
        self.assertEqual('AES', info['snmp_auth_priv_protocol'])

    @mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
    def test_parse_driver_info_snmp_inspection_false(self, isFile_mock):
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])

    @mock.patch.object(os.path, 'isdir', autospec=True)
    def test_parse_driver_info_path_verify_ca_dir(self, mock_isdir):
        mock_isdir.return_value = True
        fake_path = '/path/to/a/valid/CA'
        self.node.driver_info['ilo_verify_ca'] = fake_path
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual(fake_path, info['verify_ca'])
        mock_isdir.assert_called_once_with(fake_path)

    @mock.patch.object(os.path, 'isfile', autospec=True)
    def test_parse_driver_info_path_verify_ca_file(self, mock_isfile):
        mock_isfile.return_value = True
        fake_path = '/path/to/a/valid/CA.pem'
        self.node.driver_info['ilo_verify_ca'] = fake_path
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual(fake_path, info['verify_ca'])
        mock_isfile.assert_called_once_with(fake_path)

    def test_parse_driver_info_verify_ca_default_value(self):
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual(True, info['verify_ca'])

    def test_parse_driver_info_verify_ca_string_false(self):
        self.config(verify_ca="False", group='ilo')
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual(False, info['verify_ca'])

    def test_parse_driver_info_verify_ca_boolean_true(self):
        self.config(verify_ca=True, group='ilo')
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual(True, info['verify_ca'])

    def test_parse_driver_info_verify_ca_boolean_false(self):
        self.config(verify_ca=False, group='ilo')
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual(False, info['verify_ca'])

    def test_parse_driver_info_invalid_value_verify_ca(self):
        # Integers are not supported
        self.node.driver_info['ilo_verify_ca'] = 123456
        self.assertRaisesRegex(exception.InvalidParameterValue,
                               'Invalid value type',
                               ilo_common._parse_optional_driver_info,
                               self.node)

    @mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
    def test_parse_driver_info_snmp_true_no_auth_priv_protocols(self,
                                                                isFile_mock):
        d_info = {'ca_file': '/home/user/cafile.pem',
                  'snmp_auth_prot_password': '1234',
                  'snmp_auth_user': 'user',
                  'snmp_auth_priv_password': '4321'}
        self.node.driver_info.update(d_info)
        info = ilo_common.parse_driver_info(self.node)
        self.assertEqual(INFO_DICT['ilo_address'], info['ilo_address'])
        self.assertEqual(INFO_DICT['ilo_username'], info['ilo_username'])
        self.assertEqual(INFO_DICT['ilo_password'], info['ilo_password'])
        self.assertEqual(60, info['client_timeout'])
        self.assertEqual(443, info['client_port'])
        self.assertEqual('/home/user/cafile.pem', info['verify_ca'])
        self.assertEqual('user', info['snmp_auth_user'])
        self.assertEqual('1234', info['snmp_auth_prot_password'])
        self.assertEqual('4321', info['snmp_auth_priv_password'])

    def test_parse_driver_info_ca_file_and_snmp_inspection_true(self):
        d_info = {'ca_file': '/home/user/cafile.pem',
                  'snmp_auth_prot_password': '1234',
                  'snmp_auth_user': 'user',
                  'snmp_auth_priv_password': '4321',
                  'snmp_auth_protocol': 'SHA',
                  'snmp_auth_priv_protocol': 'AES'}
        self.node.driver_info.update(d_info)
        self._test_parse_driver_info()

    def test_parse_driver_info_snmp_true_invalid_auth_protocol(self):
        d_info = {'ca_file': '/home/user/cafile.pem',
                  'snmp_auth_prot_password': '1234',
                  'snmp_auth_user': 'user',
                  'snmp_auth_priv_password': '4321',
                  'snmp_auth_protocol': 'abc',
                  'snmp_auth_priv_protocol': 'AES'}
        self.node.driver_info.update(d_info)
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_snmp_true_invalid_priv_protocol(self):
        d_info = {'ca_file': '/home/user/cafile.pem',
                  'snmp_auth_prot_password': '1234',
                  'snmp_auth_user': 'user',
                  'snmp_auth_priv_password': '4321',
                  'snmp_auth_protocol': 'SHA',
                  'snmp_auth_priv_protocol': 'xyz'}
        self.node.driver_info.update(d_info)
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_snmp_true_integer_auth_protocol(self):
        d_info = {'ca_file': '/home/user/cafile.pem',
                  'snmp_auth_prot_password': '1234',
                  'snmp_auth_user': 'user',
                  'snmp_auth_priv_password': '4321',
                  'snmp_auth_protocol': 12,
                  'snmp_auth_priv_protocol': 'AES'}
        self.node.driver_info.update(d_info)
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_snmp_inspection_true_raises(self):
        self.node.driver_info['snmp_auth_user'] = 'abc'
        self.assertRaises(exception.MissingParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_missing_address(self):
        del self.node.driver_info['ilo_address']
        self.assertRaises(exception.MissingParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_missing_username(self):
        del self.node.driver_info['ilo_username']
        self.assertRaises(exception.MissingParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_missing_password(self):
        del self.node.driver_info['ilo_password']
        self.assertRaises(exception.MissingParameterValue,
                          ilo_common.parse_driver_info, self.node)

    @mock.patch.object(os.path, 'isfile', return_value=False, autospec=True)
    def test_parse_driver_info_invalid_cafile(self, isFile_mock):
        self.node.driver_info['ca_file'] = '/home/missing.pem'
        self.assertRaisesRegex(exception.InvalidParameterValue,
                               'ca_file "/home/missing.pem" is not found.',
                               ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_invalid_timeout(self):
        self.node.driver_info['client_timeout'] = 'qwe'
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_invalid_port(self):
        self.node.driver_info['client_port'] = 'qwe'
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)
        self.node.driver_info['client_port'] = '65536'
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)
        self.node.driver_info['console_port'] = 'invalid'
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)
        self.node.driver_info['console_port'] = '-1'
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.parse_driver_info, self.node)

    def test_parse_driver_info_missing_multiple_params(self):
        del self.node.driver_info['ilo_password']
        del self.node.driver_info['ilo_address']
        e = self.assertRaises(exception.MissingParameterValue,
                              ilo_common.parse_driver_info, self.node)
        self.assertIn('ilo_password', str(e))
        self.assertIn('ilo_address', str(e))

    def test_parse_driver_info_invalid_multiple_params(self):
        self.node.driver_info['client_timeout'] = 'qwe'
        e = self.assertRaises(exception.InvalidParameterValue,
                              ilo_common.parse_driver_info, self.node)
        self.assertIn('client_timeout', str(e))


class IloCommonMethodsTestCase(BaseIloTest):

    @mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
    def _test_get_ilo_object(self, isFile_mock, ca_file=None):
        if not mock._is_instance_mock(ilo_client):
            mock.patch.object(ilo_client, 'IloClient', autospec=True).start()
        ilo_client_mock = ilo_client.IloClient
        self.info['client_timeout'] = 600
        self.info['client_port'] = 4433
        self.info['ilo_verify_ca'] = ca_file
        self.node.driver_info = self.info
        ilo_client_mock.return_value = 'ilo_object'
        returned_ilo_object = ilo_common.get_ilo_object(self.node)
        ilo_client_mock.assert_called_with(
            self.info['ilo_address'],
            self.info['ilo_username'],
            self.info['ilo_password'],
            self.info['client_timeout'],
            self.info['client_port'],
            cacert=self.info['ilo_verify_ca'],
            snmp_credentials=None)
        self.assertEqual('ilo_object', returned_ilo_object)

    @mock.patch.object(os.path, 'isfile', return_value=True, autospec=True)
    def test_get_ilo_object_snmp(self, isFile_mock):
        if not mock._is_instance_mock(ilo_client):
            mock.patch.object(ilo_client, 'IloClient', autospec=True).start()
        ilo_client_mock = ilo_client.IloClient
        info = {'auth_user': 'user',
                'auth_prot_pp': '1234',
                'auth_priv_pp': '4321',
                'auth_protocol': 'SHA',
                'priv_protocol': 'AES',
                'snmp_inspection': True}
        d_info = {'client_timeout': 600,
                  'client_port': 4433,
                  'verify_ca': 'True',
                  'snmp_auth_user': 'user',
                  'snmp_auth_prot_password': '1234',
                  'snmp_auth_priv_password': '4321',
                  'snmp_auth_protocol': 'SHA',
                  'snmp_auth_priv_protocol': 'AES'}
        self.info.update(d_info)
        self.node.driver_info = self.info
        ilo_client_mock.return_value = 'ilo_object'
        returned_ilo_object = ilo_common.get_ilo_object(self.node)
        ilo_client_mock.assert_called_with(
            self.info['ilo_address'],
            self.info['ilo_username'],
            self.info['ilo_password'],
            self.info['client_timeout'],
            self.info['client_port'],
            cacert=self.info['verify_ca'],
            snmp_credentials=info)
        self.assertEqual('ilo_object', returned_ilo_object)

    def test_get_ilo_object_cafile(self):
        self._test_get_ilo_object(ca_file='/home/user/ilo.pem')

    def test_get_ilo_object_cafile_boolean(self):
        self._test_get_ilo_object(ca_file=True)

    def test_update_ipmi_properties(self):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ipmi_info = {
                "ipmi_address": "1.2.3.4",
                "ipmi_username": "admin",
                "ipmi_password": "fake",
                "ipmi_terminal_port": 60
            }
            self.info['console_port'] = 60
            task.node.driver_info = self.info
            ilo_common.update_ipmi_properties(task)
            actual_info = task.node.driver_info
            expected_info = dict(self.info, **ipmi_info)
            self.assertEqual(expected_info, actual_info)

    def test__get_floppy_image_name(self):
        image_name_expected = 'image-' + self.node.uuid
        image_name_actual = ilo_common._get_floppy_image_name(self.node)
        self.assertEqual(image_name_expected, image_name_actual)

    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    @mock.patch.object(images, 'create_vfat_image', spec_set=True,
                       autospec=True)
    @mock.patch.object(tempfile, 'NamedTemporaryFile', spec_set=True,
                       autospec=True)
    def test__prepare_floppy_image(self, tempfile_mock, fatimage_mock,
                                   swift_api_mock):
        mock_image_file_handle = mock.MagicMock(spec=io.BytesIO)
        mock_image_file_obj = mock.MagicMock(spec=io.BytesIO)
        mock_image_file_obj.name = 'image-tmp-file'
        mock_image_file_handle.__enter__.return_value = mock_image_file_obj

        tempfile_mock.return_value = mock_image_file_handle

        swift_obj_mock = swift_api_mock.return_value
        self.config(swift_ilo_container='ilo_cont', group='ilo')
        self.config(swift_object_expiry_timeout=1, group='ilo')
        deploy_args = {'arg1': 'val1', 'arg2': 'val2'}
        swift_obj_mock.get_temp_url.return_value = 'temp-url'
        timeout = CONF.ilo.swift_object_expiry_timeout
        object_headers = {'X-Delete-After': str(timeout)}

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            temp_url = ilo_common._prepare_floppy_image(task, deploy_args)
            node_uuid = task.node.uuid

        object_name = 'image-' + node_uuid
        fatimage_mock.assert_called_once_with('image-tmp-file',
                                              parameters=deploy_args)

        swift_obj_mock.create_object.assert_called_once_with(
            'ilo_cont', object_name, 'image-tmp-file',
            object_headers=object_headers)
        swift_obj_mock.get_temp_url.assert_called_once_with(
            'ilo_cont', object_name, timeout)
        self.assertEqual('temp-url', temp_url)

    @mock.patch.object(ilo_common, 'copy_image_to_web_server',
                       spec_set=True, autospec=True)
    @mock.patch.object(images, 'create_vfat_image', spec_set=True,
                       autospec=True)
    @mock.patch.object(tempfile, 'NamedTemporaryFile', spec_set=True,
                       autospec=True)
    def test__prepare_floppy_image_use_webserver(self, tempfile_mock,
                                                 fatimage_mock,
                                                 copy_mock):
        mock_image_file_handle = mock.MagicMock(spec=io.BytesIO)
        mock_image_file_obj = mock.MagicMock(spec=io.BytesIO)
        mock_image_file_obj.name = 'image-tmp-file'
        mock_image_file_handle.__enter__.return_value = mock_image_file_obj

        tempfile_mock.return_value = mock_image_file_handle
        self.config(use_web_server_for_images=True, group='ilo')
        deploy_args = {'arg1': 'val1', 'arg2': 'val2'}
        CONF.set_override('http_url', 'http://abc.com/httpboot', 'deploy')
        CONF.set_override('http_root', '/httpboot', 'deploy')

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            node_uuid = task.node.uuid
            object_name = 'image-' + node_uuid
            http_url = CONF.deploy.http_url + '/' + object_name
            copy_mock.return_value = "http://abc.com/httpboot/" + object_name
            temp_url = ilo_common._prepare_floppy_image(task, deploy_args)

        fatimage_mock.assert_called_once_with('image-tmp-file',
                                              parameters=deploy_args)
        copy_mock.assert_called_once_with('image-tmp-file', object_name)
        self.assertEqual(http_url, temp_url)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_attach_vmedia(self, get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        insert_media_mock = ilo_mock_object.insert_virtual_media
        set_status_mock = ilo_mock_object.set_vm_status

        ilo_common.attach_vmedia(self.node, 'FLOPPY', 'url')
        insert_media_mock.assert_called_once_with('url', device='FLOPPY')
        set_status_mock.assert_called_once_with(
            device='FLOPPY', boot_option='CONNECT', write_protect='YES')

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_attach_vmedia_fails(self, get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        set_status_mock = ilo_mock_object.set_vm_status
        exc = ilo_error.IloError('error')
        set_status_mock.side_effect = exc
        self.assertRaises(exception.IloOperationError,
                          ilo_common.attach_vmedia, self.node,
                          'FLOPPY', 'url')

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_set_boot_mode(self, get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        get_pending_boot_mode_mock = ilo_object_mock.get_pending_boot_mode
        set_pending_boot_mode_mock = ilo_object_mock.set_pending_boot_mode
        get_pending_boot_mode_mock.return_value = 'LEGACY'
        ilo_common.set_boot_mode(self.node, 'uefi')
        get_ilo_object_mock.assert_called_once_with(self.node)
        get_pending_boot_mode_mock.assert_called_once_with()
        set_pending_boot_mode_mock.assert_called_once_with('UEFI')

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_set_boot_mode_without_set_pending_boot_mode(self,
                                                         get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        get_pending_boot_mode_mock = ilo_object_mock.get_pending_boot_mode
        get_pending_boot_mode_mock.return_value = 'LEGACY'
        ilo_common.set_boot_mode(self.node, 'bios')
        get_ilo_object_mock.assert_called_once_with(self.node)
        get_pending_boot_mode_mock.assert_called_once_with()
        self.assertFalse(ilo_object_mock.set_pending_boot_mode.called)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_set_boot_mode_with_IloOperationError(self,
                                                  get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        get_pending_boot_mode_mock = ilo_object_mock.get_pending_boot_mode
        get_pending_boot_mode_mock.return_value = 'UEFI'
        set_pending_boot_mode_mock = ilo_object_mock.set_pending_boot_mode
        exc = ilo_error.IloError('error')
        set_pending_boot_mode_mock.side_effect = exc
        self.assertRaises(exception.IloOperationError,
                          ilo_common.set_boot_mode, self.node, 'bios')
        get_ilo_object_mock.assert_called_once_with(self.node)
        get_pending_boot_mode_mock.assert_called_once_with()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_current_boot_mode(self, get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        get_current_boot_mode_mock = ilo_object_mock.get_current_boot_mode
        get_current_boot_mode_mock.return_value = 'LEGACY'
        ret = ilo_common.get_current_boot_mode(self.node)
        self.assertEqual('bios', ret)
        get_ilo_object_mock.assert_called_once_with(self.node)
        get_current_boot_mode_mock.assert_called_once_with()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_current_boot_mode_fail(self, get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        get_current_boot_mode_mock = ilo_object_mock.get_current_boot_mode
        exc = ilo_error.IloError('error')
        get_current_boot_mode_mock.side_effect = exc
        self.assertRaises(exception.IloOperationError,
                          ilo_common.get_current_boot_mode, self.node)
        get_ilo_object_mock.assert_called_once_with(self.node)
        get_current_boot_mode_mock.assert_called_once_with()

    @mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_instance_info_exists(self,
                                                   set_boot_mode_mock):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_internal_info['deploy_boot_mode'] = 'bios'
            ilo_common.update_boot_mode(task)
            set_boot_mode_mock.assert_called_once_with(task.node, 'bios')

    @mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_capabilities_exist(self,
                                                 set_boot_mode_mock):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.properties['capabilities'] = 'boot_mode:bios'
            ilo_common.update_boot_mode(task)
            set_boot_mode_mock.assert_called_once_with(task.node, 'bios')

    @mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_use_def_boot_mode(self,
                                                set_boot_mode_mock):
        self.config(default_boot_mode='bios', group='ilo')
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.update_boot_mode(task)
            set_boot_mode_mock.assert_called_once_with(task.node, 'bios')
            self.assertEqual(
                'bios', task.node.driver_internal_info['deploy_boot_mode'])

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_update_boot_mode(self, get_ilo_object_mock):
        self.config(default_boot_mode="auto", group='ilo')
        ilo_mock_obj = get_ilo_object_mock.return_value
        ilo_mock_obj.get_pending_boot_mode.return_value = 'LEGACY'

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.update_boot_mode(task)
            get_ilo_object_mock.assert_called_once_with(task.node)
            ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
            self.assertEqual(
                'bios', task.node.driver_internal_info['deploy_boot_mode'])

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_unknown(self,
                                      get_ilo_object_mock):
        ilo_mock_obj = get_ilo_object_mock.return_value
        ilo_mock_obj.get_pending_boot_mode.return_value = 'UNKNOWN'
        set_pending_boot_mode_mock = ilo_mock_obj.set_pending_boot_mode

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.update_boot_mode(task)
            get_ilo_object_mock.assert_called_once_with(task.node)
            ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
            set_pending_boot_mode_mock.assert_called_once_with('UEFI')
            self.assertEqual(
                'uefi', task.node.driver_internal_info['deploy_boot_mode'])

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_unknown_except(self,
                                             get_ilo_object_mock):
        ilo_mock_obj = get_ilo_object_mock.return_value
        ilo_mock_obj.get_pending_boot_mode.return_value = 'UNKNOWN'
        set_pending_boot_mode_mock = ilo_mock_obj.set_pending_boot_mode
        exc = ilo_error.IloError('error')
        set_pending_boot_mode_mock.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.update_boot_mode, task)
            get_ilo_object_mock.assert_called_once_with(task.node)
            ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_legacy(self,
                                     get_ilo_object_mock):
        ilo_mock_obj = get_ilo_object_mock.return_value
        exc = ilo_error.IloCommandNotSupportedError('error')
        ilo_mock_obj.get_pending_boot_mode.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.update_boot_mode(task)
            get_ilo_object_mock.assert_called_once_with(task.node)
            ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
            self.assertEqual(
                'bios', task.node.driver_internal_info['deploy_boot_mode'])

    @mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
                       autospec=True)
    def test_update_boot_mode_prop_boot_mode_exist(self,
                                                   set_boot_mode_mock):

        properties = {'capabilities': 'boot_mode:uefi'}
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.properties = properties
            ilo_common.update_boot_mode(task)
            set_boot_mode_mock.assert_called_once_with(task.node, 'uefi')

    @mock.patch.object(images, 'get_temp_url_for_glance_image', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, '_prepare_floppy_image', spec_set=True,
                       autospec=True)
    def test_setup_vmedia_for_boot_with_parameters(
            self, prepare_image_mock, attach_vmedia_mock, temp_url_mock):
        parameters = {'a': 'b'}
        boot_iso = '733d1c44-a2ea-414b-aca7-69decf20d810'
        prepare_image_mock.return_value = 'floppy_url'
        temp_url_mock.return_value = 'image_url'

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.setup_vmedia_for_boot(task, boot_iso, parameters)
            prepare_image_mock.assert_called_once_with(task, parameters)
            attach_vmedia_mock.assert_any_call(task.node, 'FLOPPY',
                                               'floppy_url')

            temp_url_mock.assert_called_once_with(
                task.context, '733d1c44-a2ea-414b-aca7-69decf20d810')
            attach_vmedia_mock.assert_any_call(task.node, 'CDROM', 'image_url')

    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
                       autospec=True)
    def test_setup_vmedia_for_boot_with_swift(self, attach_vmedia_mock,
                                              swift_api_mock):
        swift_obj_mock = swift_api_mock.return_value
        boot_iso = 'swift:object-name'
        swift_obj_mock.get_temp_url.return_value = 'image_url'
        CONF.set_override('swift_ilo_container', 'ilo_cont', 'ilo')
        CONF.set_override('swift_object_expiry_timeout', 1, 'ilo')
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.setup_vmedia_for_boot(task, boot_iso)
            swift_obj_mock.get_temp_url.assert_called_once_with(
                'ilo_cont', 'object-name', 1)
            attach_vmedia_mock.assert_called_once_with(
                task.node, 'CDROM', 'image_url')

    @mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
                       autospec=True)
    def test_setup_vmedia_for_boot_with_url(self, attach_vmedia_mock):
        boot_iso = 'http://abc.com/img.iso'
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.setup_vmedia_for_boot(task, boot_iso)
            attach_vmedia_mock.assert_called_once_with(task.node, 'CDROM',
                                                       boot_iso)

    @mock.patch.object(ilo_common, 'eject_vmedia_devices',
                       spec_set=True, autospec=True)
    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
                       autospec=True)
    def test_cleanup_vmedia_boot(self, get_name_mock, swift_api_mock,
                                 eject_mock):
        swift_obj_mock = swift_api_mock.return_value
        CONF.set_override('swift_ilo_container', 'ilo_cont', 'ilo')

        get_name_mock.return_value = 'image-node-uuid'

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.cleanup_vmedia_boot(task)
            swift_obj_mock.delete_object.assert_called_once_with(
                'ilo_cont', 'image-node-uuid')
            eject_mock.assert_called_once_with(task)

    @mock.patch.object(ilo_common.LOG, 'exception', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'eject_vmedia_devices',
                       spec_set=True, autospec=True)
    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
                       autospec=True)
    def test_cleanup_vmedia_boot_exc(self, get_name_mock, swift_api_mock,
                                     eject_mock, log_mock):
        exc = exception.SwiftOperationError('error')
        swift_obj_mock = swift_api_mock.return_value
        swift_obj_mock.delete_object.side_effect = exc
        CONF.set_override('swift_ilo_container', 'ilo_cont', 'ilo')

        get_name_mock.return_value = 'image-node-uuid'

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.cleanup_vmedia_boot(task)
            swift_obj_mock.delete_object.assert_called_once_with(
                'ilo_cont', 'image-node-uuid')
            self.assertTrue(log_mock.called)
            eject_mock.assert_called_once_with(task)

    @mock.patch.object(ilo_common.LOG, 'info', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'eject_vmedia_devices',
                       spec_set=True, autospec=True)
    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
                       autospec=True)
    def test_cleanup_vmedia_boot_exc_resource_not_found(self, get_name_mock,
                                                        swift_api_mock,
                                                        eject_mock, log_mock):
        exc = exception.SwiftObjectNotFoundError('error')
        swift_obj_mock = swift_api_mock.return_value
        swift_obj_mock.delete_object.side_effect = exc
        CONF.set_override('swift_ilo_container', 'ilo_cont', 'ilo')

        get_name_mock.return_value = 'image-node-uuid'

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.cleanup_vmedia_boot(task)
            swift_obj_mock.delete_object.assert_called_once_with(
                'ilo_cont', 'image-node-uuid')
            self.assertTrue(log_mock.called)
            eject_mock.assert_called_once_with(task)

    @mock.patch.object(ilo_common, 'eject_vmedia_devices',
                       spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, 'destroy_floppy_image_from_web_server',
                       spec_set=True, autospec=True)
    def test_cleanup_vmedia_boot_for_webserver(self,
                                               destroy_image_mock,
                                               eject_mock):
        CONF.set_override('use_web_server_for_images', True, 'ilo')

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.cleanup_vmedia_boot(task)
            destroy_image_mock.assert_called_once_with(task.node)
            eject_mock.assert_called_once_with(task)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_eject_vmedia_devices(self, get_ilo_object_mock):
        ilo_object_mock = mock.MagicMock(spec=['eject_virtual_media'])
        get_ilo_object_mock.return_value = ilo_object_mock
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.eject_vmedia_devices(task)
            ilo_object_mock.eject_virtual_media.assert_has_calls(
                [mock.call('FLOPPY'), mock.call('CDROM')])

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_eject_vmedia_devices_raises(
            self, get_ilo_object_mock):
        ilo_object_mock = mock.MagicMock(spec=['eject_virtual_media'])
        get_ilo_object_mock.return_value = ilo_object_mock
        exc = ilo_error.IloError('error')
        ilo_object_mock.eject_virtual_media.side_effect = exc
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.eject_vmedia_devices,
                              task)

            ilo_object_mock.eject_virtual_media.assert_called_once_with(
                'FLOPPY')

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_secure_boot_mode(self,
                                  get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        ilo_object_mock.get_current_boot_mode.return_value = 'UEFI'
        ilo_object_mock.get_secure_boot_mode.return_value = True
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ret = ilo_common.get_secure_boot_mode(task)
            ilo_object_mock.get_current_boot_mode.assert_called_once_with()
            ilo_object_mock.get_secure_boot_mode.assert_called_once_with()
            self.assertTrue(ret)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_secure_boot_mode_bios(self,
                                       get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        ilo_object_mock.get_current_boot_mode.return_value = 'BIOS'
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ret = ilo_common.get_secure_boot_mode(task)
            ilo_object_mock.get_current_boot_mode.assert_called_once_with()
            self.assertFalse(ilo_object_mock.get_secure_boot_mode.called)
            self.assertFalse(ret)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_secure_boot_mode_fail(self,
                                       get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        exc = ilo_error.IloError('error')
        ilo_mock_object.get_current_boot_mode.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.get_secure_boot_mode,
                              task)
        ilo_mock_object.get_current_boot_mode.assert_called_once_with()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_secure_boot_mode_not_supported(self,
                                                ilo_object_mock):
        ilo_mock_object = ilo_object_mock.return_value
        exc = ilo_error.IloCommandNotSupportedError('error')
        ilo_mock_object.get_current_boot_mode.return_value = 'UEFI'
        ilo_mock_object.get_secure_boot_mode.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationNotSupported,
                              ilo_common.get_secure_boot_mode,
                              task)
        ilo_mock_object.get_current_boot_mode.assert_called_once_with()
        ilo_mock_object.get_secure_boot_mode.assert_called_once_with()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_set_secure_boot_mode(self,
                                  get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.set_secure_boot_mode(task, True)
            ilo_object_mock.set_secure_boot_mode.assert_called_once_with(True)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_set_secure_boot_mode_fail(self,
                                       get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        exc = ilo_error.IloError('error')
        ilo_mock_object.set_secure_boot_mode.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.set_secure_boot_mode,
                              task, False)
        ilo_mock_object.set_secure_boot_mode.assert_called_once_with(False)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_set_secure_boot_mode_not_supported(self,
                                                ilo_object_mock):
        ilo_mock_object = ilo_object_mock.return_value
        exc = ilo_error.IloCommandNotSupportedError('error')
        ilo_mock_object.set_secure_boot_mode.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationNotSupported,
                              ilo_common.set_secure_boot_mode,
                              task, False)
        ilo_mock_object.set_secure_boot_mode.assert_called_once_with(False)

    def test_validate_security_parameter_values(self):
        param_info = {'param': 'password_complexity'}
        param, enabled, ignored = (
            ilo_common.validate_security_parameter_values(param_info))
        self.assertEqual('password_complexity', param)
        self.assertEqual(False, ignored)
        self.assertEqual(True, enabled)

    def test_validate_security_parameter_values_no_param(self):
        param_info = {'param': None}
        self.assertRaises(exception.MissingParameterValue,
                          ilo_common.validate_security_parameter_values,
                          param_info)

    def test_validate_security_parameter_values_invalid_param(self):
        param_info = {'param': 'minimum_password_length'}
        self.assertRaises(exception.InvalidParameterValue,
                          ilo_common.validate_security_parameter_values,
                          param_info)

    @mock.patch.object(os, 'chmod', spec_set=True,
                       autospec=True)
    @mock.patch.object(shutil, 'copyfile', spec_set=True,
                       autospec=True)
    def test_copy_image_to_web_server(self, copy_mock,
                                      chmod_mock):
        CONF.set_override('http_url', 'http://x.y.z.a/webserver/', 'deploy')
        CONF.set_override('http_root', '/webserver', 'deploy')
        expected_url = "http://x.y.z.a/webserver/image-UUID"
        source = 'tmp_image_file'
        destination = "image-UUID"
        image_path = "/webserver/image-UUID"
        actual_url = ilo_common.copy_image_to_web_server(source, destination)
        self.assertEqual(expected_url, actual_url)
        copy_mock.assert_called_once_with(source, image_path)
        chmod_mock.assert_called_once_with(image_path, 0o644)

    @mock.patch.object(os, 'chmod', spec_set=True,
                       autospec=True)
    @mock.patch.object(shutil, 'copyfile', spec_set=True,
                       autospec=True)
    def test_copy_image_to_web_server_fails(self, copy_mock,
                                            chmod_mock):
        CONF.set_override('http_url', 'http://x.y.z.a/webserver/', 'deploy')
        CONF.set_override('http_root', '/webserver', 'deploy')
        source = 'tmp_image_file'
        destination = "image-UUID"
        image_path = "/webserver/image-UUID"
        exc = exception.ImageUploadFailed('reason')
        copy_mock.side_effect = exc
        self.assertRaises(exception.ImageUploadFailed,
                          ilo_common.copy_image_to_web_server,
                          source, destination)
        copy_mock.assert_called_once_with(source, image_path)
        self.assertFalse(chmod_mock.called)

    @mock.patch.object(ilo_common, 'ironic_utils', autospec=True)
    def test_remove_image_from_web_server(self, utils_mock):
        # | GIVEN |
        CONF.set_override('http_url', 'http://x.y.z.a/webserver/', 'deploy')
        CONF.set_override('http_root', '/webserver', 'deploy')
        object_name = 'tmp_image_file'
        # | WHEN |
        ilo_common.remove_image_from_web_server(object_name)
        # | THEN |
        (utils_mock.unlink_without_raise.
         assert_called_once_with("/webserver/tmp_image_file"))

    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    def test_copy_image_to_swift(self, swift_api_mock):
        # | GIVEN |
        self.config(swift_ilo_container='ilo_container', group='ilo')
        self.config(swift_object_expiry_timeout=1, group='ilo')
        container = CONF.ilo.swift_ilo_container
        timeout = CONF.ilo.swift_object_expiry_timeout

        swift_obj_mock = swift_api_mock.return_value
        destination_object_name = 'destination_object_name'
        source_file_path = 'tmp_image_file'
        object_headers = {'X-Delete-After': str(timeout)}
        # | WHEN |
        ilo_common.copy_image_to_swift(source_file_path,
                                       destination_object_name)
        # | THEN |
        swift_obj_mock.create_object.assert_called_once_with(
            container, destination_object_name, source_file_path,
            object_headers=object_headers)
        swift_obj_mock.get_temp_url.assert_called_once_with(
            container, destination_object_name, timeout)

    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    def test_copy_image_to_swift_throws_error_if_swift_operation_fails(
            self, swift_api_mock):
        # | GIVEN |
        self.config(swift_ilo_container='ilo_container', group='ilo')
        self.config(swift_object_expiry_timeout=1, group='ilo')

        swift_obj_mock = swift_api_mock.return_value
        destination_object_name = 'destination_object_name'
        source_file_path = 'tmp_image_file'
        swift_obj_mock.create_object.side_effect = (
            exception.SwiftOperationError(operation='create_object',
                                          error='failed'))
        # | WHEN | & | THEN |
        self.assertRaises(exception.SwiftOperationError,
                          ilo_common.copy_image_to_swift,
                          source_file_path, destination_object_name)

    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    def test_remove_image_from_swift(self, swift_api_mock):
        # | GIVEN |
        self.config(swift_ilo_container='ilo_container', group='ilo')
        container = CONF.ilo.swift_ilo_container

        swift_obj_mock = swift_api_mock.return_value
        object_name = 'object_name'
        # | WHEN |
        ilo_common.remove_image_from_swift(object_name)
        # | THEN |
        swift_obj_mock.delete_object.assert_called_once_with(
            container, object_name)

    @mock.patch.object(ilo_common, 'LOG', spec_set=True, autospec=True)
    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    def test_remove_image_from_swift_suppresses_notfound_exc(
            self, swift_api_mock, LOG_mock):
        # | GIVEN |
        self.config(swift_ilo_container='ilo_container', group='ilo')
        container = CONF.ilo.swift_ilo_container

        swift_obj_mock = swift_api_mock.return_value
        object_name = 'object_name'
        raised_exc = exception.SwiftObjectNotFoundError(
            operation='delete_object', obj=object_name, container=container)
        swift_obj_mock.delete_object.side_effect = raised_exc
        # | WHEN |
        ilo_common.remove_image_from_swift(object_name)
        # | THEN |
        LOG_mock.info.assert_called_once_with(
            mock.ANY, {'associated_with_msg': "", 'err': raised_exc})

    @mock.patch.object(ilo_common, 'LOG', spec_set=True, autospec=True)
    @mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
    def test_remove_image_from_swift_suppresses_operror_exc(
            self, swift_api_mock, LOG_mock):
        # | GIVEN |
        self.config(swift_ilo_container='ilo_container', group='ilo')
        container = CONF.ilo.swift_ilo_container

        swift_obj_mock = swift_api_mock.return_value
        object_name = 'object_name'
        raised_exc = exception.SwiftOperationError(operation='delete_object',
                                                   error='failed')
        swift_obj_mock.delete_object.side_effect = raised_exc
        # | WHEN |
        ilo_common.remove_image_from_swift(object_name, 'alice_in_wonderland')
        # | THEN |
        LOG_mock.exception.assert_called_once_with(
            mock.ANY, {'object_name': object_name, 'container': container,
                       'associated_with_msg': ("associated with "
                                               "alice_in_wonderland"),
                       'err': raised_exc})

    @mock.patch.object(ironic_utils, 'unlink_without_raise', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
                       autospec=True)
    def test_destroy_floppy_image_from_web_server(self, get_floppy_name_mock,
                                                  utils_mock):
        get_floppy_name_mock.return_value = 'image-uuid'
        CONF.set_override('http_root', '/webserver/', 'deploy')
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.destroy_floppy_image_from_web_server(task.node)
            get_floppy_name_mock.assert_called_once_with(task.node)
            utils_mock.assert_called_once_with('/webserver/image-uuid')

    @mock.patch.object(manager_utils, 'node_set_boot_device', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'setup_vmedia_for_boot', spec_set=True,
                       autospec=True)
    def test_setup_vmedia(self,
                          func_setup_vmedia_for_boot,
                          func_set_boot_device):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            parameters = {'a': 'b'}
            iso = '733d1c44-a2ea-414b-aca7-69decf20d810'
            ilo_common.setup_vmedia(task, iso, parameters)
            func_setup_vmedia_for_boot.assert_called_once_with(task, iso,
                                                               parameters)
            func_set_boot_device.assert_called_once_with(task,
                                                         boot_devices.CDROM)

    @mock.patch.object(manager_utils, 'node_set_boot_device', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'setup_vmedia_for_boot', spec_set=True,
                       autospec=True)
    def test_setup_vmedia_without_params(self,
                                         func_setup_vmedia_for_boot,
                                         func_set_boot_device):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            iso = '733d1c44-a2ea-414b-aca7-69decf20d810'
            ilo_common.setup_vmedia(task, iso)
            func_setup_vmedia_for_boot.assert_called_once_with(task, iso, None)
            func_set_boot_device.assert_called_once_with(task,
                                                         boot_devices.CDROM)

    @mock.patch.object(ironic_utils, 'unlink_without_raise', spec_set=True,
                       autospec=True)
    def test_remove_single_or_list_of_files_with_file_list(self, unlink_mock):
        # | GIVEN |
        file_list = ['/any_path1/any_file1',
                     '/any_path2/any_file2',
                     '/any_path3/any_file3']
        # | WHEN |
        ilo_common.remove_single_or_list_of_files(file_list)
        # | THEN |
        calls = [mock.call('/any_path1/any_file1'),
                 mock.call('/any_path2/any_file2'),
                 mock.call('/any_path3/any_file3')]
        unlink_mock.assert_has_calls(calls)

    @mock.patch.object(ironic_utils, 'unlink_without_raise', spec_set=True,
                       autospec=True)
    def test_remove_single_or_list_of_files_with_file_str(self, unlink_mock):
        # | GIVEN |
        file_path = '/any_path1/any_file'
        # | WHEN |
        ilo_common.remove_single_or_list_of_files(file_path)
        # | THEN |
        unlink_mock.assert_called_once_with('/any_path1/any_file')

    @mock.patch.object(builtins, 'open', autospec=True)
    def test_verify_image_checksum(self, open_mock):
        # | GIVEN |
        data = b'Yankee Doodle went to town riding on a pony;'
        file_like_object = io.BytesIO(data)
        open_mock().__enter__.return_value = file_like_object
        actual_hash = hashlib.md5(data).hexdigest()
        # | WHEN |
        ilo_common.verify_image_checksum(file_like_object, actual_hash)
        # | THEN |
        # no any exception thrown

    def test_verify_image_checksum_throws_for_nonexistent_file(self):
        # | GIVEN |
        invalid_file_path = '/some/invalid/file/path'
        # | WHEN | & | THEN |
        self.assertRaises(exception.ImageRefValidationFailed,
                          ilo_common.verify_image_checksum,
                          invalid_file_path, 'hash_xxx')

    @mock.patch.object(builtins, 'open', autospec=True)
    def test_verify_image_checksum_throws_for_failed_validation(self,
                                                                open_mock):
        # | GIVEN |
        data = b'Yankee Doodle went to town riding on a pony;'
        file_like_object = io.BytesIO(data)
        open_mock().__enter__.return_value = file_like_object
        invalid_hash = 'invalid_hash_value'
        # | WHEN | & | THEN |
        self.assertRaises(exception.ImageRefValidationFailed,
                          ilo_common.verify_image_checksum,
                          file_like_object,
                          invalid_hash)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_server_post_state(self,
                                   get_ilo_object_mock):
        ilo_object_mock = get_ilo_object_mock.return_value
        post_state = 'FinishedPost'
        ilo_object_mock.get_host_post_state.return_value = post_state
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ret = ilo_common.get_server_post_state(task.node)
            ilo_object_mock.get_host_post_state.assert_called_once_with()
            self.assertEqual(post_state, ret)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_server_post_state_fail(self,
                                        get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        exc = ilo_error.IloError('error')
        ilo_mock_object.get_host_post_state.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.get_server_post_state, task.node)
        ilo_mock_object.get_host_post_state.assert_called_once_with()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_get_server_post_state_not_supported(self,
                                                 ilo_object_mock):
        ilo_mock_object = ilo_object_mock.return_value
        exc = ilo_error.IloCommandNotSupportedError('error')
        ilo_mock_object.get_host_post_state.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationNotSupported,
                              ilo_common.get_server_post_state,
                              task.node)
        ilo_mock_object.get_host_post_state.assert_called_once_with()

    @mock.patch.object(os.path, 'exists', spec_set=True,
                       autospec=True)
    def test__get_certificate_file_list_none(self, path_exists_mock):
        cl = None
        CONF.set_override('webserver_verify_ca', '/file/path')
        path_exists_mock.return_value = True
        expected = ['/file/path']
        actual = ilo_common._get_certificate_file_list(cl)
        self.assertEqual(expected, actual)

    @mock.patch.object(os.path, 'exists', spec_set=True,
                       autospec=True)
    def test__get_certificate_file_list_empty(self, path_exists_mock):
        cl = []
        CONF.set_override('webserver_verify_ca', '/file/path')
        path_exists_mock.return_value = True
        expected = ['/file/path']
        actual = ilo_common._get_certificate_file_list(cl)
        self.assertEqual(expected, actual)

    @mock.patch.object(os.path, 'exists', spec_set=True,
                       autospec=True)
    def test__get_certificate_file_list_empty_no_path(self, path_exists_mock):
        cl = []
        CONF.set_override('webserver_verify_ca', '/file/path')
        path_exists_mock.return_value = False
        expected = []
        actual = ilo_common._get_certificate_file_list(cl)
        self.assertEqual(expected, actual)

    def test__get_certificate_file_list(self):
        cl = ['file/path/a', 'file/path/b']
        CONF.set_override('webserver_verify_ca', '/file/path/c')
        expected = cl
        actual = ilo_common._get_certificate_file_list(cl)
        self.assertEqual(expected, actual)

    def test__get_certificate_file_list_string_type(self):
        cl = 'file/path/a'
        CONF.set_override('webserver_verify_ca', '/file/path/c')
        self.assertRaisesRegex(exception.InvalidParameterValue,
                               "List of files is .* \"<class 'str'>\" .*",
                               ilo_common._get_certificate_file_list, cl)

    @mock.patch.object(ilo_common, '_get_certificate_file_list', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_add_certificates_false(self, get_ilo_object_mock, get_cl_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_info['ilo_add_certificates'] = "false"
            ilo_common.add_certificates(task, c_l)

        get_cl_mock.assert_not_called()
        ilo_mock_object.add_tls_certificate.assert_not_called()

    @mock.patch.object(ilo_common, '_get_certificate_file_list', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_add_certificates_true(self, get_ilo_object_mock, get_cl_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']
        get_cl_mock.return_value = c_l

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_info['ilo_add_certificates'] = "true"
            ilo_common.add_certificates(task, c_l)

        get_cl_mock.assert_called_once_with(c_l)
        ilo_mock_object.add_tls_certificate.assert_called_once_with(c_l)

    @mock.patch.object(ilo_common, '_get_certificate_file_list', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_add_certificates_None(self, get_ilo_object_mock, get_cl_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']
        get_cl_mock.return_value = c_l

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_info['ilo_add_certificates'] = None
            ilo_common.add_certificates(task, c_l)

        get_cl_mock.assert_called_once_with(c_l)
        ilo_mock_object.add_tls_certificate.assert_called_once_with(c_l)

    @mock.patch.object(ilo_common, '_get_certificate_file_list', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_add_certificates_invalid(self, get_ilo_object_mock, get_cl_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']
        get_cl_mock.return_value = c_l

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_info['ilo_add_certificates'] = "xyz"
            self.assertRaisesRegex(exception.InvalidParameterValue,
                                   "Invalid value type set in driver_info.*",
                                   ilo_common.add_certificates,
                                   task, c_l)

        get_cl_mock.assert_not_called()
        ilo_mock_object.add_tls_certificate.assert_not_called()

    @mock.patch.object(ilo_common, '_get_certificate_file_list', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_add_certificates_true_default(self, get_ilo_object_mock,
                                           get_cl_mock):

        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/A']
        get_cl_mock.return_value = c_l

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_info['ilo_add_certificates'] = "true"
            ilo_common.add_certificates(task)

        get_cl_mock.assert_called_once_with(None)
        ilo_mock_object.add_tls_certificate.assert_called_once_with(c_l)

    @mock.patch.object(ilo_common, '_get_certificate_file_list', spec_set=True,
                       autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_add_certificates_raises_ilo_error(self, get_ilo_object_mock,
                                               get_cl_mock):
        CONF.set_override('webserver_verify_ca', False)
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']
        get_cl_mock.return_value = c_l
        exc = ilo_error.IloError('error')
        ilo_mock_object.add_tls_certificate.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            task.node.driver_info['ilo_add_certificates'] = "true"
            self.assertRaises(exception.IloOperationError,
                              ilo_common.add_certificates,
                              task, c_l)

        get_cl_mock.assert_called_once_with(c_l)
        ilo_mock_object.add_tls_certificate.assert_called_once_with(c_l)

    @mock.patch.object(ilo_common, '_get_certificate_file_list',
                       spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_clear_certificates(self, get_ilo_object_mock,
                                get_certs_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']
        get_certs_mock.return_value = ['/file/path/x']

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.clear_certificates(task, c_l)

        ilo_mock_object.remove_tls_certificate.assert_called_once_with(
            cert_file_list=c_l, excl_cert_file_list=['/file/path/x'])
        get_certs_mock.assert_called_once_with(None)

    @mock.patch.object(ilo_common, '_get_certificate_file_list',
                       spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_clear_certificates_default(self, get_ilo_object_mock,
                                        get_certs_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        get_certs_mock.return_value = ['/file/path/x']

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.clear_certificates(task)

        ilo_mock_object.remove_tls_certificate.assert_called_once_with(
            cert_file_list=None, excl_cert_file_list=['/file/path/x'])
        get_certs_mock.assert_called_once_with(None)

    @mock.patch.object(ilo_common, '_get_certificate_file_list',
                       spec_set=True, autospec=True)
    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_clear_certificates_raises_ilo_error(self, get_ilo_object_mock,
                                                 get_certs_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        c_l = ['/file/path/a', '/file/path/b']
        get_certs_mock.return_value = ['/file/path/x']
        exc = ilo_error.IloError('error')
        ilo_mock_object.remove_tls_certificate.side_effect = exc

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.clear_certificates,
                              task, c_l)

        ilo_mock_object.remove_tls_certificate.assert_called_once_with(
            cert_file_list=c_l, excl_cert_file_list=['/file/path/x'])
        get_certs_mock.assert_called_once_with(None)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_setup_uefi_https_scheme_http(self, get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value
        iso = "http://1.1.1.1/image.iso"

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationNotSupported,
                              ilo_common.setup_uefi_https,
                              task, iso, True)

        ilo_mock_object.set_http_boot_url.assert_not_called()
        ilo_mock_object.set_one_time_boot.assert_not_called()
        ilo_mock_object.update_persistent_boot.assert_not_called()

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def _test_setup_uefi_https(self, get_ilo_object_mock, persistent):

        ilo_mock_object = get_ilo_object_mock.return_value

        iso = "https://1.1.1.1/image.iso"

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            ilo_common.setup_uefi_https(task, iso, persistent=persistent)

        ilo_mock_object.set_http_boot_url.assert_called_once_with(iso)
        if persistent:
            ilo_mock_object.set_one_time_boot.assert_not_called()
            ilo_mock_object.update_persistent_boot.assert_called_once_with(
                ['UEFIHTTP'])
        else:
            ilo_mock_object.set_one_time_boot.assert_called_once_with(
                'UEFIHTTP')
            ilo_mock_object.update_persistent_boot.assert_not_called()

    def test_setup_uefi_https_persistent_true(self):
        self._test_setup_uefi_https(persistent=True)

    def test_setup_uefi_https_persistent_false(self):
        self._test_setup_uefi_https(persistent=False)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_setup_uefi_https_raises_not_supported(self, get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value

        exc = ilo_error.IloCommandNotSupportedInBiosError('error')
        ilo_mock_object.set_http_boot_url.side_effect = exc

        iso = "https://1.1.1.1/image.iso"

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationNotSupported,
                              ilo_common.setup_uefi_https,
                              task, iso, True)

    @mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
                       autospec=True)
    def test_setup_uefi_https_raises_ilo_error(self, get_ilo_object_mock):
        ilo_mock_object = get_ilo_object_mock.return_value

        exc = ilo_error.IloError('error')
        ilo_mock_object.set_http_boot_url.side_effect = exc

        iso = "https://1.1.1.1/image.iso"

        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=False) as task:
            self.assertRaises(exception.IloOperationError,
                              ilo_common.setup_uefi_https,
                              task, iso, True)