ipxe boot interface

A long time ago, in a galaxy not so far away,
the ironic comunity reached consensus that we
should refactor the PXE interface such that we
had separate PXE and iPXE interfaces.

In looking at what it would take for something like
a standalone deployment to have multiarchitecture
support in their environment, it seems even more
urgent and necessary that we begin to make this
delineation. This is because while some ARM iPXE
support exists, the binary is not a commonly
shipped artifiact, so operators are largely
only left with grubaa64.efi as their bootloader.

A similar issue exists with ppc64le deployments
where they must disable iPXE, as the ppc64le
hardware expects reading a syslinux compatible
file, similarlly no iPXE loader exists.

To start this effort, we need to promote iPXE
functionality to a dedicated interface,
and remove the necessity of setting the
[pxe]ipxe_enabled setting.

Next steps, beyond this patch, would be to
begin to tease out the common private method
code in the underlying PXE interface that both
the iPXE and PXE interfaces use, and appropriately
relocate and refactor that code as necesary.

During this process, we can create an [ipxe]
configuration section, and migrate settings.

Finally, once the deprecation cycle is complete,
we will be able to remove the ipxe logic with-in
the PXE interface.

Change-Id: I392616417c48986e84e50a3ddc7567344bfe3571
Story: #1628069
Task: #10516

1 files changed, 824 insertions, 0 deletions

diff --git a/ironic/tests/unit/drivers/modules/test_ipxe.py b/ironic/tests/unit/drivers/modules/test_ipxe.py
new file mode 100644
index 000000000..64108b1e6
--- /dev/null
+++ b/ironic/tests/unit/drivers/modules/test_ipxe.py
@@ -0,0 +1,824 @@
+# 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
+
+import mock
+from oslo_config import cfg
+from oslo_serialization import jsonutils as json
+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 base_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_vendor
+from ironic.drivers.modules import deploy_utils
+from ironic.drivers.modules import ipxe
+from ironic.drivers.modules import pxe
+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)
+        self.config(group='conductor', api_url='http://127.0.0.1:1234/')
+
+    def test_get_properties(self):
+        expected = ipxe.COMMON_PROPERTIES
+        expected.update(agent_base_vendor.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(base_image_service.BaseImageService, '_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(base_image_service.BaseImageService, '_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(base_image_service.BaseImageService, '_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)
+
+    def test_validate_fail_missing_image_source(self):
+        info = dict(INST_INFO_DICT)
+        del info['image_source']
+        self.node.instance_info = json.dumps(info)
+        with task_manager.acquire(self.context, self.node.uuid,
+                                  shared=True) as task:
+            task.node['instance_info'] = json.dumps(info)
+            self.assertRaises(exception.MissingParameterValue,
+                              task.driver.boot.validate, task)
+
+    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_fail_trusted_boot_with_secure_boot(self):
+        instance_info = {"boot_option": "netboot",
+                         "secure_boot": "true",
+                         "trusted_boot": "true"}
+        properties = {'capabilities': 'trusted_boot:true'}
+        with task_manager.acquire(self.context, self.node.uuid,
+                                  shared=True) as task:
+            task.node.instance_info['capabilities'] = instance_info
+            task.node.properties = properties
+            task.node.driver_internal_info['is_whole_disk_image'] = False
+            self.assertRaises(exception.InvalidParameterValue,
+                              task.driver.boot.validate, task)
+
+    def test_validate_fail_invalid_trusted_boot_value(self):
+        properties = {'capabilities': 'trusted_boot:value'}
+        instance_info = {"trusted_boot": "value"}
+        with task_manager.acquire(self.context, self.node.uuid,
+                                  shared=True) as task:
+            task.node.properties = properties
+            task.node.instance_info['capabilities'] = instance_info
+            self.assertRaises(exception.InvalidParameterValue,
+                              task.driver.boot.validate, task)
+
+    @mock.patch.object(base_image_service.BaseImageService, '_show',
+                       autospec=True)
+    def test_validate_fail_no_image_kernel_ramdisk_props(self, mock_glance):
+        mock_glance.return_value = {'properties': {}}
+        with task_manager.acquire(self.context, self.node.uuid,
+                                  shared=True) as task:
+            self.assertRaises(exception.MissingParameterValue,
+                              task.driver.boot.validate,
+                              task)
+
+    @mock.patch.object(base_image_service.BaseImageService, '_show',
+                       autospec=True)
+    def test_validate_fail_glance_image_doesnt_exists(self, mock_glance):
+        mock_glance.side_effect = exception.ImageNotFound('not found')
+        with task_manager.acquire(self.context, self.node.uuid,
+                                  shared=True) as task:
+            self.assertRaises(exception.InvalidParameterValue,
+                              task.driver.boot.validate, task)
+
+    @mock.patch.object(base_image_service.BaseImageService, '_show',
+                       autospec=True)
+    def test_validate_fail_glance_conn_problem(self, mock_glance):
+        exceptions = (exception.GlanceConnectionFailed('connection fail'),
+                      exception.ImageNotAuthorized('not authorized'),
+                      exception.Invalid('invalid'))
+        mock_glance.side_effect = exceptions
+        for exc in exceptions:
+            with task_manager.acquire(self.context, self.node.uuid,
+                                      shared=True) as task:
+                self.assertRaises(exception.InvalidParameterValue,
+                                  task.driver.boot.validate, 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')
+    @mock.patch.object(pxe, '_get_instance_image_info', autospec=True)
+    @mock.patch.object(pxe, '_get_image_info', autospec=True)
+    @mock.patch.object(pxe, '_cache_ramdisk_kernel', autospec=True)
+    @mock.patch.object(pxe, '_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):
+        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)
+            task.driver.boot.prepare_ramdisk(task, {'foo': 'bar'})
+            mock_deploy_img_info.assert_called_once_with(task.node, mode=mode)
+            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=False)
+            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'})
+                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'})
+                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'})
+            elif mode == 'rescue':
+                    mock_cache_r_k.assert_called_once_with(
+                        task, {'rescue_kernel': 'a', 'rescue_ramdisk': 'r'})
+            if uefi:
+                mock_pxe_config.assert_called_once_with(
+                    task, {'foo': 'bar'}, CONF.pxe.uefi_pxe_config_template,
+                    ipxe_enabled=True)
+            else:
+                mock_pxe_config.assert_called_once_with(
+                    task, {'foo': 'bar'}, CONF.pxe.pxe_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()
+        self.config(group='pxe', ipxe_enabled=False)
+        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')
+        render_mock.assert_called_once_with(
+            CONF.pxe.ipxe_boot_script,
+            {'ipxe_for_mac_uri': 'pxelinux.cfg/'})
+
+    @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()
+        self.config(group='pxe', ipxe_enabled=False)
+        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')
+        render_mock.assert_called_once_with(
+            CONF.pxe.ipxe_boot_script,
+            {'ipxe_for_mac_uri': 'pxelinux.cfg/'})
+
+    @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.config(group='pxe', ipxe_enabled=False)
+        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_enabled=False)
+        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')
+
+    @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_enabled=False)
+        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')
+
+    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, '_clean_up_pxe_env', autospec=True)
+    @mock.patch.object(pxe, '_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)
+            get_image_info_mock.assert_called_once_with(task.node, mode=mode)
+
+    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(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, '_cache_ramdisk_kernel', autospec=True)
+    @mock.patch.object(pxe, '_get_instance_image_info', autospec=True)
+    def test_prepare_instance_netboot(
+            self, get_image_info_mock, cache_mock,
+            dhcp_factory_mock, switch_pxe_config_mock,
+            set_boot_device_mock):
+        provider_mock = mock.MagicMock()
+        dhcp_factory_mock.return_value = provider_mock
+        image_info = {'kernel': ('', '/path/to/kernel'),
+                      'ramdisk': ('', '/path/to/ramdisk')}
+        get_image_info_mock.return_value = image_info
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            dhcp_opts = pxe_utils.dhcp_options_for_instance(task)
+            pxe_config_path = pxe_utils.get_pxe_config_file_path(
+                task.node.uuid)
+            task.node.properties['capabilities'] = 'boot_mode:bios'
+            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
+
+            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)
+            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
+            switch_pxe_config_mock.assert_called_once_with(
+                pxe_config_path, "30212642-09d3-467f-8e09-21685826ab50",
+                'bios', False, False, False, False, ipxe_enabled=True)
+            set_boot_device_mock.assert_called_once_with(task,
+                                                         boot_devices.PXE,
+                                                         persistent=True)
+
+    @mock.patch('os.path.isfile', return_value=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, '_cache_ramdisk_kernel', autospec=True)
+    @mock.patch.object(pxe, '_get_instance_image_info', autospec=True)
+    def test_prepare_instance_netboot_active(
+            self, get_image_info_mock, cache_mock,
+            dhcp_factory_mock, switch_pxe_config_mock,
+            set_boot_device_mock, create_pxe_config_mock, isfile_mock):
+        provider_mock = mock.MagicMock()
+        dhcp_factory_mock.return_value = provider_mock
+        image_info = {'kernel': ('', '/path/to/kernel'),
+                      'ramdisk': ('', '/path/to/ramdisk')}
+        get_image_info_mock.return_value = image_info
+        self.node.provision_state = states.ACTIVE
+        self.node.save()
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            dhcp_opts = pxe_utils.dhcp_options_for_instance(task)
+            pxe_config_path = pxe_utils.get_pxe_config_file_path(
+                task.node.uuid)
+            task.node.properties['capabilities'] = 'boot_mode:bios'
+            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
+
+            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)
+            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
+            create_pxe_config_mock.assert_called_once_with(
+                task, mock.ANY, CONF.pxe.pxe_config_template,
+                ipxe_enabled=True)
+            switch_pxe_config_mock.assert_called_once_with(
+                pxe_config_path, "30212642-09d3-467f-8e09-21685826ab50",
+                'bios', False, False, False, False, ipxe_enabled=True)
+            self.assertFalse(set_boot_device_mock.called)
+
+    @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')
+    @mock.patch.object(pxe, '_cache_ramdisk_kernel', autospec=True)
+    @mock.patch.object(pxe, '_get_instance_image_info', autospec=True)
+    def test_prepare_instance_netboot_missing_root_uuid(
+            self, get_image_info_mock, cache_mock,
+            dhcp_factory_mock, switch_pxe_config_mock,
+            set_boot_device_mock):
+        provider_mock = mock.MagicMock()
+        dhcp_factory_mock.return_value = provider_mock
+        image_info = {'kernel': ('', '/path/to/kernel'),
+                      'ramdisk': ('', '/path/to/ramdisk')}
+        get_image_info_mock.return_value = image_info
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            dhcp_opts = pxe_utils.dhcp_options_for_instance(task)
+            task.node.properties['capabilities'] = 'boot_mode:bios'
+            task.node.driver_internal_info['is_whole_disk_image'] = False
+
+            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)
+            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
+            self.assertFalse(switch_pxe_config_mock.called)
+            self.assertFalse(set_boot_device_mock.called)
+
+    # NOTE(TheJulia): The log mock below is attached to the iPXE interface
+    # which directly logs the warning that is being checked for.
+    @mock.patch.object(ipxe.LOG, 'warning', autospec=True)
+    @mock.patch.object(pxe_utils, 'clean_up_pxe_config', autospec=True)
+    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
+    @mock.patch.object(dhcp_factory, 'DHCPFactory')
+    @mock.patch.object(pxe, '_cache_ramdisk_kernel', autospec=True)
+    @mock.patch.object(pxe, '_get_instance_image_info', autospec=True)
+    def test_prepare_instance_whole_disk_image_missing_root_uuid(
+            self, get_image_info_mock, cache_mock,
+            dhcp_factory_mock, set_boot_device_mock,
+            clean_up_pxe_mock, log_mock):
+        provider_mock = mock.MagicMock()
+        dhcp_factory_mock.return_value = provider_mock
+        get_image_info_mock.return_value = {}
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            dhcp_opts = pxe_utils.dhcp_options_for_instance(task)
+            task.node.properties['capabilities'] = 'boot_mode:bios'
+            task.node.driver_internal_info['is_whole_disk_image'] = 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, {})
+            provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
+            self.assertTrue(log_mock.called)
+            clean_up_pxe_mock.assert_called_once_with(task)
+            set_boot_device_mock.assert_called_once_with(
+                task, boot_devices.DISK, 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, '_cache_ramdisk_kernel', autospec=True)
+    @mock.patch.object(pxe, '_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(ipxe_enabled=False, group='pxe')
+        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)
+            pxe_config_path = pxe_utils.get_pxe_config_file_path(
+                task.node.uuid)
+            task.node.properties['capabilities'] = 'boot_mode:bios'
+            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.pxe_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)
+            set_boot_device_mock.assert_called_once_with(task,
+                                                         boot_devices.PXE,
+                                                         persistent=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_localboot(self, clean_up_pxe_config_mock,
+                                        set_boot_device_mock):
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            instance_info = task.node.instance_info
+            instance_info['capabilities'] = {'boot_option': 'local'}
+            task.node.instance_info = instance_info
+            task.node.save()
+            task.driver.boot.prepare_instance(task)
+            clean_up_pxe_config_mock.assert_called_once_with(task)
+            set_boot_device_mock.assert_called_once_with(task,
+                                                         boot_devices.DISK,
+                                                         persistent=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_is_force_persistent_boot_device_enabled(
+            self, clean_up_pxe_config_mock, set_boot_device_mock):
+        with task_manager.acquire(self.context, self.node.uuid) as task:
+            instance_info = task.node.instance_info
+            instance_info['capabilities'] = {'boot_option': 'local'}
+            task.node.instance_info = instance_info
+            task.node.save()
+            task.driver.boot.prepare_instance(task)
+            clean_up_pxe_config_mock.assert_called_once_with(task)
+            driver_info = task.node.driver_info
+            driver_info['force_persistent _boot_device'] = True
+            task.node.driver_info = driver_info
+            set_boot_device_mock.assert_called_once_with(task,
+                                                         boot_devices.DISK,
+                                                         persistent=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_localboot_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:
+            instance_info = task.node.instance_info
+            instance_info['capabilities'] = {'boot_option': 'local'}
+            task.node.instance_info = instance_info
+            task.node.save()
+            task.driver.boot.prepare_instance(task)
+            clean_up_pxe_config_mock.assert_called_once_with(task)
+            self.assertFalse(set_boot_device_mock.called)
+
+    @mock.patch.object(pxe, '_clean_up_pxe_env', autospec=True)
+    @mock.patch.object(pxe, '_get_instance_image_info', autospec=True)
+    def test_clean_up_instance(self, get_image_info_mock,
+                               clean_up_pxe_env_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)
+            get_image_info_mock.assert_called_once_with(
+                task, ipxe_enabled=True)
+
+
+@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)