path: root/ironic/tests/unit/drivers/modules/test_ipxe.py

# Copyright 2013 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 iPXE driver."""

import os
from unittest import mock

from oslo_config import cfg
from oslo_utils import uuidutils

from ironic.common import boot_devices
from ironic.common import boot_modes
from ironic.common import dhcp_factory
from ironic.common import exception
from ironic.common.glance_service import image_service
from ironic.common import pxe_utils
from ironic.common import states
from ironic.common import utils as common_utils
from ironic.conductor import task_manager
from ironic.conductor import utils as manager_utils
from ironic.drivers import base as drivers_base
from ironic.drivers.modules import agent_base
from ironic.drivers.modules import boot_mode_utils
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules import ipxe
from ironic.drivers.modules import pxe_base
from ironic.drivers.modules.storage import noop as noop_storage
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

CONF = cfg.CONF

INST_INFO_DICT = db_utils.get_test_pxe_instance_info()
DRV_INFO_DICT = db_utils.get_test_pxe_driver_info()
DRV_INTERNAL_INFO_DICT = db_utils.get_test_pxe_driver_internal_info()


# NOTE(TheJulia): This code is essentially a bulk copy of the
# test_pxe file with some contextual modifications to enforce
# use of ipxe while also explicitly having it globally disabled
# in the conductor.
@mock.patch.object(ipxe.iPXEBoot, '__init__', lambda self: None)
class iPXEBootTestCase(db_base.DbTestCase):

    driver = 'fake-hardware'
    boot_interface = 'ipxe'
    driver_info = DRV_INFO_DICT
    driver_internal_info = DRV_INTERNAL_INFO_DICT

    def setUp(self):
        super(iPXEBootTestCase, self).setUp()
        self.context.auth_token = 'fake'
        self.config_temp_dir('tftp_root', group='pxe')
        self.config_temp_dir('images_path', group='pxe')
        self.config_temp_dir('http_root', group='deploy')
        self.config(group='deploy', http_url='http://myserver')
        instance_info = INST_INFO_DICT

        self.config(enabled_boot_interfaces=[self.boot_interface,
                                             'ipxe', 'fake'])
        self.node = obj_utils.create_test_node(
            self.context,
            driver=self.driver,
            boot_interface=self.boot_interface,
            # Avoid fake properties in get_properties() output
            vendor_interface='no-vendor',
            instance_info=instance_info,
            driver_info=self.driver_info,
            driver_internal_info=self.driver_internal_info)
        self.port = obj_utils.create_test_port(self.context,
                                               node_id=self.node.id)

    def test_get_properties(self):
        expected = pxe_base.COMMON_PROPERTIES
        expected.update(agent_base.VENDOR_PROPERTIES)
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            self.assertEqual(expected, task.driver.get_properties())

    @mock.patch.object(image_service.GlanceImageService, 'show', autospec=True)
    def test_validate_good(self, mock_glance):
        mock_glance.return_value = {'properties': {'kernel_id': 'fake-kernel',
                                                   'ramdisk_id': 'fake-initr'}}
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            task.driver.boot.validate(task)

    @mock.patch.object(image_service.GlanceImageService, 'show', autospec=True)
    def test_validate_good_whole_disk_image(self, mock_glance):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            task.node.driver_internal_info['is_whole_disk_image'] = True
            task.driver.boot.validate(task)

    @mock.patch.object(image_service.GlanceImageService, 'show', autospec=True)
    @mock.patch.object(noop_storage.NoopStorage, 'should_write_image',
                       autospec=True)
    def test_validate_skip_check_write_image_false(self, mock_write,
                                                   mock_glance):
        mock_write.return_value = False
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            task.driver.boot.validate(task)
        self.assertFalse(mock_glance.called)

    def test_validate_fail_missing_deploy_kernel(self):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            del task.node.driver_info['deploy_kernel']
            self.assertRaises(exception.MissingParameterValue,
                              task.driver.boot.validate, task)

    def test_validate_fail_missing_deploy_ramdisk(self):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            del task.node.driver_info['deploy_ramdisk']
            self.assertRaises(exception.MissingParameterValue,
                              task.driver.boot.validate, task)

    @mock.patch.object(image_service.GlanceImageService, 'show', autospec=True)
    @mock.patch('ironic.drivers.modules.deploy_utils.get_boot_option',
                autospec=True)
    def test_validate_with_boot_iso(self, mock_boot_option, mock_glance):
        self.node.instance_info = {
            'boot_iso': "glance://image"
        }
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            task.driver.boot.validate(task)
            mock_boot_option.assert_called_with(task.node)
            mock_glance.assert_not_called()

    @mock.patch('ironic.drivers.modules.deploy_utils.get_boot_option',
                return_value='ramdisk', autospec=True)
    def test_validate_with_boot_iso_and_image_source(self, mock_boot_option):
        i_info = self.node.instance_info
        i_info['image_source'] = "http://localhost:1234/image"
        i_info['boot_iso'] = "http://localhost:1234/boot.iso"
        self.node.instance_info = i_info
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            self.assertRaises(exception.InvalidParameterValue,
                              task.driver.boot.validate,
                              task)
            mock_boot_option.assert_called_once_with(task.node)

    @mock.patch('ironic.drivers.modules.deploy_utils.get_boot_option',
                return_value='local', autospec=True)
    def test_validate_no_image_source_for_local_boot(self, mock_boot_option):
        with task_manager.acquire(self.context, self.node.uuid,
                                  shared=True) as task:
            del task.node['instance_info']['image_source']
            task.driver.boot.validate(task)
            mock_boot_option.assert_called_with(task.node)

    def test_validate_fail_no_port(self):
        new_node = obj_utils.create_test_node(
            self.context,
            uuid='aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee',
            driver=self.driver, boot_interface=self.boot_interface,
            instance_info=INST_INFO_DICT, driver_info=DRV_INFO_DICT)
        with task_manager.acquire(self.context, new_node.uuid,
                                  shared=True) as task:
            self.assertRaises(exception.MissingParameterValue,
                              task.driver.boot.validate, task)

    def test_validate_inspection(self):
        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.driver.boot.validate_inspection(task)

    def test_validate_inspection_no_inspection_ramdisk(self):
        driver_info = self.node.driver_info
        del driver_info['deploy_ramdisk']
        self.node.driver_info = driver_info
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            self.assertRaises(exception.UnsupportedDriverExtension,
                              task.driver.boot.validate_inspection, task)

    # TODO(TheJulia): Many of the interfaces mocked below are private PXE
    # interface methods. As time progresses, these will need to be migrated
    # and refactored as we begin to separate PXE and iPXE interfaces.
    @mock.patch.object(manager_utils, 'node_get_boot_mode', autospec=True)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    @mock.patch.object(pxe_utils, 'get_image_info', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'build_pxe_config_options', autospec=True)
    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    def _test_prepare_ramdisk(self, mock_pxe_config,
                              mock_build_pxe, mock_cache_r_k,
                              mock_deploy_img_info,
                              mock_instance_img_info,
                              dhcp_factory_mock,
                              set_boot_device_mock,
                              get_boot_mode_mock,
                              uefi=False,
                              cleaning=False,
                              ipxe_use_swift=False,
                              whole_disk_image=False,
                              mode='deploy',
                              node_boot_mode=None,
                              persistent=False):
        mock_build_pxe.return_value = {}
        kernel_label = '%s_kernel' % mode
        ramdisk_label = '%s_ramdisk' % mode
        mock_deploy_img_info.return_value = {kernel_label: 'a',
                                             ramdisk_label: 'r'}
        if whole_disk_image:
            mock_instance_img_info.return_value = {}
        else:
            mock_instance_img_info.return_value = {'kernel': 'b'}
        mock_pxe_config.return_value = None
        mock_cache_r_k.return_value = None
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        get_boot_mode_mock.return_value = node_boot_mode
        driver_internal_info = self.node.driver_internal_info
        driver_internal_info['is_whole_disk_image'] = whole_disk_image
        self.node.driver_internal_info = driver_internal_info
        if mode == 'rescue':
            mock_deploy_img_info.return_value = {
                'rescue_kernel': 'a',
                'rescue_ramdisk': 'r'}
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            dhcp_opts = pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=4)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            task.driver.boot.prepare_ramdisk(task, {'foo': 'bar'})
            mock_deploy_img_info.assert_called_once_with(task.node, mode=mode,
                                                         ipxe_enabled=True)
            provider_mock.update_dhcp.assert_called_once_with(
                task, dhcp_opts)
            if self.node.provision_state == states.DEPLOYING:
                get_boot_mode_mock.assert_called_once_with(task)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=persistent)
            if ipxe_use_swift:
                if whole_disk_image:
                    self.assertFalse(mock_cache_r_k.called)
                else:
                    mock_cache_r_k.assert_called_once_with(
                        task, {'kernel': 'b'},
                        ipxe_enabled=True)
                mock_instance_img_info.assert_called_once_with(
                    task, ipxe_enabled=True)
            elif not cleaning and mode == 'deploy':
                mock_cache_r_k.assert_called_once_with(
                    task,
                    {'deploy_kernel': 'a', 'deploy_ramdisk': 'r',
                     'kernel': 'b'},
                    ipxe_enabled=True)
                mock_instance_img_info.assert_called_once_with(
                    task, ipxe_enabled=True)
            elif mode == 'deploy':
                mock_cache_r_k.assert_called_once_with(
                    task, {'deploy_kernel': 'a', 'deploy_ramdisk': 'r'},
                    ipxe_enabled=True)
            elif mode == 'rescue':
                mock_cache_r_k.assert_called_once_with(
                    task, {'rescue_kernel': 'a', 'rescue_ramdisk': 'r'},
                    ipxe_enabled=True)
            mock_pxe_config.assert_called_once_with(
                task, {}, CONF.pxe.ipxe_config_template,
                ipxe_enabled=True)

    def test_prepare_ramdisk(self):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        self._test_prepare_ramdisk()

    def test_prepare_ramdisk_rescue(self):
        self.node.provision_state = states.RESCUING
        self.node.save()
        self._test_prepare_ramdisk(mode='rescue')

    def test_prepare_ramdisk_uefi(self):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        properties = self.node.properties
        properties['capabilities'] = 'boot_mode:uefi'
        self.node.properties = properties
        self.node.save()
        self._test_prepare_ramdisk(uefi=True)

    @mock.patch.object(os.path, 'isfile', lambda path: True)
    @mock.patch.object(common_utils, 'file_has_content', lambda *args: False)
    @mock.patch('ironic.common.utils.write_to_file', autospec=True)
    @mock.patch('ironic.common.utils.render_template', autospec=True)
    def test_prepare_ramdisk_ipxe_with_copy_file_different(
            self, render_mock, write_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        render_mock.return_value = 'foo'
        self._test_prepare_ramdisk()
        write_mock.assert_called_once_with(
            os.path.join(
                CONF.deploy.http_root,
                os.path.basename(CONF.pxe.ipxe_boot_script)),
            'foo', 0o644)
        render_mock.assert_called_once_with(
            CONF.pxe.ipxe_boot_script,
            {'ipxe_for_mac_uri': 'pxelinux.cfg/',
             'ipxe_fallback_script': None})

    @mock.patch.object(os.path, 'isfile', lambda path: False)
    @mock.patch('ironic.common.utils.file_has_content', autospec=True)
    @mock.patch('ironic.common.utils.write_to_file', autospec=True)
    @mock.patch('ironic.common.utils.render_template', autospec=True)
    def test_prepare_ramdisk_ipxe_with_copy_no_file(
            self, render_mock, write_mock, file_has_content_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        render_mock.return_value = 'foo'
        self._test_prepare_ramdisk()
        self.assertFalse(file_has_content_mock.called)
        write_mock.assert_called_once_with(
            os.path.join(
                CONF.deploy.http_root,
                os.path.basename(CONF.pxe.ipxe_boot_script)),
            'foo', 0o644)
        render_mock.assert_called_once_with(
            CONF.pxe.ipxe_boot_script,
            {'ipxe_for_mac_uri': 'pxelinux.cfg/',
             'ipxe_fallback_script': None})

    @mock.patch.object(os.path, 'isfile', lambda path: True)
    @mock.patch.object(common_utils, 'file_has_content', lambda *args: True)
    @mock.patch('ironic.common.utils.write_to_file', autospec=True)
    @mock.patch('ironic.common.utils.render_template', autospec=True)
    def test_prepare_ramdisk_ipxe_without_copy(
            self, render_mock, write_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        self._test_prepare_ramdisk()
        self.assertFalse(write_mock.called)

    @mock.patch.object(common_utils, 'render_template', lambda *args: 'foo')
    @mock.patch('ironic.common.utils.write_to_file', autospec=True)
    def test_prepare_ramdisk_ipxe_swift(self, write_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        self.config(group='pxe', ipxe_use_swift=True)
        self._test_prepare_ramdisk(ipxe_use_swift=True)
        write_mock.assert_called_once_with(
            os.path.join(
                CONF.deploy.http_root,
                os.path.basename(CONF.pxe.ipxe_boot_script)),
            'foo', 0o644)

    @mock.patch.object(common_utils, 'render_template', lambda *args: 'foo')
    @mock.patch('ironic.common.utils.write_to_file', autospec=True)
    def test_prepare_ramdisk_ipxe_swift_whole_disk_image(
            self, write_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        self.config(group='pxe', ipxe_use_swift=True)
        self._test_prepare_ramdisk(ipxe_use_swift=True, whole_disk_image=True)
        write_mock.assert_called_once_with(
            os.path.join(
                CONF.deploy.http_root,
                os.path.basename(CONF.pxe.ipxe_boot_script)),
            'foo', 0o644)

    def test_prepare_ramdisk_cleaning(self):
        self.node.provision_state = states.CLEANING
        self.node.save()
        self._test_prepare_ramdisk(cleaning=True)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_set_boot_mode_on_bm(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        properties = self.node.properties
        properties['capabilities'] = 'boot_mode:uefi'
        self.node.properties = properties
        self.node.save()
        self._test_prepare_ramdisk(uefi=True)
        set_boot_mode_mock.assert_called_once_with(mock.ANY, boot_modes.UEFI)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_set_boot_mode_on_ironic(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        self._test_prepare_ramdisk(node_boot_mode=boot_modes.LEGACY_BIOS)

        with task_manager.acquire(self.context, self.node.uuid) as task:
            driver_internal_info = task.node.driver_internal_info
            self.assertIn('deploy_boot_mode', driver_internal_info)
            self.assertEqual(boot_modes.LEGACY_BIOS,
                             driver_internal_info['deploy_boot_mode'])
            self.assertEqual(set_boot_mode_mock.call_count, 0)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_set_default_boot_mode_on_ironic_bios(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()

        self.config(default_boot_mode=boot_modes.LEGACY_BIOS, group='deploy')

        self._test_prepare_ramdisk()

        with task_manager.acquire(self.context, self.node.uuid) as task:
            driver_internal_info = task.node.driver_internal_info
            self.assertIn('deploy_boot_mode', driver_internal_info)
            self.assertEqual(boot_modes.LEGACY_BIOS,
                             driver_internal_info['deploy_boot_mode'])
            self.assertEqual(set_boot_mode_mock.call_count, 1)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_set_default_boot_mode_on_ironic_uefi(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()

        self.config(default_boot_mode=boot_modes.UEFI, group='deploy')

        self._test_prepare_ramdisk(uefi=True)

        with task_manager.acquire(self.context, self.node.uuid) as task:
            driver_internal_info = task.node.driver_internal_info
            self.assertIn('deploy_boot_mode', driver_internal_info)
            self.assertEqual(boot_modes.UEFI,
                             driver_internal_info['deploy_boot_mode'])
            self.assertEqual(set_boot_mode_mock.call_count, 1)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_conflicting_boot_modes(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        properties = self.node.properties
        properties['capabilities'] = 'boot_mode:uefi'
        self.node.properties = properties
        self.node.save()
        self._test_prepare_ramdisk(uefi=True,
                                   node_boot_mode=boot_modes.LEGACY_BIOS)
        set_boot_mode_mock.assert_called_once_with(mock.ANY, boot_modes.UEFI)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_conflicting_boot_modes_set_unsupported(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        properties = self.node.properties
        properties['capabilities'] = 'boot_mode:uefi'
        self.node.properties = properties
        self.node.save()
        set_boot_mode_mock.side_effect = exception.UnsupportedDriverExtension(
            extension='management', driver='test-driver'
        )
        self.assertRaises(exception.UnsupportedDriverExtension,
                          self._test_prepare_ramdisk,
                          uefi=True, node_boot_mode=boot_modes.LEGACY_BIOS)

    @mock.patch.object(manager_utils, 'node_set_boot_mode', autospec=True)
    def test_prepare_ramdisk_set_boot_mode_not_called(
            self, set_boot_mode_mock):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        properties = self.node.properties
        properties['capabilities'] = 'boot_mode:uefi'
        self.node.properties = properties
        self.node.save()
        self._test_prepare_ramdisk(uefi=True, node_boot_mode=boot_modes.UEFI)
        self.assertEqual(set_boot_mode_mock.call_count, 0)

    @mock.patch.object(pxe_utils, 'clean_up_pxe_env', autospec=True)
    @mock.patch.object(pxe_utils, 'get_image_info', autospec=True)
    def _test_clean_up_ramdisk(self, get_image_info_mock,
                               clean_up_pxe_env_mock, mode='deploy'):
        with task_manager.acquire(self.context, self.node.uuid) as task:
            kernel_label = '%s_kernel' % mode
            ramdisk_label = '%s_ramdisk' % mode
            image_info = {kernel_label: ['', '/path/to/' + kernel_label],
                          ramdisk_label: ['', '/path/to/' + ramdisk_label]}
            get_image_info_mock.return_value = image_info
            task.driver.boot.clean_up_ramdisk(task)
            clean_up_pxe_env_mock.assert_called_once_with(
                task, image_info, ipxe_enabled=True)
            get_image_info_mock.assert_called_once_with(
                task.node, mode=mode, ipxe_enabled=True)

    def test_clean_up_ramdisk(self):
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        self._test_clean_up_ramdisk()

    def test_clean_up_ramdisk_rescue(self):
        self.node.provision_state = states.RESCUING
        self.node.save()
        self._test_clean_up_ramdisk(mode='rescue')

    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_ramdisk(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            set_boot_device_mock, mock_create_pxe_config):
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        image_info = {'kernel': ('', '/path/to/kernel'),
                      'ramdisk': ('', '/path/to/ramdisk')}
        i_info_caps = {"boot_mode": "bios"}
        kernel_arg = "meow"
        get_image_info_mock.return_value = image_info

        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.node.properties['capabilities'] = 'boot_mode:bios'
            i_info = task.node.instance_info
            i_info['capabilities'] = i_info_caps
            i_info['kernel_append_params'] = kernel_arg
            task.node.instance_info = i_info
            task.node.deploy_interface = 'ramdisk'
            task.node.save()
            dhcp_opts = pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)

            task.driver.boot.prepare_instance(task)
            get_image_info_mock.assert_called_once_with(
                task, ipxe_enabled=True)
            cache_mock.assert_called_once_with(task, image_info,
                                               ipxe_enabled=True)
            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, None,
                'bios', False, iscsi_boot=False, ramdisk_boot=True,
                ipxe_enabled=True, anaconda_boot=False)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=True)
            expected_params = {
                'aki_path': 'http://myserver/' + task.node.uuid + '/kernel',
                'ari_path': 'http://myserver/' + task.node.uuid + '/ramdisk',
                'pxe_append_params': 'meow ipa-debug=1 ipa-global-request-id'
                                     '=' + task.context.request_id,
                'tftp_server': mock.ANY,
                'ipxe_timeout': 0}
            mock_create_pxe_config.assert_called_once_with(
                task, expected_params, mock.ANY, ipxe_enabled=True)

    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_ramdisk_bios(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            set_boot_device_mock, mock_create_pxe_config):
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        image_info = {'kernel': ('', '/path/to/kernel'),
                      'ramdisk': ('', '/path/to/ramdisk')}
        kernel_arg = "meow"
        get_image_info_mock.return_value = image_info

        with task_manager.acquire(self.context, self.node.uuid) as task:
            i_info = task.node.instance_info
            i_info['kernel_append_params'] = kernel_arg
            task.node.instance_info = i_info
            task.node.deploy_interface = 'ramdisk'
            task.node.save()
            dhcp_opts = pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)

            task.driver.boot.prepare_instance(task)
            get_image_info_mock.assert_called_once_with(
                task, ipxe_enabled=True)
            cache_mock.assert_called_once_with(task, image_info,
                                               ipxe_enabled=True)
            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, None,
                'uefi', False, iscsi_boot=False, ramdisk_boot=True,
                ipxe_enabled=True, anaconda_boot=False)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=True)
            expected_params = {
                'aki_path': 'http://myserver/' + task.node.uuid + '/kernel',
                'ari_path': 'http://myserver/' + task.node.uuid + '/ramdisk',
                'pxe_append_params': 'meow ipa-debug=1 ipa-global-request-id'
                                     '=' + task.context.request_id,
                'tftp_server': mock.ANY,
                'ipxe_timeout': 0}
            mock_create_pxe_config.assert_called_once_with(
                task, expected_params, mock.ANY, ipxe_enabled=True)

    @mock.patch('os.path.isfile', lambda filename: False)
    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    @mock.patch.object(deploy_utils, 'is_iscsi_boot', lambda task: True)
    @mock.patch.object(noop_storage.NoopStorage, 'should_write_image',
                       lambda task: False)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_netboot_iscsi_bios(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            set_boot_device_mock, create_pxe_config_mock):
        http_url = 'http://192.1.2.3:1234'
        self.config(http_url=http_url, group='deploy')
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        vol_id = uuidutils.generate_uuid()
        obj_utils.create_test_volume_target(
            self.context, node_id=self.node.id, volume_type='iscsi',
            boot_index=0, volume_id='1234', uuid=vol_id,
            properties={'target_lun': 0,
                        'target_portal': 'fake_host:3260',
                        'target_iqn': 'fake_iqn',
                        'auth_username': 'fake_username',
                        'auth_password': 'fake_password'})
        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.node.driver_internal_info = {
                'boot_from_volume': vol_id}
            task.node.properties['capabilities'] = 'boot_mode:bios'
            task.node.instance_info['capabilities'] = {'boot_mode': 'bios'}
            dhcp_opts = pxe_utils.dhcp_options_for_instance(task,
                                                            ipxe_enabled=True)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)

            task.driver.boot.prepare_instance(task)
            self.assertFalse(get_image_info_mock.called)
            self.assertFalse(cache_mock.called)
            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
            create_pxe_config_mock.assert_called_once_with(
                task, mock.ANY, CONF.pxe.ipxe_config_template,
                ipxe_enabled=True)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, None, boot_modes.LEGACY_BIOS, False,
                ipxe_enabled=True, iscsi_boot=True, ramdisk_boot=False,
                anaconda_boot=False)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=True)

    @mock.patch('os.path.isfile', lambda filename: False)
    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    @mock.patch.object(deploy_utils, 'is_iscsi_boot', lambda task: True)
    @mock.patch.object(noop_storage.NoopStorage, 'should_write_image',
                       lambda task: False)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_netboot_iscsi(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            set_boot_device_mock, create_pxe_config_mock):
        http_url = 'http://192.1.2.3:1234'
        self.config(http_url=http_url, group='deploy')
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        vol_id = uuidutils.generate_uuid()
        obj_utils.create_test_volume_target(
            self.context, node_id=self.node.id, volume_type='iscsi',
            boot_index=0, volume_id='1234', uuid=vol_id,
            properties={'target_lun': 0,
                        'target_portal': 'fake_host:3260',
                        'target_iqn': 'fake_iqn',
                        'auth_username': 'fake_username',
                        'auth_password': 'fake_password'})

        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.node.driver_internal_info = {
                'boot_from_volume': vol_id}
            dhcp_opts = pxe_utils.dhcp_options_for_instance(task,
                                                            ipxe_enabled=True)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)

            task.driver.boot.prepare_instance(task)
            self.assertFalse(get_image_info_mock.called)
            self.assertFalse(cache_mock.called)
            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
            create_pxe_config_mock.assert_called_once_with(
                task, mock.ANY, CONF.pxe.ipxe_config_template,
                ipxe_enabled=True)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, None, boot_modes.UEFI, False,
                ipxe_enabled=True, iscsi_boot=True, ramdisk_boot=False,
                anaconda_boot=False)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=True)

    @mock.patch('os.path.isfile', lambda filename: False)
    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_netboot_ramdisk(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            set_boot_device_mock, create_pxe_config_mock):
        http_url = 'http://192.1.2.3:1234'
        self.config(http_url=http_url, group='deploy')
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        self.node.deploy_interface = 'ramdisk'
        self.node.instance_info = {'boot_iso': 'http://1.2.3.4:1234/boot.iso'}
        image_info = {'kernel': ('', '/path/to/kernel'),
                      'deploy_kernel': ('', '/path/to/kernel'),
                      'ramdisk': ('', '/path/to/ramdisk'),
                      'deploy_ramdisk': ('', '/path/to/ramdisk')}
        get_image_info_mock.return_value = image_info
        self.node.provision_state = states.DEPLOYING
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            dhcp_opts = pxe_utils.dhcp_options_for_instance(task,
                                                            ipxe_enabled=True)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)
            task.driver.boot.prepare_instance(task)
            self.assertTrue(get_image_info_mock.called)
            self.assertTrue(cache_mock.called)
            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
            create_pxe_config_mock.assert_called_once_with(
                task, mock.ANY, CONF.pxe.ipxe_config_template,
                ipxe_enabled=True)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, None, boot_modes.UEFI, False,
                ipxe_enabled=True, iscsi_boot=False, ramdisk_boot=True,
                anaconda_boot=False)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=True)

    @mock.patch('os.path.isfile', lambda filename: False)
    @mock.patch.object(pxe_utils, 'create_pxe_config', autospec=True)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_ramdisk_with_kernel_arg(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            set_boot_device_mock, create_pxe_config_mock):
        http_url = 'http://192.1.2.3:1234'
        self.config(http_url=http_url, group='deploy')
        self.config(enabled_deploy_interfaces='ramdisk')
        provider_mock = mock.MagicMock()
        dhcp_factory_mock.return_value = provider_mock
        self.node.instance_info = {'ramdisk_kernel_arguments': 'cat meow'}
        image_info = {'kernel': ('', '/path/to/kernel'),
                      'deploy_kernel': ('', '/path/to/kernel'),
                      'ramdisk': ('', '/path/to/ramdisk'),
                      'deploy_ramdisk': ('', '/path/to/ramdisk')}
        get_image_info_mock.return_value = image_info
        self.node.provision_state = states.DEPLOYING
        self.node.deploy_interface = 'ramdisk'
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            dhcp_opts = pxe_utils.dhcp_options_for_instance(task,
                                                            ipxe_enabled=True)
            dhcp_opts += pxe_utils.dhcp_options_for_instance(
                task, ipxe_enabled=True, ip_version=6)
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)
            task.driver.boot.prepare_instance(task)
            self.assertTrue(get_image_info_mock.called)
            self.assertTrue(cache_mock.called)
            uuid = self.node.uuid
            expected_params = {
                'aki_path': 'http://192.1.2.3:1234/' + uuid + '/kernel',
                'ari_path': 'http://192.1.2.3:1234/' + uuid + '/ramdisk',
                'ramdisk_opts': 'cat meow',
                'pxe_append_params': 'nofb vga=normal ipa-debug=1 '
                                     'ipa-global-request-'
                                     'id=' + task.context.request_id,
                'tftp_server': mock.ANY,
                'ipxe_timeout': 0
            }
            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
            create_pxe_config_mock.assert_called_once_with(
                task, expected_params, CONF.pxe.ipxe_config_template,
                ipxe_enabled=True)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, None, boot_modes.UEFI, False,
                ipxe_enabled=True, iscsi_boot=False, ramdisk_boot=True,
                anaconda_boot=False)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.PXE,
                                                         persistent=True)

    @mock.patch.object(boot_mode_utils, 'configure_secure_boot_if_needed',
                       autospec=True)
    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(pxe_utils, 'clean_up_pxe_config', autospec=True)
    def test_prepare_instance(self, clean_up_pxe_config_mock,
                              set_boot_device_mock, secure_boot_mock):
        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.driver.boot.prepare_instance(task)
            clean_up_pxe_config_mock.assert_called_once_with(
                task, ipxe_enabled=True)
            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.DISK,
                                                         persistent=True)
            secure_boot_mock.assert_called_once_with(task)

    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(pxe_utils, 'clean_up_pxe_config', autospec=True)
    def test_prepare_instance_active(self, clean_up_pxe_config_mock,
                                     set_boot_device_mock):
        self.node.provision_state = states.ACTIVE
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.driver.boot.prepare_instance(task)
            clean_up_pxe_config_mock.assert_called_once_with(
                task, ipxe_enabled=True)
            self.assertFalse(set_boot_device_mock.called)

    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
    @mock.patch.object(pxe_utils, 'clean_up_pxe_config', autospec=True)
    @mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
    @mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
    @mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_prepare_instance_with_fallback(
            self, get_image_info_mock, cache_mock,
            dhcp_factory_mock, switch_pxe_config_mock,
            clean_up_pxe_config_mock, set_boot_device_mock):
        self.config(enable_netboot_fallback=True, group='pxe')
        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.node.driver_internal_info['root_uuid_or_disk_id'] = (
                "30212642-09d3-467f-8e09-21685826ab50")
            task.node.driver_internal_info['is_whole_disk_image'] = False
            pxe_config_path = pxe_utils.get_pxe_config_file_path(
                task.node.uuid, ipxe_enabled=True)

            task.driver.boot.prepare_instance(task)

            set_boot_device_mock.assert_called_once_with(task,
                                                         boot_devices.DISK,
                                                         persistent=True)
            switch_pxe_config_mock.assert_called_once_with(
                pxe_config_path, "30212642-09d3-467f-8e09-21685826ab50",
                'uefi', True, iscsi_boot=False, ramdisk_boot=False,
                ipxe_enabled=True, anaconda_boot=False)
            # No clean up
            self.assertFalse(clean_up_pxe_config_mock.called)
            # No netboot configuration beyond the PXE files
            self.assertFalse(get_image_info_mock.called)
            self.assertFalse(cache_mock.called)
            self.assertFalse(dhcp_factory_mock.return_value.update_dhcp.called)

    @mock.patch.object(boot_mode_utils, 'deconfigure_secure_boot_if_needed',
                       autospec=True)
    @mock.patch.object(pxe_utils, 'clean_up_pxe_env', autospec=True)
    @mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
    def test_clean_up_instance(self, get_image_info_mock,
                               clean_up_pxe_env_mock,
                               secure_boot_mock):
        with task_manager.acquire(self.context, self.node.uuid) as task:
            image_info = {'kernel': ['', '/path/to/kernel'],
                          'ramdisk': ['', '/path/to/ramdisk']}
            get_image_info_mock.return_value = image_info
            task.driver.boot.clean_up_instance(task)
            clean_up_pxe_env_mock.assert_called_once_with(
                task, image_info, ipxe_enabled=True)
            get_image_info_mock.assert_called_once_with(
                task, ipxe_enabled=True)
            secure_boot_mock.assert_called_once_with(task)


@mock.patch.object(ipxe.iPXEBoot, '__init__', lambda self: None)
class iPXEValidateRescueTestCase(db_base.DbTestCase):

    def setUp(self):
        super(iPXEValidateRescueTestCase, self).setUp()
        for iface in drivers_base.ALL_INTERFACES:
            impl = 'fake'
            if iface == 'network':
                impl = 'flat'
            if iface == 'rescue':
                impl = 'agent'
            if iface == 'boot':
                impl = 'ipxe'
            config_kwarg = {'enabled_%s_interfaces' % iface: [impl],
                            'default_%s_interface' % iface: impl}
            self.config(**config_kwarg)
        self.config(enabled_hardware_types=['fake-hardware'])
        driver_info = DRV_INFO_DICT
        driver_info.update({'rescue_ramdisk': 'my_ramdisk',
                            'rescue_kernel': 'my_kernel'})
        instance_info = INST_INFO_DICT
        instance_info.update({'rescue_password': 'password'})
        n = {
            'driver': 'fake-hardware',
            'instance_info': instance_info,
            'driver_info': driver_info,
            'driver_internal_info': DRV_INTERNAL_INFO_DICT,
        }
        self.node = obj_utils.create_test_node(self.context, **n)

    def test_validate_rescue(self):
        with task_manager.acquire(self.context, self.node.uuid) as task:
            task.driver.boot.validate_rescue(task)

    def test_validate_rescue_no_rescue_ramdisk(self):
        driver_info = self.node.driver_info
        del driver_info['rescue_ramdisk']
        self.node.driver_info = driver_info
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            self.assertRaisesRegex(exception.MissingParameterValue,
                                   'Missing.*rescue_ramdisk',
                                   task.driver.boot.validate_rescue, task)

    def test_validate_rescue_fails_no_rescue_kernel(self):
        driver_info = self.node.driver_info
        del driver_info['rescue_kernel']
        self.node.driver_info = driver_info
        self.node.save()
        with task_manager.acquire(self.context, self.node.uuid) as task:
            self.assertRaisesRegex(exception.MissingParameterValue,
                                   'Missing.*rescue_kernel',
                                   task.driver.boot.validate_rescue, task)