Move ramdisk deploy to its own module

It does not depend on PXE and is actually often used with virtual media.

Change-Id: Ida6edf819dbb3d1a51c465b4e109eafd977fd66c
This commit is contained in:
Dmitry Tantsur 2021-09-06 16:30:52 +02:00
parent 75304deefb
commit 88d6b99cc7
6 changed files with 404 additions and 334 deletions

View File

@ -29,6 +29,7 @@ from ironic.drivers.modules.network import noop as noop_net
from ironic.drivers.modules import noop
from ironic.drivers.modules import noop_mgmt
from ironic.drivers.modules import pxe
from ironic.drivers.modules import ramdisk
from ironic.drivers.modules.storage import cinder
from ironic.drivers.modules.storage import external as external_storage
from ironic.drivers.modules.storage import noop as noop_storage
@ -49,7 +50,7 @@ class GenericHardware(hardware_type.AbstractHardwareType):
def supported_deploy_interfaces(self):
"""List of supported deploy interfaces."""
return [agent.AgentDeploy, ansible_deploy.AnsibleDeploy,
pxe.PXERamdiskDeploy, pxe.PXEAnacondaDeploy,
ramdisk.RamdiskDeploy, pxe.PXEAnacondaDeploy,
agent.CustomAgentDeploy]
@property

View File

@ -19,7 +19,6 @@ from ironic_lib import metrics_utils
from oslo_log import log as logging
from ironic.common import boot_devices
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common import states
from ironic.conductor import task_manager
@ -38,74 +37,6 @@ class PXEBoot(pxe_base.PXEBaseMixin, base.BootInterface):
capabilities = ['ramdisk_boot', 'pxe_boot']
class PXERamdiskDeploy(agent_base.AgentBaseMixin, agent_base.HeartbeatMixin,
base.DeployInterface):
def get_properties(self):
return {}
def validate(self, task):
if 'ramdisk_boot' not in task.driver.boot.capabilities:
raise exception.InvalidParameterValue(
message=_('Invalid configuration: The boot interface '
'must have the `ramdisk_boot` capability. '
'You are using an incompatible boot interface.'))
task.driver.boot.validate(task)
# Validate node capabilities
deploy_utils.validate_capabilities(task.node)
@METRICS.timer('RamdiskDeploy.deploy')
@base.deploy_step(priority=100)
@task_manager.require_exclusive_lock
def deploy(self, task):
if ('configdrive' in task.node.instance_info
and 'ramdisk_boot_configdrive' not in
task.driver.boot.capabilities):
# TODO(dtantsur): make it an actual error?
LOG.warning('A configuration drive is present in the ramdisk '
'deployment request of node %(node)s with boot '
'interface %(drv)s. The configuration drive will be '
'ignored for this deployment.',
{'node': task.node, 'drv': task.node.boot_interface})
manager_utils.node_power_action(task, states.POWER_OFF)
# Tenant neworks must enable connectivity to the boot
# location, as reboot() can otherwise be very problematic.
# IDEA(TheJulia): Maybe a "trusted environment" mode flag
# that we otherwise fail validation on for drivers that
# require explicit security postures.
with manager_utils.power_state_for_network_configuration(task):
task.driver.network.configure_tenant_networks(task)
# calling boot.prepare_instance will also set the node
# to PXE boot, and update PXE templates accordingly
task.driver.boot.prepare_instance(task)
# Power-on the instance, with PXE prepared, we're done.
manager_utils.node_power_action(task, states.POWER_ON)
LOG.info('Deployment setup for node %s done', task.node.uuid)
return None
@METRICS.timer('RamdiskDeploy.prepare')
@task_manager.require_exclusive_lock
def prepare(self, task):
node = task.node
deploy_utils.populate_storage_driver_internal_info(task)
if node.provision_state == states.DEPLOYING:
# Ask the network interface to validate itself so
# we can ensure we are able to proceed.
task.driver.network.validate(task)
manager_utils.node_power_action(task, states.POWER_OFF)
# NOTE(TheJulia): If this was any other interface, we would
# unconfigure tenant networks, add provisioning networks, etc.
task.driver.storage.attach_volumes(task)
if node.provision_state in (states.ACTIVE, states.UNRESCUING):
# In the event of takeover or unrescue.
task.driver.boot.prepare_instance(task)
class PXEAnacondaDeploy(agent_base.AgentBaseMixin, agent_base.HeartbeatMixin,
base.DeployInterface):

View File

@ -0,0 +1,99 @@
# 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.
"""
Ramdisk Deploy Interface
"""
from ironic_lib import metrics_utils
from oslo_log import log as logging
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common import states
from ironic.conductor import task_manager
from ironic.conductor import utils as manager_utils
from ironic.drivers import base
from ironic.drivers.modules import agent_base
from ironic.drivers.modules import deploy_utils
LOG = logging.getLogger(__name__)
METRICS = metrics_utils.get_metrics_logger(__name__)
class RamdiskDeploy(agent_base.AgentBaseMixin, agent_base.HeartbeatMixin,
base.DeployInterface):
def get_properties(self):
return {}
def validate(self, task):
if 'ramdisk_boot' not in task.driver.boot.capabilities:
raise exception.InvalidParameterValue(
message=_('Invalid configuration: The boot interface '
'must have the `ramdisk_boot` capability. '
'You are using an incompatible boot interface.'))
task.driver.boot.validate(task)
# Validate node capabilities
deploy_utils.validate_capabilities(task.node)
@METRICS.timer('RamdiskDeploy.deploy')
@base.deploy_step(priority=100)
@task_manager.require_exclusive_lock
def deploy(self, task):
if ('configdrive' in task.node.instance_info
and 'ramdisk_boot_configdrive' not in
task.driver.boot.capabilities):
# TODO(dtantsur): make it an actual error?
LOG.warning('A configuration drive is present in the ramdisk '
'deployment request of node %(node)s with boot '
'interface %(drv)s. The configuration drive will be '
'ignored for this deployment.',
{'node': task.node, 'drv': task.node.boot_interface})
manager_utils.node_power_action(task, states.POWER_OFF)
# Tenant neworks must enable connectivity to the boot
# location, as reboot() can otherwise be very problematic.
# IDEA(TheJulia): Maybe a "trusted environment" mode flag
# that we otherwise fail validation on for drivers that
# require explicit security postures.
with manager_utils.power_state_for_network_configuration(task):
task.driver.network.configure_tenant_networks(task)
# calling boot.prepare_instance will also set the node
# to boot, and update the templates accordingly
task.driver.boot.prepare_instance(task)
# Power-on the instance, with PXE prepared, we're done.
manager_utils.node_power_action(task, states.POWER_ON)
LOG.info('Deployment setup for node %s done', task.node.uuid)
return None
@METRICS.timer('RamdiskDeploy.prepare')
@task_manager.require_exclusive_lock
def prepare(self, task):
node = task.node
deploy_utils.populate_storage_driver_internal_info(task)
if node.provision_state == states.DEPLOYING:
# Ask the network interface to validate itself so
# we can ensure we are able to proceed.
task.driver.network.validate(task)
manager_utils.node_power_action(task, states.POWER_OFF)
# NOTE(TheJulia): If this was any other interface, we would
# unconfigure tenant networks, add provisioning networks, etc.
task.driver.storage.attach_volumes(task)
if node.provision_state in (states.ACTIVE, states.UNRESCUING):
# In the event of takeover or unrescue.
task.driver.boot.prepare_instance(task)

View File

@ -831,269 +831,6 @@ class PXEBootTestCase(db_base.DbTestCase):
secure_boot_mock.assert_called_once_with(task)
class PXERamdiskDeployTestCase(db_base.DbTestCase):
def setUp(self):
super(PXERamdiskDeployTestCase, self).setUp()
self.temp_dir = tempfile.mkdtemp()
self.config(tftp_root=self.temp_dir, group='pxe')
self.temp_dir = tempfile.mkdtemp()
self.config(images_path=self.temp_dir, group='pxe')
self.config(enabled_deploy_interfaces=['ramdisk'])
self.config(enabled_boot_interfaces=['pxe'])
for iface in drivers_base.ALL_INTERFACES:
impl = 'fake'
if iface == 'network':
impl = 'noop'
if iface == 'deploy':
impl = 'ramdisk'
if iface == 'boot':
impl = 'pxe'
config_kwarg = {'enabled_%s_interfaces' % iface: [impl],
'default_%s_interface' % iface: impl}
self.config(**config_kwarg)
self.config(enabled_hardware_types=['fake-hardware'])
instance_info = {'kernel': 'kernelUUID',
'ramdisk': 'ramdiskUUID'}
self.node = obj_utils.create_test_node(
self.context,
driver='fake-hardware',
instance_info=instance_info,
driver_info=DRV_INFO_DICT,
driver_internal_info=DRV_INTERNAL_INFO_DICT)
self.port = obj_utils.create_test_port(self.context,
node_id=self.node.id)
@mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
@mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
@mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
@mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
@mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
def test_prepare_instance_ramdisk(
self, get_image_info_mock, cache_mock,
dhcp_factory_mock, switch_pxe_config_mock,
set_boot_device_mock):
provider_mock = mock.MagicMock()
dhcp_factory_mock.return_value = provider_mock
self.node.provision_state = states.DEPLOYING
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, ipxe_enabled=False)
dhcp_opts += pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=False, ip_version=6)
pxe_config_path = pxe_utils.get_pxe_config_file_path(
task.node.uuid)
task.node.properties['capabilities'] = 'boot_option:netboot'
task.node.driver_internal_info['is_whole_disk_image'] = False
task.driver.deploy.prepare(task)
task.driver.deploy.deploy(task)
get_image_info_mock.assert_called_once_with(task,
ipxe_enabled=False)
cache_mock.assert_called_once_with(
task, image_info, ipxe_enabled=False)
provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
switch_pxe_config_mock.assert_called_once_with(
pxe_config_path, None,
'bios', False, ipxe_enabled=False, iscsi_boot=False,
ramdisk_boot=True, anaconda_boot=False)
set_boot_device_mock.assert_called_once_with(task,
boot_devices.PXE,
persistent=True)
@mock.patch.object(pxe.LOG, 'warning', autospec=True)
@mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
@mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
@mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
@mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
def test_deploy(self, mock_image_info, mock_cache,
mock_dhcp_factory, mock_switch_config, mock_warning):
image_info = {'kernel': ('', '/path/to/kernel'),
'ramdisk': ('', '/path/to/ramdisk')}
mock_image_info.return_value = image_info
i_info = self.node.instance_info
i_info.update({'capabilities': {'boot_option': 'ramdisk'}})
self.node.instance_info = i_info
self.node.save()
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertIsNone(task.driver.deploy.deploy(task))
mock_image_info.assert_called_once_with(task, ipxe_enabled=False)
mock_cache.assert_called_once_with(
task, image_info, ipxe_enabled=False)
self.assertFalse(mock_warning.called)
i_info['configdrive'] = 'meow'
self.node.instance_info = i_info
self.node.save()
mock_warning.reset_mock()
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertIsNone(task.driver.deploy.deploy(task))
self.assertTrue(mock_warning.called)
@mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
def test_prepare(self, mock_prepare_instance):
node = self.node
node.provision_state = states.DEPLOYING
node.instance_info = {}
node.save()
with task_manager.acquire(self.context, node.uuid) as task:
task.driver.deploy.prepare(task)
self.assertFalse(mock_prepare_instance.called)
@mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
def test_prepare_active(self, mock_prepare_instance):
node = self.node
node.provision_state = states.ACTIVE
node.save()
with task_manager.acquire(self.context, node.uuid) as task:
task.driver.deploy.prepare(task)
mock_prepare_instance.assert_called_once_with(mock.ANY, task)
@mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
def test_prepare_unrescuing(self, mock_prepare_instance):
node = self.node
node.provision_state = states.UNRESCUING
node.save()
with task_manager.acquire(self.context, node.uuid) as task:
task.driver.deploy.prepare(task)
mock_prepare_instance.assert_called_once_with(mock.ANY, task)
@mock.patch.object(deploy_utils, 'validate_image_properties',
autospec=True)
def test_validate(self, mock_validate_img):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.deploy.validate(task)
self.assertTrue(mock_validate_img.called)
@mock.patch.object(deploy_utils, 'validate_image_properties',
autospec=True)
def test_validate_with_boot_iso(self, mock_validate_img):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.instance_info = {
'boot_iso': 'isoUUID'
}
task.driver.deploy.validate(task)
self.assertTrue(mock_validate_img.called)
@mock.patch.object(fake.FakeBoot, 'validate', autospec=True)
@mock.patch.object(deploy_utils, 'validate_image_properties',
autospec=True)
def test_validate_interface_mismatch(self, mock_validate_image,
mock_boot_validate):
node = self.node
node.boot_interface = 'fake'
node.save()
self.config(enabled_boot_interfaces=['fake'],
default_boot_interface='fake')
with task_manager.acquire(self.context, node.uuid) as task:
error = self.assertRaises(exception.InvalidParameterValue,
task.driver.deploy.validate, task)
error_message = ('Invalid configuration: The boot interface must '
'have the `ramdisk_boot` capability. You are '
'using an incompatible boot interface.')
self.assertEqual(error_message, str(error))
self.assertFalse(mock_boot_validate.called)
self.assertFalse(mock_validate_image.called)
@mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
def test_validate_calls_boot_validate(self, mock_validate):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.deploy.validate(task)
mock_validate.assert_called_once_with(mock.ANY, task)
@mock.patch.object(manager_utils, 'restore_power_state_if_needed',
autospec=True)
@mock.patch.object(manager_utils, 'power_on_node_if_needed',
autospec=True)
@mock.patch.object(pxe.LOG, 'warning', autospec=True)
@mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
@mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
@mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
@mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
def test_deploy_with_smartnic_port(
self, mock_image_info, mock_cache,
mock_dhcp_factory, mock_switch_config, mock_warning,
power_on_node_if_needed_mock, restore_power_state_mock):
image_info = {'kernel': ('', '/path/to/kernel'),
'ramdisk': ('', '/path/to/ramdisk')}
mock_image_info.return_value = image_info
i_info = self.node.instance_info
i_info.update({'capabilities': {'boot_option': 'ramdisk'}})
self.node.instance_info = i_info
self.node.save()
with task_manager.acquire(self.context, self.node.uuid) as task:
power_on_node_if_needed_mock.return_value = states.POWER_OFF
self.assertIsNone(task.driver.deploy.deploy(task))
mock_image_info.assert_called_once_with(task, ipxe_enabled=False)
mock_cache.assert_called_once_with(
task, image_info, ipxe_enabled=False)
self.assertFalse(mock_warning.called)
power_on_node_if_needed_mock.assert_called_once_with(task)
restore_power_state_mock.assert_called_once_with(
task, states.POWER_OFF)
i_info['configdrive'] = 'meow'
self.node.instance_info = i_info
self.node.save()
mock_warning.reset_mock()
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertIsNone(task.driver.deploy.deploy(task))
self.assertTrue(mock_warning.called)
@mock.patch.object(agent_base, 'get_steps', autospec=True)
def test_get_clean_steps(self, mock_get_steps):
# Test getting clean steps
mock_steps = [{'priority': 10, 'interface': 'deploy',
'step': 'erase_devices'}]
mock_get_steps.return_value = mock_steps
with task_manager.acquire(self.context, self.node.uuid) as task:
steps = task.driver.deploy.get_clean_steps(task)
mock_get_steps.assert_called_once_with(
task, 'clean', interface='deploy',
override_priorities={'erase_devices': None,
'erase_devices_metadata': None})
self.assertEqual(mock_steps, steps)
def test_get_deploy_steps(self):
# Only the default deploy step exists in the ramdisk deploy
expected = [{'argsinfo': None, 'interface': 'deploy', 'priority': 100,
'step': 'deploy'}]
with task_manager.acquire(self.context, self.node.uuid) as task:
steps = task.driver.deploy.get_deploy_steps(task)
self.assertEqual(expected, steps)
@mock.patch.object(agent_base, 'execute_step', autospec=True)
def test_execute_clean_step(self, mock_execute_step):
step = {
'priority': 10,
'interface': 'deploy',
'step': 'erase_devices',
'reboot_requested': False
}
with task_manager.acquire(self.context, self.node.uuid) as task:
result = task.driver.deploy.execute_clean_step(task, step)
self.assertIs(result, mock_execute_step.return_value)
mock_execute_step.assert_called_once_with(task, step, 'clean')
@mock.patch.object(deploy_utils, 'prepare_inband_cleaning', autospec=True)
def test_prepare_cleaning(self, prepare_inband_cleaning_mock):
prepare_inband_cleaning_mock.return_value = states.CLEANWAIT
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertEqual(
states.CLEANWAIT, task.driver.deploy.prepare_cleaning(task))
prepare_inband_cleaning_mock.assert_called_once_with(
task, manage_boot=True)
@mock.patch.object(deploy_utils, 'tear_down_inband_cleaning',
autospec=True)
def test_tear_down_cleaning(self, tear_down_cleaning_mock):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.deploy.tear_down_cleaning(task)
tear_down_cleaning_mock.assert_called_once_with(
task, manage_boot=True)
class PXEAnacondaDeployTestCase(db_base.DbTestCase):
def setUp(self):

View File

@ -0,0 +1,302 @@
# 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 ramdisk deploy."""
import tempfile
from unittest import mock
from oslo_config import cfg
from ironic.common import boot_devices
from ironic.common import dhcp_factory
from ironic.common import exception
from ironic.common import pxe_utils
from ironic.common import states
from ironic.conductor import task_manager
from ironic.conductor import utils as manager_utils
from ironic.drivers import base as drivers_base
from ironic.drivers.modules import agent_base
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules import fake
from ironic.drivers.modules import pxe
from ironic.drivers.modules import ramdisk
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
DRV_INFO_DICT = db_utils.get_test_pxe_driver_info()
DRV_INTERNAL_INFO_DICT = db_utils.get_test_pxe_driver_internal_info()
class RamdiskDeployTestCase(db_base.DbTestCase):
def setUp(self):
super(RamdiskDeployTestCase, self).setUp()
self.temp_dir = tempfile.mkdtemp()
self.config(tftp_root=self.temp_dir, group='pxe')
self.temp_dir = tempfile.mkdtemp()
self.config(images_path=self.temp_dir, group='pxe')
self.config(enabled_deploy_interfaces=['ramdisk'])
self.config(enabled_boot_interfaces=['pxe'])
for iface in drivers_base.ALL_INTERFACES:
impl = 'fake'
if iface == 'network':
impl = 'noop'
if iface == 'deploy':
impl = 'ramdisk'
if iface == 'boot':
impl = 'pxe'
config_kwarg = {'enabled_%s_interfaces' % iface: [impl],
'default_%s_interface' % iface: impl}
self.config(**config_kwarg)
self.config(enabled_hardware_types=['fake-hardware'])
instance_info = {'kernel': 'kernelUUID',
'ramdisk': 'ramdiskUUID'}
self.node = obj_utils.create_test_node(
self.context,
driver='fake-hardware',
instance_info=instance_info,
driver_info=DRV_INFO_DICT,
driver_internal_info=DRV_INTERNAL_INFO_DICT)
self.port = obj_utils.create_test_port(self.context,
node_id=self.node.id)
@mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
@mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
@mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
@mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
@mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
def test_prepare_instance_ramdisk(
self, get_image_info_mock, cache_mock,
dhcp_factory_mock, switch_pxe_config_mock,
set_boot_device_mock):
provider_mock = mock.MagicMock()
dhcp_factory_mock.return_value = provider_mock
self.node.provision_state = states.DEPLOYING
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, ipxe_enabled=False)
dhcp_opts += pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=False, ip_version=6)
pxe_config_path = pxe_utils.get_pxe_config_file_path(
task.node.uuid)
task.node.properties['capabilities'] = 'boot_option:netboot'
task.node.driver_internal_info['is_whole_disk_image'] = False
task.driver.deploy.prepare(task)
task.driver.deploy.deploy(task)
get_image_info_mock.assert_called_once_with(task,
ipxe_enabled=False)
cache_mock.assert_called_once_with(
task, image_info, ipxe_enabled=False)
provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
switch_pxe_config_mock.assert_called_once_with(
pxe_config_path, None,
'bios', False, ipxe_enabled=False, iscsi_boot=False,
ramdisk_boot=True, anaconda_boot=False)
set_boot_device_mock.assert_called_once_with(task,
boot_devices.PXE,
persistent=True)
@mock.patch.object(ramdisk.LOG, 'warning', autospec=True)
@mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
@mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
@mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
@mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
def test_deploy(self, mock_image_info, mock_cache,
mock_dhcp_factory, mock_switch_config, mock_warning):
image_info = {'kernel': ('', '/path/to/kernel'),
'ramdisk': ('', '/path/to/ramdisk')}
mock_image_info.return_value = image_info
i_info = self.node.instance_info
i_info.update({'capabilities': {'boot_option': 'ramdisk'}})
self.node.instance_info = i_info
self.node.save()
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertIsNone(task.driver.deploy.deploy(task))
mock_image_info.assert_called_once_with(task, ipxe_enabled=False)
mock_cache.assert_called_once_with(
task, image_info, ipxe_enabled=False)
self.assertFalse(mock_warning.called)
i_info['configdrive'] = 'meow'
self.node.instance_info = i_info
self.node.save()
mock_warning.reset_mock()
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertIsNone(task.driver.deploy.deploy(task))
self.assertTrue(mock_warning.called)
@mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
def test_prepare(self, mock_prepare_instance):
node = self.node
node.provision_state = states.DEPLOYING
node.instance_info = {}
node.save()
with task_manager.acquire(self.context, node.uuid) as task:
task.driver.deploy.prepare(task)
self.assertFalse(mock_prepare_instance.called)
@mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
def test_prepare_active(self, mock_prepare_instance):
node = self.node
node.provision_state = states.ACTIVE
node.save()
with task_manager.acquire(self.context, node.uuid) as task:
task.driver.deploy.prepare(task)
mock_prepare_instance.assert_called_once_with(mock.ANY, task)
@mock.patch.object(pxe.PXEBoot, 'prepare_instance', autospec=True)
def test_prepare_unrescuing(self, mock_prepare_instance):
node = self.node
node.provision_state = states.UNRESCUING
node.save()
with task_manager.acquire(self.context, node.uuid) as task:
task.driver.deploy.prepare(task)
mock_prepare_instance.assert_called_once_with(mock.ANY, task)
@mock.patch.object(deploy_utils, 'validate_image_properties',
autospec=True)
def test_validate(self, mock_validate_img):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.deploy.validate(task)
self.assertTrue(mock_validate_img.called)
@mock.patch.object(deploy_utils, 'validate_image_properties',
autospec=True)
def test_validate_with_boot_iso(self, mock_validate_img):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.instance_info = {
'boot_iso': 'isoUUID'
}
task.driver.deploy.validate(task)
self.assertTrue(mock_validate_img.called)
@mock.patch.object(fake.FakeBoot, 'validate', autospec=True)
@mock.patch.object(deploy_utils, 'validate_image_properties',
autospec=True)
def test_validate_interface_mismatch(self, mock_validate_image,
mock_boot_validate):
node = self.node
node.boot_interface = 'fake'
node.save()
self.config(enabled_boot_interfaces=['fake'],
default_boot_interface='fake')
with task_manager.acquire(self.context, node.uuid) as task:
error = self.assertRaises(exception.InvalidParameterValue,
task.driver.deploy.validate, task)
error_message = ('Invalid configuration: The boot interface must '
'have the `ramdisk_boot` capability. You are '
'using an incompatible boot interface.')
self.assertEqual(error_message, str(error))
self.assertFalse(mock_boot_validate.called)
self.assertFalse(mock_validate_image.called)
@mock.patch.object(pxe.PXEBoot, 'validate', autospec=True)
def test_validate_calls_boot_validate(self, mock_validate):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.deploy.validate(task)
mock_validate.assert_called_once_with(mock.ANY, task)
@mock.patch.object(manager_utils, 'restore_power_state_if_needed',
autospec=True)
@mock.patch.object(manager_utils, 'power_on_node_if_needed',
autospec=True)
@mock.patch.object(ramdisk.LOG, 'warning', autospec=True)
@mock.patch.object(deploy_utils, 'switch_pxe_config', autospec=True)
@mock.patch.object(dhcp_factory, 'DHCPFactory', autospec=True)
@mock.patch.object(pxe_utils, 'cache_ramdisk_kernel', autospec=True)
@mock.patch.object(pxe_utils, 'get_instance_image_info', autospec=True)
def test_deploy_with_smartnic_port(
self, mock_image_info, mock_cache,
mock_dhcp_factory, mock_switch_config, mock_warning,
power_on_node_if_needed_mock, restore_power_state_mock):
image_info = {'kernel': ('', '/path/to/kernel'),
'ramdisk': ('', '/path/to/ramdisk')}
mock_image_info.return_value = image_info
i_info = self.node.instance_info
i_info.update({'capabilities': {'boot_option': 'ramdisk'}})
self.node.instance_info = i_info
self.node.save()
with task_manager.acquire(self.context, self.node.uuid) as task:
power_on_node_if_needed_mock.return_value = states.POWER_OFF
self.assertIsNone(task.driver.deploy.deploy(task))
mock_image_info.assert_called_once_with(task, ipxe_enabled=False)
mock_cache.assert_called_once_with(
task, image_info, ipxe_enabled=False)
self.assertFalse(mock_warning.called)
power_on_node_if_needed_mock.assert_called_once_with(task)
restore_power_state_mock.assert_called_once_with(
task, states.POWER_OFF)
i_info['configdrive'] = 'meow'
self.node.instance_info = i_info
self.node.save()
mock_warning.reset_mock()
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertIsNone(task.driver.deploy.deploy(task))
self.assertTrue(mock_warning.called)
@mock.patch.object(agent_base, 'get_steps', autospec=True)
def test_get_clean_steps(self, mock_get_steps):
# Test getting clean steps
mock_steps = [{'priority': 10, 'interface': 'deploy',
'step': 'erase_devices'}]
mock_get_steps.return_value = mock_steps
with task_manager.acquire(self.context, self.node.uuid) as task:
steps = task.driver.deploy.get_clean_steps(task)
mock_get_steps.assert_called_once_with(
task, 'clean', interface='deploy',
override_priorities={'erase_devices': None,
'erase_devices_metadata': None})
self.assertEqual(mock_steps, steps)
def test_get_deploy_steps(self):
# Only the default deploy step exists in the ramdisk deploy
expected = [{'argsinfo': None, 'interface': 'deploy', 'priority': 100,
'step': 'deploy'}]
with task_manager.acquire(self.context, self.node.uuid) as task:
steps = task.driver.deploy.get_deploy_steps(task)
self.assertEqual(expected, steps)
@mock.patch.object(agent_base, 'execute_step', autospec=True)
def test_execute_clean_step(self, mock_execute_step):
step = {
'priority': 10,
'interface': 'deploy',
'step': 'erase_devices',
'reboot_requested': False
}
with task_manager.acquire(self.context, self.node.uuid) as task:
result = task.driver.deploy.execute_clean_step(task, step)
self.assertIs(result, mock_execute_step.return_value)
mock_execute_step.assert_called_once_with(task, step, 'clean')
@mock.patch.object(deploy_utils, 'prepare_inband_cleaning', autospec=True)
def test_prepare_cleaning(self, prepare_inband_cleaning_mock):
prepare_inband_cleaning_mock.return_value = states.CLEANWAIT
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertEqual(
states.CLEANWAIT, task.driver.deploy.prepare_cleaning(task))
prepare_inband_cleaning_mock.assert_called_once_with(
task, manage_boot=True)
@mock.patch.object(deploy_utils, 'tear_down_inband_cleaning',
autospec=True)
def test_tear_down_cleaning(self, tear_down_cleaning_mock):
with task_manager.acquire(self.context, self.node.uuid) as task:
task.driver.deploy.tear_down_cleaning(task)
tear_down_cleaning_mock.assert_called_once_with(
task, manage_boot=True)

View File

@ -90,7 +90,7 @@ ironic.hardware.interfaces.deploy =
custom-agent = ironic.drivers.modules.agent:CustomAgentDeploy
direct = ironic.drivers.modules.agent:AgentDeploy
fake = ironic.drivers.modules.fake:FakeDeploy
ramdisk = ironic.drivers.modules.pxe:PXERamdiskDeploy
ramdisk = ironic.drivers.modules.ramdisk:RamdiskDeploy
ironic.hardware.interfaces.inspect =
fake = ironic.drivers.modules.fake:FakeInspect