Fix ``agent`` deploy interface to call ``boot.prepare_instance``

``agent`` deploy interface do not call ``boot.prepare_instance``
if image being provisioned is whole disk image. This commit fixes
that issue.
It also updates ``validate`` method of neutron network interface
module to validate if it can support boot options requested for
instance image.

Change-Id: Ibd49d65f4512f2fa417794b66f4007d82f02e2ac
Story: 1713916
Task: 9259
Story: 1750958
Task: 9288
(cherry picked from commit 4fa1075b95e2e5262f98395275590dcd3833ab74)

7 files changed, 487 insertions, 108 deletions

diff --git a/ironic/drivers/modules/agent.py b/ironic/drivers/modules/agent.py
index 12b72f381..b10219de3 100644
--- a/ironic/drivers/modules/agent.py
+++ b/ironic/drivers/modules/agent.py
@@ -15,6 +15,7 @@
 from ironic_lib import metrics_utils
 from ironic_lib import utils as il_utils
 from oslo_log import log
+from oslo_utils import excutils
 from oslo_utils import units
 import six.moves.urllib_parse as urlparse
 
@@ -280,35 +281,63 @@ class AgentDeployMixin(agent_base_vendor.AgentDeployMixin):
             LOG.error(msg)
             deploy_utils.set_failed_state(task, msg)
             return
+
+        # If `boot_option` is set to `netboot`, PXEBoot.prepare_instance()
+        # would need root_uuid of the whole disk image to add it into the
+        # pxe config to perform chain boot.
+        # IPA would have returned us the 'root_uuid_or_disk_id' if image
+        # being provisioned is a whole disk image. IPA would also provide us
+        # 'efi_system_partition_uuid' if the image being provisioned is a
+        # partition image.
+        # In case of local boot using partition image, we need both
+        # 'root_uuid_or_disk_id' and 'efi_system_partition_uuid' to configure
+        # bootloader for local boot.
+        driver_internal_info = task.node.driver_internal_info
+        root_uuid = self._get_uuid_from_result(task, 'root_uuid')
+        if root_uuid:
+            driver_internal_info['root_uuid_or_disk_id'] = root_uuid
+            task.node.driver_internal_info = driver_internal_info
+            task.node.save()
+        elif iwdi and CONF.agent.manage_agent_boot:
+            # IPA version less than 3.1.0 will not return root_uuid for
+            # whole disk image. Also IPA version introduced a requirement
+            # for hexdump utility that may not be always available. Need to
+            # fall back to older behavior for the same.
+            LOG.warning("With the deploy ramdisk based on Ironic Python Agent "
+                        "version 3.1.0 and beyond, the drivers using "
+                        "`direct` deploy interface performs `netboot` or "
+                        "`local` boot for whole disk image based on value "
+                        "of boot option setting. When you upgrade Ironic "
+                        "Python Agent in your deploy ramdisk, ensure that "
+                        "boot option is set appropriately for the node %s. "
+                        "The boot option can be set using configuration "
+                        "`[deploy]/default_boot_option` or as a `boot_option` "
+                        "capability in node's `properties['capabilities']`. "
+                        "Also please note that this functionality requires "
+                        "`hexdump` command in the ramdisk.", node.uuid)
+
+        efi_sys_uuid = None
         if not iwdi:
-            root_uuid = self._get_uuid_from_result(task, 'root_uuid')
             if deploy_utils.get_boot_mode_for_deploy(node) == 'uefi':
                 efi_sys_uuid = (
                     self._get_uuid_from_result(task,
                                                'efi_system_partition_uuid'))
-            else:
-                efi_sys_uuid = None
-            driver_internal_info = task.node.driver_internal_info
-            driver_internal_info['root_uuid_or_disk_id'] = root_uuid
-            task.node.driver_internal_info = driver_internal_info
-            task.node.save()
-            self.prepare_instance_to_boot(task, root_uuid, efi_sys_uuid)
         LOG.info('Image successfully written to node %s', node.uuid)
-        LOG.debug('Rebooting node %s to instance', node.uuid)
-        if iwdi:
+
+        if CONF.agent.manage_agent_boot:
+            # It is necessary to invoke prepare_instance() of the node's
+            # boot interface, so that the any necessary configurations like
+            # setting of the boot mode (e.g. UEFI secure boot) which cannot
+            # be done on node during deploy stage can be performed.
+            LOG.debug('Executing driver specific tasks before booting up the '
+                      'instance for node %s', node.uuid)
+            self.prepare_instance_to_boot(task, root_uuid, efi_sys_uuid)
+        else:
             manager_utils.node_set_boot_device(task, 'disk', persistent=True)
 
+        LOG.debug('Rebooting node %s to instance', node.uuid)
         self.reboot_and_finish_deploy(task)
 
-        # NOTE(TheJulia): If we deployed a whole disk image, we
-        # should expect a whole disk image and clean-up the tftp files
-        # on-disk incase the node is disregarding the boot preference.
-        # TODO(rameshg87): Not all in-tree drivers using reboot_to_instance
-        # have a boot interface. So include a check for now. Remove this
-        # check once all in-tree drivers have a boot interface.
-        if task.driver.boot and iwdi:
-            task.driver.boot.clean_up_ramdisk(task)
-
 
 class AgentDeploy(AgentDeployMixin, base.DeployInterface):
     """Interface for deploy-related actions."""
@@ -448,11 +477,36 @@ class AgentDeploy(AgentDeployMixin, base.DeployInterface):
             wrong occurred during the power action.
         :raises: exception.ImageRefValidationFailed if image_source is not
             Glance href and is not HTTP(S) URL.
+        :raises: exception.InvalidParameterValue if network validation fails.
         :raises: any boot interface's prepare_ramdisk exceptions.
         """
         node = task.node
         deploy_utils.populate_storage_driver_internal_info(task)
         if node.provision_state == states.DEPLOYING:
+            # Validate network interface to ensure that it supports boot
+            # options configured on the node.
+            try:
+                task.driver.network.validate(task)
+            except exception.InvalidParameterValue:
+                # For 'neutron' network interface validation will fail
+                # if node is using 'netboot' boot option while provisioning
+                # a whole disk image. Updating 'boot_option' in node's
+                # 'instance_info' to 'local for backward compatibility.
+                # TODO(stendulker): Fail here once the default boot
+                # option is local.
+                with excutils.save_and_reraise_exception(reraise=False) as ctx:
+                    instance_info = node.instance_info
+                    capabilities = instance_info.get('capabilities', {})
+                    if 'boot_option' not in capabilities:
+                        capabilities['boot_option'] = 'local'
+                        instance_info['capabilities'] = capabilities
+                        node.instance_info = instance_info
+                        node.save()
+                        # Re-validate the network interface
+                        task.driver.network.validate(task)
+                    else:
+                        ctx.reraise = True
+
             # Adding the node to provisioning network so that the dhcp
             # options get added for the provisioning port.
             manager_utils.node_power_action(task, states.POWER_OFF)
diff --git a/ironic/drivers/modules/network/neutron.py b/ironic/drivers/modules/network/neutron.py
index 7bef57ce0..5bba613db 100644
--- a/ironic/drivers/modules/network/neutron.py
+++ b/ironic/drivers/modules/network/neutron.py
@@ -20,7 +20,9 @@ from oslo_log import log
 from ironic.common import exception
 from ironic.common.i18n import _
 from ironic.common import neutron
+from ironic.common import states
 from ironic.drivers import base
+from ironic.drivers.modules import deploy_utils
 from ironic.drivers.modules.network import common
 
 LOG = log.getLogger(__name__)
@@ -59,6 +61,15 @@ class NeutronNetwork(common.NeutronVIFPortIDMixin,
         """
         self.get_cleaning_network_uuid(task)
         self.get_provisioning_network_uuid(task)
+        node = task.node
+        if (node.provision_state == states.DEPLOYING and
+            node.driver_internal_info.get('is_whole_disk_image') and
+            deploy_utils.get_boot_option(node) == 'netboot'):
+            error_msg = (_('The node %s cannot perform "local" boot for '
+                           'whole disk image when node is using "neutron" '
+                           'network and is configured with "netboot" boot '
+                           'option.') % node.uuid)
+            raise exception.InvalidParameterValue(error_msg)
 
     def add_provisioning_network(self, task):
         """Add the provisioning network to a node.
diff --git a/ironic/drivers/modules/pxe.py b/ironic/drivers/modules/pxe.py
index 486623779..e79310b98 100644
--- a/ironic/drivers/modules/pxe.py
+++ b/ironic/drivers/modules/pxe.py
@@ -630,7 +630,10 @@ class PXEBoot(base.BootInterface):
                     LOG.warning("The disk id for the whole disk image can't "
                                 "be found, unable to switch the pxe config "
                                 "from deployment mode to service (boot) mode "
-                                "for node %(node)s", {"node": task.node.uuid})
+                                "for node %(node)s. Booting the instance "
+                                "from disk.", {"node": task.node.uuid})
+                    pxe_utils.clean_up_pxe_config(task)
+                    boot_device = boot_devices.DISK
             else:
                 _build_service_pxe_config(task, instance_image_info,
                                           root_uuid_or_disk_id)
diff --git a/ironic/tests/unit/drivers/modules/network/test_neutron.py b/ironic/tests/unit/drivers/modules/network/test_neutron.py
index 7bff4892d..0b7381471 100644
--- a/ironic/tests/unit/drivers/modules/network/test_neutron.py
+++ b/ironic/tests/unit/drivers/modules/network/test_neutron.py
@@ -19,6 +19,7 @@ from oslo_utils import uuidutils
 
 from ironic.common import exception
 from ironic.common import neutron as neutron_common
+from ironic.common import states
 from ironic.conductor import task_manager
 from ironic.drivers import base as drivers_base
 from ironic.drivers.modules.network import neutron
@@ -104,6 +105,65 @@ class NeutronInterfaceTestCase(db_base.DbTestCase):
                                         context=task.context)],
                              validate_mock.call_args_list)
 
+    @mock.patch.object(neutron_common, 'validate_network', autospec=True)
+    def test_validate_boot_option_netboot(self, validate_mock):
+        driver_internal_info = self.node.driver_internal_info
+        driver_internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = driver_internal_info
+        boot_option = {'capabilities': '{"boot_option": "netboot"}'}
+        self.node.instance_info = boot_option
+        self.node.provision_state = states.DEPLOYING
+        self.node.save()
+        with task_manager.acquire(self.context, self.node.id) as task:
+            self.assertRaisesRegex(
+                exception.InvalidParameterValue,
+                'cannot perform "local" boot for whole disk image',
+                self.interface.validate, task)
+            self.assertEqual([mock.call(CONF.neutron.cleaning_network,
+                                        'cleaning network',
+                                        context=task.context),
+                              mock.call(CONF.neutron.provisioning_network,
+                                        'provisioning network',
+                                        context=task.context)],
+                             validate_mock.call_args_list)
+
+    @mock.patch.object(neutron_common, 'validate_network', autospec=True)
+    def test_validate_boot_option_netboot_no_exc(self, validate_mock):
+        CONF.set_override('default_boot_option', 'netboot', 'deploy')
+        driver_internal_info = self.node.driver_internal_info
+        driver_internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = driver_internal_info
+        self.node.provision_state = states.AVAILABLE
+        self.node.save()
+        with task_manager.acquire(self.context, self.node.id) as task:
+            self.interface.validate(task)
+            self.assertEqual([mock.call(CONF.neutron.cleaning_network,
+                                        'cleaning network',
+                                        context=task.context),
+                              mock.call(CONF.neutron.provisioning_network,
+                                        'provisioning network',
+                                        context=task.context)],
+                             validate_mock.call_args_list)
+
+    @mock.patch.object(neutron_common, 'validate_network', autospec=True)
+    def test_validate_boot_option_local(self, validate_mock):
+        driver_internal_info = self.node.driver_internal_info
+        driver_internal_info['is_whole_disk_image'] = True
+        self.node.driver_internal_info = driver_internal_info
+        boot_option = {'capabilities': '{"boot_option": "local"}'}
+        self.node.instance_info = boot_option
+        self.node.provision_state = states.DEPLOYING
+        self.node.save()
+        with task_manager.acquire(self.context, self.node.id) as task:
+            self.interface.validate(task)
+            self.assertEqual([mock.call(CONF.neutron.cleaning_network,
+                                        'cleaning network',
+                                        context=task.context),
+                              mock.call(CONF.neutron.provisioning_network,
+                                        'provisioning network',
+                                        context=task.context)],
+                             validate_mock.call_args_list)
+
     @mock.patch.object(neutron_common, 'validate_network',
                        side_effect=lambda n, t, context=None: n)
     @mock.patch.object(neutron_common, 'rollback_ports')
diff --git a/ironic/tests/unit/drivers/modules/test_agent.py b/ironic/tests/unit/drivers/modules/test_agent.py
index 3303e0b34..72e7356e7 100644
--- a/ironic/tests/unit/drivers/modules/test_agent.py
+++ b/ironic/tests/unit/drivers/modules/test_agent.py
@@ -19,6 +19,7 @@ from oslo_config import cfg
 
 from ironic.common import dhcp_factory
 from ironic.common import exception
+from ironic.common import image_service
 from ironic.common import images
 from ironic.common import raid
 from ironic.common import states
@@ -31,6 +32,7 @@ from ironic.drivers.modules import agent_client
 from ironic.drivers.modules import deploy_utils
 from ironic.drivers.modules import fake
 from ironic.drivers.modules.network import flat as flat_network
+from ironic.drivers.modules.network import neutron as neutron_network
 from ironic.drivers.modules import pxe
 from ironic.drivers.modules.storage import noop as noop_storage
 from ironic.drivers import utils as driver_utils
@@ -342,8 +344,11 @@ class TestAgentDeploy(db_base.DbTestCase):
     @mock.patch.object(flat_network.FlatNetwork,
                        'unconfigure_tenant_networks',
                        spec_set=True, autospec=True)
+    @mock.patch.object(flat_network.FlatNetwork, 'validate',
+                       spec_set=True, autospec=True)
     def test_prepare(
-            self, unconfigure_tenant_net_mock, add_provisioning_net_mock,
+            self, validate_net_mock,
+            unconfigure_tenant_net_mock, add_provisioning_net_mock,
             build_instance_info_mock, build_options_mock,
             pxe_prepare_ramdisk_mock, storage_driver_info_mock,
             storage_attach_volumes_mock):
@@ -352,30 +357,215 @@ class TestAgentDeploy(db_base.DbTestCase):
             task.node.provision_state = states.DEPLOYING
             build_instance_info_mock.return_value = {'foo': 'bar'}
             build_options_mock.return_value = {'a': 'b'}
-
             self.driver.prepare(task)
+            storage_driver_info_mock.assert_called_once_with(task)
+            validate_net_mock.assert_called_once_with(mock.ANY, task)
+            add_provisioning_net_mock.assert_called_once_with(mock.ANY, task)
+            unconfigure_tenant_net_mock.assert_called_once_with(mock.ANY, task)
+            storage_attach_volumes_mock.assert_called_once_with(
+                task.driver.storage, task)
+            build_instance_info_mock.assert_called_once_with(task)
+            build_options_mock.assert_called_once_with(task.node)
+            pxe_prepare_ramdisk_mock.assert_called_once_with(
+                task, {'a': 'b'})
+        self.node.refresh()
+        self.assertEqual('bar', self.node.instance_info['foo'])
 
+    @mock.patch.object(noop_storage.NoopStorage, 'attach_volumes',
+                       autospec=True)
+    @mock.patch.object(deploy_utils, 'populate_storage_driver_internal_info')
+    @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk')
+    @mock.patch.object(deploy_utils, 'build_agent_options')
+    @mock.patch.object(deploy_utils, 'build_instance_info_for_deploy')
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'add_provisioning_network',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'unconfigure_tenant_networks',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork, 'validate',
+                       spec_set=True, autospec=True)
+    def test_prepare_with_neutron_net(
+            self, validate_net_mock,
+            unconfigure_tenant_net_mock, add_provisioning_net_mock,
+            build_instance_info_mock, build_options_mock,
+            pxe_prepare_ramdisk_mock, storage_driver_info_mock,
+            storage_attach_volumes_mock):
+        node = self.node
+        node.network_interface = 'neutron'
+        node.save()
+        with task_manager.acquire(
+                self.context, self.node['uuid'], shared=False) as task:
+            task.node.provision_state = states.DEPLOYING
+            build_instance_info_mock.return_value = {'foo': 'bar'}
+            build_options_mock.return_value = {'a': 'b'}
+            self.driver.prepare(task)
+            storage_driver_info_mock.assert_called_once_with(task)
+            validate_net_mock.assert_called_once_with(mock.ANY, task)
+            add_provisioning_net_mock.assert_called_once_with(mock.ANY, task)
+            unconfigure_tenant_net_mock.assert_called_once_with(mock.ANY, task)
+            storage_attach_volumes_mock.assert_called_once_with(
+                task.driver.storage, task)
             build_instance_info_mock.assert_called_once_with(task)
             build_options_mock.assert_called_once_with(task.node)
             pxe_prepare_ramdisk_mock.assert_called_once_with(
                 task, {'a': 'b'})
+        self.node.refresh()
+        self.assertEqual('bar', self.node.instance_info['foo'])
+
+    @mock.patch.object(noop_storage.NoopStorage, 'attach_volumes',
+                       autospec=True)
+    @mock.patch.object(deploy_utils, 'populate_storage_driver_internal_info')
+    @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk')
+    @mock.patch.object(deploy_utils, 'build_agent_options')
+    @mock.patch.object(image_service.HttpImageService, 'validate_href',
+                       autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'add_provisioning_network',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'unconfigure_tenant_networks',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork, 'validate',
+                       spec_set=True, autospec=True)
+    def test_prepare_with_neutron_net_exc_no_capabilities(
+            self, validate_net_mock,
+            unconfigure_tenant_net_mock, add_provisioning_net_mock,
+            validate_href_mock, build_options_mock,
+            pxe_prepare_ramdisk_mock, storage_driver_info_mock,
+            storage_attach_volumes_mock):
+        node = self.node
+        node.network_interface = 'neutron'
+        node.save()
+        validate_net_mock.side_effect = [
+            exception.InvalidParameterValue('invalid'), None]
+        with task_manager.acquire(
+                self.context, self.node['uuid'], shared=False) as task:
+            task.node.provision_state = states.DEPLOYING
+            build_options_mock.return_value = {'a': 'b'}
+            self.driver.prepare(task)
+            storage_driver_info_mock.assert_called_once_with(task)
+            self.assertEqual(2, validate_net_mock.call_count)
             add_provisioning_net_mock.assert_called_once_with(mock.ANY, task)
             unconfigure_tenant_net_mock.assert_called_once_with(mock.ANY, task)
+            storage_attach_volumes_mock.assert_called_once_with(
+                task.driver.storage, task)
+            validate_href_mock.assert_called_once_with(mock.ANY, 'fake-image',
+                                                       secret=False)
+            build_options_mock.assert_called_once_with(task.node)
+            pxe_prepare_ramdisk_mock.assert_called_once_with(
+                task, {'a': 'b'})
+        self.node.refresh()
+        capabilities = self.node.instance_info['capabilities']
+        self.assertEqual('local', capabilities['boot_option'])
+
+    @mock.patch.object(noop_storage.NoopStorage, 'attach_volumes',
+                       autospec=True)
+    @mock.patch.object(deploy_utils, 'populate_storage_driver_internal_info')
+    @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk')
+    @mock.patch.object(deploy_utils, 'build_agent_options')
+    @mock.patch.object(image_service.HttpImageService, 'validate_href',
+                       autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'add_provisioning_network',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'unconfigure_tenant_networks',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork, 'validate',
+                       spec_set=True, autospec=True)
+    def test_prepare_with_neutron_net_exc_no_capabilities_overwrite(
+            self, validate_net_mock,
+            unconfigure_tenant_net_mock, add_provisioning_net_mock,
+            validate_href_mock, build_options_mock,
+            pxe_prepare_ramdisk_mock, storage_driver_info_mock,
+            storage_attach_volumes_mock):
+        node = self.node
+        node.network_interface = 'neutron'
+        instance_info = node.instance_info
+        instance_info['capabilities'] = {"cat": "meow"}
+        node.instance_info = instance_info
+        node.save()
+        validate_net_mock.side_effect = [
+            exception.InvalidParameterValue('invalid'), None]
+        with task_manager.acquire(
+                self.context, self.node['uuid'], shared=False) as task:
+            task.node.provision_state = states.DEPLOYING
+            build_options_mock.return_value = {'a': 'b'}
+            self.driver.prepare(task)
             storage_driver_info_mock.assert_called_once_with(task)
+            self.assertEqual(2, validate_net_mock.call_count)
+            add_provisioning_net_mock.assert_called_once_with(mock.ANY, task)
+            unconfigure_tenant_net_mock.assert_called_once_with(mock.ANY, task)
             storage_attach_volumes_mock.assert_called_once_with(
                 task.driver.storage, task)
+            validate_href_mock.assert_called_once_with(mock.ANY, 'fake-image',
+                                                       secret=False)
+            build_options_mock.assert_called_once_with(task.node)
+            pxe_prepare_ramdisk_mock.assert_called_once_with(
+                task, {'a': 'b'})
+        self.node.refresh()
+        capabilities = self.node.instance_info['capabilities']
+        self.assertEqual('local', capabilities['boot_option'])
+        self.assertEqual('meow', capabilities['cat'])
 
+    @mock.patch.object(noop_storage.NoopStorage, 'attach_volumes',
+                       autospec=True)
+    @mock.patch.object(deploy_utils, 'populate_storage_driver_internal_info')
+    @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk')
+    @mock.patch.object(deploy_utils, 'build_agent_options')
+    @mock.patch.object(deploy_utils, 'build_instance_info_for_deploy')
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'add_provisioning_network',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork,
+                       'unconfigure_tenant_networks',
+                       spec_set=True, autospec=True)
+    @mock.patch.object(neutron_network.NeutronNetwork, 'validate',
+                       spec_set=True, autospec=True)
+    def test_prepare_with_neutron_net_exc_reraise(
+            self, validate_net_mock,
+            unconfigure_tenant_net_mock, add_provisioning_net_mock,
+            build_instance_info_mock, build_options_mock,
+            pxe_prepare_ramdisk_mock, storage_driver_info_mock,
+            storage_attach_volumes_mock):
+        node = self.node
+        node.network_interface = 'neutron'
+        instance_info = node.instance_info
+        instance_info['capabilities'] = {"boot_option": "netboot"}
+        node.instance_info = instance_info
+        node.save()
+        validate_net_mock.side_effect = (
+            exception.InvalidParameterValue('invalid'))
+        with task_manager.acquire(
+                self.context, self.node['uuid'], shared=False) as task:
+            task.node.provision_state = states.DEPLOYING
+            self.assertRaises(exception.InvalidParameterValue,
+                              task.driver.deploy.prepare,
+                              task)
+            storage_driver_info_mock.assert_called_once_with(task)
+            validate_net_mock.assert_called_once_with(mock.ANY, task)
+            self.assertFalse(add_provisioning_net_mock.called)
+            self.assertFalse(unconfigure_tenant_net_mock.called)
+            self.assertFalse(storage_attach_volumes_mock.called)
+            self.assertFalse(build_instance_info_mock.called)
+            self.assertFalse(build_options_mock.called)
+            self.assertFalse(pxe_prepare_ramdisk_mock.called)
         self.node.refresh()
-        self.assertEqual('bar', self.node.instance_info['foo'])
+        capabilities = self.node.instance_info['capabilities']
+        self.assertEqual('netboot', capabilities['boot_option'])
 
     @mock.patch.object(flat_network.FlatNetwork, 'add_provisioning_network',
                        spec_set=True, autospec=True)
+    @mock.patch.object(flat_network.FlatNetwork, 'validate',
+                       spec_set=True, autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk')
     @mock.patch.object(deploy_utils, 'build_agent_options')
     @mock.patch.object(deploy_utils, 'build_instance_info_for_deploy')
     def test_prepare_manage_agent_boot_false(
-            self, build_instance_info_mock, build_options_mock,
-            pxe_prepare_ramdisk_mock, add_provisioning_net_mock):
+            self, build_instance_info_mock,
+            build_options_mock, pxe_prepare_ramdisk_mock,
+            validate_net_mock, add_provisioning_net_mock):
         self.config(group='agent', manage_agent_boot=False)
         with task_manager.acquire(
                 self.context, self.node['uuid'], shared=False) as task:
@@ -384,6 +574,7 @@ class TestAgentDeploy(db_base.DbTestCase):
 
             self.driver.prepare(task)
 
+            validate_net_mock.assert_called_once_with(mock.ANY, task)
             build_instance_info_mock.assert_called_once_with(task)
             add_provisioning_net_mock.assert_called_once_with(mock.ANY, task)
             self.assertFalse(build_options_mock.called)
@@ -461,6 +652,8 @@ class TestAgentDeploy(db_base.DbTestCase):
     @mock.patch.object(flat_network.FlatNetwork,
                        'unconfigure_tenant_networks',
                        spec_set=True, autospec=True)
+    @mock.patch.object(flat_network.FlatNetwork, 'validate',
+                       spec_set=True, autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk', autospec=True)
     @mock.patch.object(deploy_utils, 'build_agent_options', autospec=True)
@@ -469,16 +662,16 @@ class TestAgentDeploy(db_base.DbTestCase):
     def test_prepare_storage_write_false(
             self, build_instance_info_mock, build_options_mock,
             pxe_prepare_ramdisk_mock, pxe_prepare_instance_mock,
-            remove_tenant_net_mock, add_provisioning_net_mock,
-            storage_driver_info_mock, storage_attach_volumes_mock,
-            should_write_image_mock):
+            validate_net_mock, remove_tenant_net_mock,
+            add_provisioning_net_mock, storage_driver_info_mock,
+            storage_attach_volumes_mock, should_write_image_mock):
         should_write_image_mock.return_value = False
         with task_manager.acquire(
                 self.context, self.node['uuid'], shared=False) as task:
             task.node.provision_state = states.DEPLOYING
 
             self.driver.prepare(task)
-
+            validate_net_mock.assert_called_once_with(mock.ANY, task)
             self.assertFalse(build_instance_info_mock.called)
             self.assertFalse(build_options_mock.called)
             self.assertFalse(pxe_prepare_ramdisk_mock.called)
@@ -509,6 +702,8 @@ class TestAgentDeploy(db_base.DbTestCase):
 
     @mock.patch.object(flat_network.FlatNetwork, 'add_provisioning_network',
                        spec_set=True, autospec=True)
+    @mock.patch.object(flat_network.FlatNetwork, 'validate',
+                       spec_set=True, autospec=True)
     @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk')
     @mock.patch.object(deploy_utils, 'build_agent_options')
     @mock.patch.object(deploy_utils, 'build_instance_info_for_deploy')
@@ -518,6 +713,7 @@ class TestAgentDeploy(db_base.DbTestCase):
                                       build_instance_info_mock,
                                       build_options_mock,
                                       pxe_prepare_ramdisk_mock,
+                                      validate_net_mock,
                                       add_provisioning_net_mock):
         mock_write.return_value = False
         with task_manager.acquire(
@@ -527,6 +723,7 @@ class TestAgentDeploy(db_base.DbTestCase):
             build_options_mock.return_value = {'a': 'b'}
 
             self.driver.prepare(task)
+            validate_net_mock.assert_called_once_with(mock.ANY, task)
             build_instance_info_mock.assert_not_called()
             build_options_mock.assert_not_called()
             pxe_prepare_ramdisk_mock.assert_not_called()
@@ -754,6 +951,7 @@ class TestAgentDeploy(db_base.DbTestCase):
             self.assertEqual(states.ACTIVE,
                              task.node.target_provision_state)
 
+    @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
     @mock.patch.object(agent.AgentDeployMixin, '_get_uuid_from_result',
                        autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
@@ -761,17 +959,17 @@ class TestAgentDeploy(db_base.DbTestCase):
                        spec=types.FunctionType)
     @mock.patch.object(agent_client.AgentClient, 'power_off',
                        spec=types.FunctionType)
-    @mock.patch.object(pxe.PXEBoot, 'prepare_instance',
+    @mock.patch.object(agent.AgentDeployMixin, 'prepare_instance_to_boot',
                        autospec=True)
     @mock.patch('ironic.drivers.modules.agent.AgentDeployMixin'
                 '.check_deploy_success', autospec=True)
-    @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True)
-    def test_reboot_to_instance(self, clean_pxe_mock, check_deploy_mock,
-                                prepare_mock, power_off_mock,
+    def test_reboot_to_instance(self, check_deploy_mock,
+                                prepare_instance_mock, power_off_mock,
                                 get_power_state_mock, node_power_action_mock,
-                                uuid_mock):
+                                uuid_mock, log_mock):
+        self.config(manage_agent_boot=True, group='agent')
         check_deploy_mock.return_value = None
-        uuid_mock.return_value = 'root_uuid'
+        uuid_mock.return_value = None
         self.node.provision_state = states.DEPLOYWAIT
         self.node.target_provision_state = states.ACTIVE
         self.node.save()
@@ -779,23 +977,25 @@ class TestAgentDeploy(db_base.DbTestCase):
                                   shared=False) as task:
             get_power_state_mock.return_value = states.POWER_OFF
             task.node.driver_internal_info['is_whole_disk_image'] = True
-
             task.driver.deploy.reboot_to_instance(task)
-
-            clean_pxe_mock.assert_called_once_with(task.driver.boot, task)
             check_deploy_mock.assert_called_once_with(mock.ANY, task.node)
+            uuid_mock.assert_called_once_with(mock.ANY, task, 'root_uuid')
+            self.assertNotIn('root_uuid_or_disk_id',
+                             task.node.driver_internal_info)
+            self.assertTrue(log_mock.called)
+            self.assertIn("Ironic Python Agent version 3.1.0 and beyond",
+                          log_mock.call_args[0][0])
+            prepare_instance_mock.assert_called_once_with(mock.ANY, task,
+                                                          None, None)
             power_off_mock.assert_called_once_with(task.node)
             get_power_state_mock.assert_called_once_with(task)
             node_power_action_mock.assert_called_once_with(
                 task, states.POWER_ON)
-            self.assertFalse(prepare_mock.called)
             self.assertEqual(states.ACTIVE, task.node.provision_state)
             self.assertEqual(states.NOSTATE, task.node.target_provision_state)
-            self.assertNotIn('root_uuid_or_disk_id',
-                             task.node.driver_internal_info)
-            self.assertFalse(uuid_mock.called)
 
-    @mock.patch.object(deploy_utils, 'get_boot_mode_for_deploy', autospec=True)
+    @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
+    @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
     @mock.patch.object(agent.AgentDeployMixin, '_get_uuid_from_result',
                        autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
@@ -803,48 +1003,44 @@ class TestAgentDeploy(db_base.DbTestCase):
                        spec=types.FunctionType)
     @mock.patch.object(agent_client.AgentClient, 'power_off',
                        spec=types.FunctionType)
-    @mock.patch.object(pxe.PXEBoot, 'prepare_instance',
+    @mock.patch.object(agent.AgentDeployMixin, 'prepare_instance_to_boot',
                        autospec=True)
     @mock.patch('ironic.drivers.modules.agent.AgentDeployMixin'
                 '.check_deploy_success', autospec=True)
-    @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True)
-    def test_reboot_to_instance_partition_image(self, clean_pxe_mock,
-                                                check_deploy_mock,
-                                                prepare_mock, power_off_mock,
-                                                get_power_state_mock,
-                                                node_power_action_mock,
-                                                uuid_mock, boot_mode_mock):
+    def test_reboot_to_instance_no_manage_agent_boot(self, check_deploy_mock,
+                                                     prepare_instance_mock,
+                                                     power_off_mock,
+                                                     get_power_state_mock,
+                                                     node_power_action_mock,
+                                                     uuid_mock, bootdev_mock,
+                                                     log_mock):
+        self.config(manage_agent_boot=False, group='agent')
         check_deploy_mock.return_value = None
-        uuid_mock.return_value = 'root_uuid'
+        uuid_mock.return_value = None
         self.node.provision_state = states.DEPLOYWAIT
         self.node.target_provision_state = states.ACTIVE
         self.node.save()
-        boot_mode_mock.return_value = 'bios'
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             get_power_state_mock.return_value = states.POWER_OFF
-            driver_internal_info = task.node.driver_internal_info
-            driver_internal_info['is_whole_disk_image'] = False
-            task.node.driver_internal_info = driver_internal_info
-
+            task.node.driver_internal_info['is_whole_disk_image'] = True
             task.driver.deploy.reboot_to_instance(task)
-
-            self.assertFalse(clean_pxe_mock.called)
             check_deploy_mock.assert_called_once_with(mock.ANY, task.node)
+            uuid_mock.assert_called_once_with(mock.ANY, task, 'root_uuid')
+            self.assertNotIn('root_uuid_or_disk_id',
+                             task.node.driver_internal_info)
+            self.assertFalse(log_mock.called)
+            self.assertFalse(prepare_instance_mock.called)
+            bootdev_mock.assert_called_once_with(task, 'disk', persistent=True)
             power_off_mock.assert_called_once_with(task.node)
             get_power_state_mock.assert_called_once_with(task)
             node_power_action_mock.assert_called_once_with(
                 task, states.POWER_ON)
-            prepare_mock.assert_called_once_with(task.driver.boot, task)
             self.assertEqual(states.ACTIVE, task.node.provision_state)
             self.assertEqual(states.NOSTATE, task.node.target_provision_state)
-            driver_int_info = task.node.driver_internal_info
-            self.assertEqual('root_uuid',
-                             driver_int_info['root_uuid_or_disk_id']),
-            uuid_mock.assert_called_once_with(task.driver.deploy,
-                                              task, 'root_uuid')
-            boot_mode_mock.assert_called_once_with(task.node)
 
+    @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
+    @mock.patch.object(deploy_utils, 'get_boot_mode_for_deploy', autospec=True)
     @mock.patch.object(agent.AgentDeployMixin, '_get_uuid_from_result',
                        autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
@@ -852,45 +1048,48 @@ class TestAgentDeploy(db_base.DbTestCase):
                        spec=types.FunctionType)
     @mock.patch.object(agent_client.AgentClient, 'power_off',
                        spec=types.FunctionType)
-    @mock.patch.object(pxe.PXEBoot, 'prepare_instance',
+    @mock.patch.object(agent.AgentDeployMixin, 'prepare_instance_to_boot',
                        autospec=True)
     @mock.patch('ironic.drivers.modules.agent.AgentDeployMixin'
                 '.check_deploy_success', autospec=True)
-    @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True)
-    def test_reboot_to_instance_boot_none(self, clean_pxe_mock,
-                                          check_deploy_mock,
-                                          prepare_mock, power_off_mock,
-                                          get_power_state_mock,
-                                          node_power_action_mock,
-                                          uuid_mock):
+    def test_reboot_to_instance_partition_image(self, check_deploy_mock,
+                                                prepare_instance_mock,
+                                                power_off_mock,
+                                                get_power_state_mock,
+                                                node_power_action_mock,
+                                                uuid_mock, boot_mode_mock,
+                                                log_mock):
         check_deploy_mock.return_value = None
+        uuid_mock.return_value = 'root_uuid'
         self.node.provision_state = states.DEPLOYWAIT
         self.node.target_provision_state = states.ACTIVE
         self.node.save()
+        boot_mode_mock.return_value = 'bios'
         with task_manager.acquire(self.context, self.node.uuid,
                                   shared=False) as task:
             get_power_state_mock.return_value = states.POWER_OFF
             driver_internal_info = task.node.driver_internal_info
-            driver_internal_info['is_whole_disk_image'] = True
+            driver_internal_info['is_whole_disk_image'] = False
             task.node.driver_internal_info = driver_internal_info
-            task.driver.boot = None
-
             task.driver.deploy.reboot_to_instance(task)
-
-            self.assertFalse(clean_pxe_mock.called)
-            self.assertFalse(prepare_mock.called)
-            power_off_mock.assert_called_once_with(task.node)
             check_deploy_mock.assert_called_once_with(mock.ANY, task.node)
-            self.assertNotIn('root_uuid_or_disk_id',
-                             task.node.driver_internal_info)
-
+            uuid_mock.assert_called_once_with(mock.ANY,
+                                              task, 'root_uuid')
+            driver_int_info = task.node.driver_internal_info
+            self.assertEqual('root_uuid',
+                             driver_int_info['root_uuid_or_disk_id']),
+            boot_mode_mock.assert_called_once_with(task.node)
+            self.assertFalse(log_mock.called)
+            prepare_instance_mock.assert_called_once_with(mock.ANY, task,
+                                                          'root_uuid', None)
+            power_off_mock.assert_called_once_with(task.node)
             get_power_state_mock.assert_called_once_with(task)
             node_power_action_mock.assert_called_once_with(
                 task, states.POWER_ON)
             self.assertEqual(states.ACTIVE, task.node.provision_state)
             self.assertEqual(states.NOSTATE, task.node.target_provision_state)
-            self.assertFalse(uuid_mock.called)
 
+    @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
     @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True)
     @mock.patch.object(agent.AgentDeployMixin, '_get_uuid_from_result',
                        autospec=True)
@@ -899,15 +1098,14 @@ class TestAgentDeploy(db_base.DbTestCase):
                        spec=types.FunctionType)
     @mock.patch.object(agent_client.AgentClient, 'power_off',
                        spec=types.FunctionType)
-    @mock.patch.object(pxe.PXEBoot, 'prepare_instance',
+    @mock.patch.object(agent.AgentDeployMixin, 'prepare_instance_to_boot',
                        autospec=True)
     @mock.patch('ironic.drivers.modules.agent.AgentDeployMixin'
                 '.check_deploy_success', autospec=True)
-    @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True)
     def test_reboot_to_instance_boot_error(
-            self, clean_pxe_mock, check_deploy_mock, prepare_mock,
+            self, check_deploy_mock, prepare_instance_mock,
             power_off_mock, get_power_state_mock, node_power_action_mock,
-            uuid_mock, collect_ramdisk_logs_mock):
+            uuid_mock, collect_ramdisk_logs_mock, log_mock):
         check_deploy_mock.return_value = "Error"
         uuid_mock.return_value = None
         self.node.provision_state = states.DEPLOYWAIT
@@ -917,20 +1115,17 @@ class TestAgentDeploy(db_base.DbTestCase):
                                   shared=False) as task:
             get_power_state_mock.return_value = states.POWER_OFF
             task.node.driver_internal_info['is_whole_disk_image'] = True
-            task.driver.boot = None
             task.driver.deploy.reboot_to_instance(task)
-
-            self.assertFalse(clean_pxe_mock.called)
-            self.assertFalse(prepare_mock.called)
-            self.assertFalse(power_off_mock.called)
             check_deploy_mock.assert_called_once_with(mock.ANY, task.node)
+            self.assertFalse(prepare_instance_mock.called)
+            self.assertFalse(log_mock.called)
+            self.assertFalse(power_off_mock.called)
+            collect_ramdisk_logs_mock.assert_called_once_with(task.node)
             self.assertEqual(states.DEPLOYFAIL, task.node.provision_state)
             self.assertEqual(states.ACTIVE, task.node.target_provision_state)
-            collect_ramdisk_logs_mock.assert_called_once_with(task.node)
 
-    @mock.patch.object(agent_base_vendor.AgentDeployMixin,
-                       'configure_local_boot', autospec=True)
-    @mock.patch.object(deploy_utils, 'try_set_boot_device', autospec=True)
+    @mock.patch.object(agent.LOG, 'warning', spec_set=True, autospec=True)
+    @mock.patch.object(deploy_utils, 'get_boot_mode_for_deploy', autospec=True)
     @mock.patch.object(agent.AgentDeployMixin, '_get_uuid_from_result',
                        autospec=True)
     @mock.patch.object(manager_utils, 'node_power_action', autospec=True)
@@ -938,19 +1133,17 @@ class TestAgentDeploy(db_base.DbTestCase):
                        spec=types.FunctionType)
     @mock.patch.object(agent_client.AgentClient, 'power_off',
                        spec=types.FunctionType)
-    @mock.patch.object(pxe.PXEBoot, 'prepare_instance',
+    @mock.patch.object(agent.AgentDeployMixin, 'prepare_instance_to_boot',
                        autospec=True)
     @mock.patch('ironic.drivers.modules.agent.AgentDeployMixin'
                 '.check_deploy_success', autospec=True)
-    @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True)
-    def test_reboot_to_instance_localboot(self, clean_pxe_mock,
-                                          check_deploy_mock,
-                                          prepare_mock, power_off_mock,
+    def test_reboot_to_instance_localboot(self, check_deploy_mock,
+                                          prepare_instance_mock,
+                                          power_off_mock,
                                           get_power_state_mock,
                                           node_power_action_mock,
-                                          uuid_mock,
-                                          bootdev_mock,
-                                          configure_mock):
+                                          uuid_mock, boot_mode_mock,
+                                          log_mock):
         check_deploy_mock.return_value = None
         uuid_mock.side_effect = ['root_uuid', 'efi_uuid']
         self.node.provision_state = states.DEPLOYWAIT
@@ -965,11 +1158,21 @@ class TestAgentDeploy(db_base.DbTestCase):
             task.node.driver_internal_info = driver_internal_info
             boot_option = {'capabilities': '{"boot_option": "local"}'}
             task.node.instance_info = boot_option
+            boot_mode_mock.return_value = 'uefi'
             task.driver.deploy.reboot_to_instance(task)
 
-            self.assertFalse(clean_pxe_mock.called)
             check_deploy_mock.assert_called_once_with(mock.ANY, task.node)
-            self.assertFalse(bootdev_mock.called)
+            driver_int_info = task.node.driver_internal_info
+            self.assertEqual('root_uuid',
+                             driver_int_info['root_uuid_or_disk_id']),
+            uuid_mock_calls = [
+                mock.call(mock.ANY, task, 'root_uuid'),
+                mock.call(mock.ANY, task, 'efi_system_partition_uuid')]
+            uuid_mock.assert_has_calls(uuid_mock_calls)
+            boot_mode_mock.assert_called_once_with(task.node)
+            self.assertFalse(log_mock.called)
+            prepare_instance_mock.assert_called_once_with(
+                mock.ANY, task, 'root_uuid', 'efi_uuid')
             power_off_mock.assert_called_once_with(task.node)
             get_power_state_mock.assert_called_once_with(task)
             node_power_action_mock.assert_called_once_with(
diff --git a/ironic/tests/unit/drivers/modules/test_pxe.py b/ironic/tests/unit/drivers/modules/test_pxe.py
index 62826883f..c699a7131 100644
--- a/ironic/tests/unit/drivers/modules/test_pxe.py
+++ b/ironic/tests/unit/drivers/modules/test_pxe.py
@@ -1201,6 +1201,34 @@ class PXEBootTestCase(db_base.DbTestCase):
             self.assertFalse(switch_pxe_config_mock.called)
             self.assertFalse(set_boot_device_mock.called)
 
+    @mock.patch.object(pxe.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.node, task.context)
+            cache_mock.assert_called_once_with(
+                task.context, task.node, {})
+            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)
diff --git a/releasenotes/notes/fix-prepare-instance-for-agent-interface-56753bdf04dd581f.yaml b/releasenotes/notes/fix-prepare-instance-for-agent-interface-56753bdf04dd581f.yaml
new file mode 100644
index 000000000..44f22f245
--- /dev/null
+++ b/releasenotes/notes/fix-prepare-instance-for-agent-interface-56753bdf04dd581f.yaml
@@ -0,0 +1,20 @@
+---
+fixes:
+  - |
+    Fixes ``direct`` deploy interface to invoke ``boot.prepare_instance``
+    irrespective of image type being provisioned. It was calling
+    ``boot.prepare_instance`` only if the image being provisioned is a
+    partition image. See bugs `1713916
+    <https://storyboard.openstack.org/#!/story/1713916>`_ and `1750958
+    <https://storyboard.openstack.org/#!/story/1750958>`_ for details.
+upgrade:
+  - |
+    With the deploy ramdisk based on Ironic Python Agent version 3.1.0
+    and beyond, the drivers using ``direct`` deploy interface performs
+    ``netboot`` or ``local`` boot for whole disk image based on value
+    of boot option setting. When you upgrade Ironic Python Agent in your
+    deploy ramdisk, ensure that boot option is set appropriately for the
+    node. The boot option can be set using configuration
+    ``[deploy]/default_boot_option`` or as a ``boot_option`` capability
+    in node's ``properties['capabilities']``. Also please note that this
+    functionality requires ``hexdump`` command in the ramdisk.