openstack/ironic
Code Issues Proposed changes

Finalize removal of ipxe_enabled option

Browse Source 
Remove the dynamically registered ipxe_enabled option and say goodbye.
Further extracts common bits to the PXEBaseMixin, tuning tests here and
there.

Story: 2007003
Task: 37779

Change-Id: I7c1b2a984d45bd63b4e95b62ce02960924c2ce17
This commit is contained in:
Kaifeng Wang 2019-12-10 19:14:18 +08:00
parent 6ba983b615
commit 34d34b3a9d
11 changed files with 346 additions and 433 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
devstack/lib/ironic
View File

@ -1567,7 +1567,6 @@ function configure_ironic_conductor {
local pxebin
pxebin=`basename $IRONIC_PXE_BOOT_IMAGE`
uefipxebin=`basename $(get_uefi_ipxe_boot_file)`
iniset $IRONIC_CONF_FILE pxe ipxe_enabled True
iniset $IRONIC_CONF_FILE pxe pxe_config_template '$pybasedir/drivers/modules/ipxe_config.template'
iniset $IRONIC_CONF_FILE pxe pxe_bootfile_name $pxebin
iniset $IRONIC_CONF_FILE pxe uefi_pxe_config_template '$pybasedir/drivers/modules/ipxe_config.template'

7
ironic/common/config.py
View File

@ -29,11 +29,4 @@ def parse_args(argv, default_config_files=None):
version=version.version_info.release_string(),
default_config_files=default_config_files)
rpc.init(cfg.CONF)
# TODO(kaifeng) Remove ipxe_enabled option handling after ipxe support
# is completely removed from the pxe interface.
ipxe_enabled = cfg.BoolOpt('ipxe_enabled', default=False,
deprecated_for_removal=True)
cfg.CONF.register_opt(ipxe_enabled, group='pxe')
cfg.CONF.set_override('ipxe_enabled', False, group='pxe')
profiler_opts.set_defaults(cfg.CONF)

12
ironic/common/pxe_utils.py
View File

@ -20,7 +20,6 @@ from ironic_lib import utils as ironic_utils
from oslo_log import log as logging
from oslo_utils import excutils
from oslo_utils import fileutils
from oslo_utils import importutils
from oslo_utils import netutils
from ironic.common import dhcp_factory
@ -546,16 +545,7 @@ def is_ipxe_enabled(task):
:returns: boolean true if ``[pxe]ipxe_enabled`` is configured
or if the task driver instance is the iPXE driver.
"""
# NOTE(TheJulia): importutils used here as we seem to get in circular
# import weirdness otherwise, specifically when the classes that use
# the pxe interface as their parent.
# TODO(TheJulia): We should remove this as soon as it is no longer
# required to help us bridge the split of the interfaces and helper
# methods.
iPXEBoot = importutils.import_class(
'ironic.drivers.modules.ipxe.iPXEBoot')
return CONF.pxe.ipxe_enabled or isinstance(task.driver.boot,
iPXEBoot)
return 'ipxe_boot' in task.driver.boot.capabilities
def parse_driver_info(node, mode='deploy'):

94
ironic/drivers/modules/ipxe.py
View File

@ -15,24 +15,9 @@
iPXE Boot Interface
"""
from ironic_lib import metrics_utils
from oslo_log import log as logging
from ironic.common import exception
from ironic.common.glance_service import service_utils
from ironic.common.i18n import _
from ironic.common import pxe_utils
from ironic.conf import CONF
from ironic.drivers import base
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules import pxe
from ironic.drivers.modules import pxe_base
from ironic.drivers import utils as driver_utils
LOG = logging.getLogger(__name__)
METRICS = metrics_utils.get_metrics_logger(__name__)
COMMON_PROPERTIES = pxe_base.COMMON_PROPERTIES
class iPXEBoot(pxe_base.PXEBaseMixin, base.BootInterface):
@ -43,82 +28,3 @@ class iPXEBoot(pxe_base.PXEBaseMixin, base.BootInterface):
def __init__(self):
pxe_utils.create_ipxe_boot_script()
def _validate_common(self, task):
node = task.node
if not driver_utils.get_node_mac_addresses(task):
raise exception.MissingParameterValue(
_("Node %s does not have any port associated with it.")
% node.uuid)
if not CONF.deploy.http_url or not CONF.deploy.http_root:
raise exception.MissingParameterValue(_(
"iPXE boot is enabled but no HTTP URL or HTTP "
"root was specified."))
# Check the trusted_boot capabilities value.
deploy_utils.validate_capabilities(node)
if deploy_utils.is_trusted_boot_requested(node):
# Check if 'boot_option' and boot mode is compatible with
# trusted boot.
# NOTE(TheJulia): So in theory (huge theory here, not put to
# practice or tested), that one can define the kernel as tboot
# and define the actual kernel and ramdisk as appended data.
# Similar to how one can iPXE load the XEN hypervisor.
# tboot mailing list seem to indicate pxe/ipxe support, or
# more specifically avoiding breaking the scenarios of use,
# but there is also no definitive documentation on the subject.
LOG.warning('Trusted boot has been requested for %(node)s in '
'concert with iPXE. This is not a supported '
'configuration for an ironic deployment.',
{'node': node.uuid})
pxe.validate_boot_parameters_for_trusted_boot(node)
pxe_utils.parse_driver_info(node)
@METRICS.timer('iPXEBoot.validate')
def validate(self, task):
"""Validate the PXE-specific info for booting deploy/instance images.
This method validates the PXE-specific info for booting the
ramdisk and instance on the node. If invalid, raises an
exception; otherwise returns None.
:param task: a task from TaskManager.
:returns: None
:raises: InvalidParameterValue, if some parameters are invalid.
:raises: MissingParameterValue, if some required parameters are
missing.
"""
self._validate_common(task)
# NOTE(TheJulia): If we're not writing an image, we can skip
# the remainder of this method.
if (not task.driver.storage.should_write_image(task)):
return
node = task.node
d_info = deploy_utils.get_image_instance_info(node)
if (node.driver_internal_info.get('is_whole_disk_image')
or deploy_utils.get_boot_option(node) == 'local'):
props = []
elif service_utils.is_glance_image(d_info['image_source']):
props = ['kernel_id', 'ramdisk_id']
else:
props = ['kernel', 'ramdisk']
deploy_utils.validate_image_properties(task.context, d_info, props)
@METRICS.timer('iPXEBoot.validate_inspection')
def validate_inspection(self, task):
"""Validate that the node has required properties for inspection.
:param task: A TaskManager instance with the node being checked
:raises: UnsupportedDriverExtension
"""
try:
self._validate_common(task)
except exception.MissingParameterValue:
# Fall back to non-managed in-band inspection
raise exception.UnsupportedDriverExtension(
driver=task.node.driver, extension='inspection')

79
ironic/drivers/modules/pxe.py
View File

@ -19,9 +19,7 @@ from ironic_lib import metrics_utils
from oslo_log import log as logging
from ironic.common import exception
from ironic.common.glance_service import service_utils
from ironic.common.i18n import _
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
@ -29,89 +27,14 @@ from ironic.drivers import base
from ironic.drivers.modules import agent
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules import pxe_base
from ironic.drivers import utils as driver_utils
LOG = logging.getLogger(__name__)
METRICS = metrics_utils.get_metrics_logger(__name__)
COMMON_PROPERTIES = pxe_base.COMMON_PROPERTIES
# NOTE(TheJulia): This was previously a public method to the code being
# moved. This mapping should be removed in the T* cycle.
validate_boot_parameters_for_trusted_boot = pxe_utils.validate_boot_parameters_for_trusted_boot # noqa
TFTPImageCache = pxe_utils.TFTPImageCache
# NOTE(TheJulia): End section of mappings for migrated common pxe code.
class PXEBoot(pxe_base.PXEBaseMixin, base.BootInterface):
# TODO(TheJulia): iscsi_volume_boot should be removed from
# the list below once ipxe support is removed from the PXE
# interface.
capabilities = ['iscsi_volume_boot', 'ramdisk_boot', 'pxe_boot']
def _validate_common(self, task):
node = task.node
if not driver_utils.get_node_mac_addresses(task):
raise exception.MissingParameterValue(
_("Node %s does not have any port associated with it.")
% node.uuid)
# Check the trusted_boot capabilities value.
deploy_utils.validate_capabilities(node)
if deploy_utils.is_trusted_boot_requested(node):
# Check if 'boot_option' and boot mode is compatible with
# trusted boot.
validate_boot_parameters_for_trusted_boot(node)
pxe_utils.parse_driver_info(node)
@METRICS.timer('PXEBoot.validate')
def validate(self, task):
"""Validate the PXE-specific info for booting deploy/instance images.
This method validates the PXE-specific info for booting the
ramdisk and instance on the node. If invalid, raises an
exception; otherwise returns None.
:param task: a task from TaskManager.
:returns: None
:raises: InvalidParameterValue, if some parameters are invalid.
:raises: MissingParameterValue, if some required parameters are
missing.
"""
self._validate_common(task)
# NOTE(TheJulia): If we're not writing an image, we can skip
# the remainder of this method.
if (not task.driver.storage.should_write_image(task)):
return
node = task.node
d_info = deploy_utils.get_image_instance_info(node)
if (node.driver_internal_info.get('is_whole_disk_image')
or deploy_utils.get_boot_option(node) == 'local'):
props = []
elif service_utils.is_glance_image(d_info['image_source']):
props = ['kernel_id', 'ramdisk_id']
else:
props = ['kernel', 'ramdisk']
deploy_utils.validate_image_properties(task.context, d_info, props)
@METRICS.timer('PXEBoot.validate_inspection')
def validate_inspection(self, task):
"""Validate that the node has required properties for inspection.
:param task: A TaskManager instance with the node being checked
:raises: UnsupportedDriverExtension
"""
try:
self._validate_common(task)
except exception.MissingParameterValue:
# Fall back to non-managed in-band inspection
raise exception.UnsupportedDriverExtension(
driver=task.node.driver, extension='inspection')
capabilities = ['ramdisk_boot', 'pxe_boot']
class PXERamdiskDeploy(agent.AgentDeploy):

83
ironic/drivers/modules/pxe_base.py
View File

@ -22,6 +22,7 @@ from oslo_utils import strutils
from ironic.common import boot_devices
from ironic.common import dhcp_factory
from ironic.common import exception
from ironic.common.glance_service import service_utils
from ironic.common.i18n import _
from ironic.common import pxe_utils
from ironic.common import states
@ -29,6 +30,7 @@ from ironic.conductor import task_manager
from ironic.conductor import utils as manager_utils
from ironic.drivers.modules import boot_mode_utils
from ironic.drivers.modules import deploy_utils
from ironic.drivers import utils as driver_utils
CONF = cfg.CONF
@ -306,6 +308,73 @@ class PXEBaseMixin(object):
manager_utils.node_set_boot_device(task, boot_device,
persistent=persistent)
def _validate_common(self, task):
node = task.node
if not driver_utils.get_node_mac_addresses(task):
raise exception.MissingParameterValue(
_("Node %s does not have any port associated with it.")
% node.uuid)
if self.ipxe_enabled:
if not CONF.deploy.http_url or not CONF.deploy.http_root:
raise exception.MissingParameterValue(_(
"iPXE boot is enabled but no HTTP URL or HTTP "
"root was specified."))
# Check the trusted_boot capabilities value.
deploy_utils.validate_capabilities(node)
if deploy_utils.is_trusted_boot_requested(node):
# Check if 'boot_option' and boot mode is compatible with
# trusted boot.
if self.ipxe_enabled:
# NOTE(TheJulia): So in theory (huge theory here, not put to
# practice or tested), that one can define the kernel as tboot
# and define the actual kernel and ramdisk as appended data.
# Similar to how one can iPXE load the XEN hypervisor.
# tboot mailing list seem to indicate pxe/ipxe support, or
# more specifically avoiding breaking the scenarios of use,
# but there is also no definitive documentation on the subject.
LOG.warning('Trusted boot has been requested for %(node)s in '
'concert with iPXE. This is not a supported '
'configuration for an ironic deployment.',
{'node': node.uuid})
pxe_utils.validate_boot_parameters_for_trusted_boot(node)
pxe_utils.parse_driver_info(node)
@METRICS.timer('PXEBaseMixin.validate')
def validate(self, task):
"""Validate the PXE-specific info for booting deploy/instance images.
This method validates the PXE-specific info for booting the
ramdisk and instance on the node. If invalid, raises an
exception; otherwise returns None.
:param task: a task from TaskManager.
:returns: None
:raises: InvalidParameterValue, if some parameters are invalid.
:raises: MissingParameterValue, if some required parameters are
missing.
"""
self._validate_common(task)
# NOTE(TheJulia): If we're not writing an image, we can skip
# the remainder of this method.
if (not task.driver.storage.should_write_image(task)):
return
node = task.node
d_info = deploy_utils.get_image_instance_info(node)
if (node.driver_internal_info.get('is_whole_disk_image')
or deploy_utils.get_boot_option(node) == 'local'):
props = []
elif service_utils.is_glance_image(d_info['image_source']):
props = ['kernel_id', 'ramdisk_id']
else:
props = ['kernel', 'ramdisk']
deploy_utils.validate_image_properties(task.context, d_info, props)
@METRICS.timer('PXEBaseMixin.validate_rescue')
def validate_rescue(self, task):
"""Validate that the node has required properties for rescue.
@ -316,6 +385,20 @@ class PXEBaseMixin(object):
"""
pxe_utils.parse_driver_info(task.node, mode='rescue')
@METRICS.timer('PXEBaseMixin.validate_inspection')
def validate_inspection(self, task):
"""Validate that the node has required properties for inspection.
:param task: A TaskManager instance with the node being checked
:raises: UnsupportedDriverExtension
"""
try:
self._validate_common(task)
except exception.MissingParameterValue:
# Fall back to non-managed in-band inspection
raise exception.UnsupportedDriverExtension(
driver=task.node.driver, extension='inspection')
def _persistent_ramdisk_boot(self, node):
"""If the ramdisk should be configured as a persistent boot device."""
value = node.driver_info.get('force_persistent_boot_device', 'Default')

466
ironic/tests/unit/common/test_pxe_utils.py
View File

@ -308,7 +308,6 @@ class TestPXEUtils(db_base.DbTestCase):
@mock.patch('ironic.common.utils.create_link_without_raise', autospec=True)
@mock.patch('ironic_lib.utils.unlink_without_raise', autospec=True)
def test__write_mac_ipxe_configs(self, unlink_mock, create_link_mock):
self.config(ipxe_enabled=True, group='pxe')
port_1 = object_utils.create_test_port(
self.context, node_id=self.node.id,
address='11:22:33:44:55:66', uuid=uuidutils.generate_uuid())
@ -526,7 +525,6 @@ class TestPXEUtils(db_base.DbTestCase):
def test_create_pxe_config_uefi_ipxe(self, ensure_tree_mock, render_mock,
write_mock, link_mac_pxe_mock,
chmod_mock):
self.config(ipxe_enabled=True, group='pxe')
ipxe_template = "ironic/drivers/modules/ipxe_config.template"
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.properties['capabilities'] = 'boot_mode:uefi'
@ -740,107 +738,6 @@ class TestPXEUtils(db_base.DbTestCase):
self._test_get_kernel_ramdisk_info(expected_dir, mode='rescue',
ipxe_enabled=True)
def _dhcp_options_for_instance_ipxe(self, task, boot_file, ip_version=4):
self.config(ipxe_enabled=True, group='pxe')
self.config(ipxe_boot_script='/test/boot.ipxe', group='pxe')
self.config(tftp_root='/tftp-path/', group='pxe')
if ip_version == 4:
self.config(tftp_server='192.0.2.1', group='pxe')
self.config(http_url='http://192.0.3.2:1234', group='deploy')
self.config(ipxe_boot_script='/test/boot.ipxe', group='pxe')
elif ip_version == 6:
self.config(tftp_server='ff80::1', group='pxe')
self.config(http_url='http://[ff80::1]:1234', group='deploy')
self.config(dhcp_provider='isc', group='dhcp')
if ip_version == 6:
# NOTE(TheJulia): DHCPv6 RFCs seem to indicate that the prior
# options are not imported, although they may be supported
# by vendors. The apparent proper option is to return a
# URL in the field https://tools.ietf.org/html/rfc5970#section-3
expected_boot_script_url = 'http://[ff80::1]:1234/boot.ipxe'
expected_info = [{'opt_name': '!175,59',
'opt_value': 'tftp://[ff80::1]/fake-bootfile',
'ip_version': ip_version},
{'opt_name': '59',
'opt_value': expected_boot_script_url,
'ip_version': ip_version}]
elif ip_version == 4:
expected_boot_script_url = 'http://192.0.3.2:1234/boot.ipxe'
expected_info = [{'opt_name': '!175,67',
'opt_value': boot_file,
'ip_version': ip_version},
{'opt_name': '66',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': '150',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': '67',
'opt_value': expected_boot_script_url,
'ip_version': ip_version},
{'opt_name': 'server-ip-address',
'opt_value': '192.0.2.1',
'ip_version': ip_version}]
self.assertItemsEqual(expected_info,
pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=True))
self.config(dhcp_provider='neutron', group='dhcp')
if ip_version == 6:
# Boot URL variable set from prior test of isc parameters.
expected_info = [{'opt_name': 'tag:!ipxe6,59',
'opt_value': 'tftp://[ff80::1]/fake-bootfile',
'ip_version': ip_version},
{'opt_name': 'tag:ipxe6,59',
'opt_value': expected_boot_script_url,
'ip_version': ip_version}]
elif ip_version == 4:
expected_info = [{'opt_name': 'tag:!ipxe,67',
'opt_value': boot_file,
'ip_version': ip_version},
{'opt_name': '66',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': '150',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': 'tag:ipxe,67',
'opt_value': expected_boot_script_url,
'ip_version': ip_version},
{'opt_name': 'server-ip-address',
'opt_value': '192.0.2.1',
'ip_version': ip_version}]
self.assertItemsEqual(expected_info,
pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=True))
def test_dhcp_options_for_instance_ipxe_bios(self):
self.config(ip_version=4, group='pxe')
boot_file = 'fake-bootfile-bios'
self.config(pxe_bootfile_name=boot_file, group='pxe')
with task_manager.acquire(self.context, self.node.uuid) as task:
self._dhcp_options_for_instance_ipxe(task, boot_file)
def test_dhcp_options_for_instance_ipxe_uefi(self):
self.config(ip_version=4, group='pxe')
boot_file = 'fake-bootfile-uefi'
self.config(uefi_pxe_bootfile_name=boot_file, group='pxe')
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.properties['capabilities'] = 'boot_mode:uefi'
self._dhcp_options_for_instance_ipxe(task, boot_file)
def test_dhcp_options_for_ipxe_ipv6(self):
self.config(ip_version=6, group='pxe')
boot_file = 'fake-bootfile'
self.config(pxe_bootfile_name=boot_file, group='pxe')
with task_manager.acquire(self.context, self.node.uuid) as task:
self._dhcp_options_for_instance_ipxe(task, boot_file, ip_version=6)
@mock.patch('ironic.common.utils.rmtree_without_raise', autospec=True)
@mock.patch('ironic_lib.utils.unlink_without_raise', autospec=True)
@mock.patch('ironic.common.dhcp_factory.DHCPFactory.provider',
@ -924,29 +821,6 @@ class TestPXEUtils(db_base.DbTestCase):
rmtree_mock.assert_called_once_with(
os.path.join(CONF.pxe.tftp_root, self.node.uuid))
@mock.patch('ironic.common.utils.rmtree_without_raise', autospec=True)
@mock.patch('ironic_lib.utils.unlink_without_raise', autospec=True)
def test_clean_up_ipxe_config_uefi(self, unlink_mock, rmtree_mock):
self.config(ipxe_enabled=True, group='pxe')
address = "aa:aa:aa:aa:aa:aa"
properties = {'capabilities': 'boot_mode:uefi'}
object_utils.create_test_port(self.context, node_id=self.node.id,
address=address)
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.properties = properties
pxe_utils.clean_up_pxe_config(task, ipxe_enabled=True)
ensure_calls = [
mock.call("/httpboot/pxelinux.cfg/%s"
% address.replace(':', '-')),
mock.call("/httpboot/%s.conf" % address)
]
unlink_mock.assert_has_calls(ensure_calls)
rmtree_mock.assert_called_once_with(
os.path.join(CONF.deploy.http_root, self.node.uuid))
def test_get_tftp_path_prefix_with_trailing_slash(self):
self.config(tftp_root='/tftpboot-path/', group='pxe')
path_prefix = pxe_utils.get_tftp_path_prefix()
@ -1154,6 +1028,118 @@ class PXEInterfacesTestCase(db_base.DbTestCase):
image_info = pxe_utils.get_instance_image_info(task)
self.assertEqual({}, image_info)
@mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
def test__cache_tftp_images_master_path(self, mock_fetch_image):
temp_dir = tempfile.mkdtemp()
self.config(tftp_root=temp_dir, group='pxe')
self.config(tftp_master_path=os.path.join(temp_dir,
'tftp_master_path'),
group='pxe')
image_path = os.path.join(temp_dir, self.node.uuid,
'deploy_kernel')
image_info = {'deploy_kernel': ('deploy_kernel', image_path)}
fileutils.ensure_tree(CONF.pxe.tftp_master_path)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
pxe_utils.cache_ramdisk_kernel(task, image_info)
mock_fetch_image.assert_called_once_with(self.context,
mock.ANY,
[('deploy_kernel',
image_path)],
True)
@mock.patch.object(pxe_utils, 'TFTPImageCache', lambda: None)
@mock.patch.object(fileutils, 'ensure_tree', autospec=True)
@mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
def test_cache_ramdisk_kernel(self, mock_fetch_image, mock_ensure_tree):
fake_pxe_info = {'foo': 'bar'}
expected_path = os.path.join(CONF.pxe.tftp_root, self.node.uuid)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
pxe_utils.cache_ramdisk_kernel(task, fake_pxe_info)
mock_ensure_tree.assert_called_with(expected_path)
mock_fetch_image.assert_called_once_with(
self.context, mock.ANY, list(fake_pxe_info.values()), True)
@mock.patch.object(pxe_utils, 'TFTPImageCache', lambda: None)
@mock.patch.object(fileutils, 'ensure_tree', autospec=True)
@mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
def test_cache_ramdisk_kernel_ipxe(self, mock_fetch_image,
mock_ensure_tree):
fake_pxe_info = {'foo': 'bar'}
expected_path = os.path.join(CONF.deploy.http_root,
self.node.uuid)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
pxe_utils.cache_ramdisk_kernel(task, fake_pxe_info,
ipxe_enabled=True)
mock_ensure_tree.assert_called_with(expected_path)
mock_fetch_image.assert_called_once_with(self.context, mock.ANY,
list(fake_pxe_info.values()),
True)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_one(self, mock_log):
properties = {'capabilities': 'boot_mode:uefi'}
instance_info = {"boot_option": "netboot"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = False
self.assertRaises(exception.InvalidParameterValue,
pxe_utils.validate_boot_parameters_for_trusted_boot,
self.node)
self.assertTrue(mock_log.called)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_two(self, mock_log):
properties = {'capabilities': 'boot_mode:bios'}
instance_info = {"boot_option": "local"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = False
self.assertRaises(exception.InvalidParameterValue,
pxe_utils.validate_boot_parameters_for_trusted_boot,
self.node)
self.assertTrue(mock_log.called)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_three(self, mock_log):
properties = {'capabilities': 'boot_mode:bios'}
instance_info = {"boot_option": "netboot"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = True
self.assertRaises(exception.InvalidParameterValue,
pxe_utils.validate_boot_parameters_for_trusted_boot,
self.node)
self.assertTrue(mock_log.called)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_pass(self, mock_log):
properties = {'capabilities': 'boot_mode:bios'}
instance_info = {"boot_option": "netboot"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = False
pxe_utils.validate_boot_parameters_for_trusted_boot(self.node)
self.assertFalse(mock_log.called)
@mock.patch.object(pxe.PXEBoot, '__init__', lambda self: None)
class PXEBuildConfigOptionsTestCase(db_base.DbTestCase):
def setUp(self):
super(PXEBuildConfigOptionsTestCase, self).setUp()
n = {
'driver': 'fake-hardware',
'boot_interface': 'pxe',
'instance_info': INST_INFO_DICT,
'driver_info': DRV_INFO_DICT,
'driver_internal_info': DRV_INTERNAL_INFO_DICT,
}
self.config_temp_dir('http_root', group='deploy')
self.node = object_utils.create_test_node(self.context, **n)
@mock.patch('ironic.common.utils.render_template', autospec=True)
def _test_build_pxe_config_options_pxe(self, render_mock,
whle_dsk_img=False,
@ -1294,6 +1280,122 @@ class PXEInterfacesTestCase(db_base.DbTestCase):
'ipxe_timeout': 0}
self.assertEqual(expected_options, options)
@mock.patch.object(ipxe.iPXEBoot, '__init__', lambda self: None)
class iPXEBuildConfigOptionsTestCase(db_base.DbTestCase):
def setUp(self):
super(iPXEBuildConfigOptionsTestCase, self).setUp()
n = {
'driver': 'fake-hardware',
'boot_interface': 'ipxe',
'instance_info': INST_INFO_DICT,
'driver_info': DRV_INFO_DICT,
'driver_internal_info': DRV_INTERNAL_INFO_DICT,
}
self.config(enabled_boot_interfaces=['ipxe'])
self.config_temp_dir('http_root', group='deploy')
self.node = object_utils.create_test_node(self.context, **n)
def _dhcp_options_for_instance_ipxe(self, task, boot_file, ip_version=4):
self.config(ipxe_boot_script='/test/boot.ipxe', group='pxe')
self.config(tftp_root='/tftp-path/', group='pxe')
if ip_version == 4:
self.config(tftp_server='192.0.2.1', group='pxe')
self.config(http_url='http://192.0.3.2:1234', group='deploy')
self.config(ipxe_boot_script='/test/boot.ipxe', group='pxe')
elif ip_version == 6:
self.config(tftp_server='ff80::1', group='pxe')
self.config(http_url='http://[ff80::1]:1234', group='deploy')
self.config(dhcp_provider='isc', group='dhcp')
if ip_version == 6:
# NOTE(TheJulia): DHCPv6 RFCs seem to indicate that the prior
# options are not imported, although they may be supported
# by vendors. The apparent proper option is to return a
# URL in the field https://tools.ietf.org/html/rfc5970#section-3
expected_boot_script_url = 'http://[ff80::1]:1234/boot.ipxe'
expected_info = [{'opt_name': '!175,59',
'opt_value': 'tftp://[ff80::1]/fake-bootfile',
'ip_version': ip_version},
{'opt_name': '59',
'opt_value': expected_boot_script_url,
'ip_version': ip_version}]
elif ip_version == 4:
expected_boot_script_url = 'http://192.0.3.2:1234/boot.ipxe'
expected_info = [{'opt_name': '!175,67',
'opt_value': boot_file,
'ip_version': ip_version},
{'opt_name': '66',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': '150',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': '67',
'opt_value': expected_boot_script_url,
'ip_version': ip_version},
{'opt_name': 'server-ip-address',
'opt_value': '192.0.2.1',
'ip_version': ip_version}]
self.assertItemsEqual(expected_info,
pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=True))
self.config(dhcp_provider='neutron', group='dhcp')
if ip_version == 6:
# Boot URL variable set from prior test of isc parameters.
expected_info = [{'opt_name': 'tag:!ipxe6,59',
'opt_value': 'tftp://[ff80::1]/fake-bootfile',
'ip_version': ip_version},
{'opt_name': 'tag:ipxe6,59',
'opt_value': expected_boot_script_url,
'ip_version': ip_version}]
elif ip_version == 4:
expected_info = [{'opt_name': 'tag:!ipxe,67',
'opt_value': boot_file,
'ip_version': ip_version},
{'opt_name': '66',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': '150',
'opt_value': '192.0.2.1',
'ip_version': ip_version},
{'opt_name': 'tag:ipxe,67',
'opt_value': expected_boot_script_url,
'ip_version': ip_version},
{'opt_name': 'server-ip-address',
'opt_value': '192.0.2.1',
'ip_version': ip_version}]
self.assertItemsEqual(expected_info,
pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=True))
def test_dhcp_options_for_instance_ipxe_bios(self):
self.config(ip_version=4, group='pxe')
boot_file = 'fake-bootfile-bios'
self.config(pxe_bootfile_name=boot_file, group='pxe')
with task_manager.acquire(self.context, self.node.uuid) as task:
self._dhcp_options_for_instance_ipxe(task, boot_file)
def test_dhcp_options_for_instance_ipxe_uefi(self):
self.config(ip_version=4, group='pxe')
boot_file = 'fake-bootfile-uefi'
self.config(uefi_pxe_bootfile_name=boot_file, group='pxe')
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.properties['capabilities'] = 'boot_mode:uefi'
self._dhcp_options_for_instance_ipxe(task, boot_file)
def test_dhcp_options_for_ipxe_ipv6(self):
self.config(ip_version=6, group='pxe')
boot_file = 'fake-bootfile'
self.config(pxe_bootfile_name=boot_file, group='pxe')
with task_manager.acquire(self.context, self.node.uuid) as task:
self._dhcp_options_for_instance_ipxe(task, boot_file, ip_version=6)
@mock.patch('ironic.common.image_service.GlanceImageService',
autospec=True)
@mock.patch('ironic.common.utils.render_template', autospec=True)
@ -1319,7 +1421,6 @@ class PXEInterfacesTestCase(db_base.DbTestCase):
tftp_server = CONF.pxe.tftp_server
http_url = 'http://192.1.2.3:1234'
self.config(ipxe_enabled=True, group='pxe')
self.config(http_url=http_url, group='deploy')
kernel_label = '%s_kernel' % mode
@ -1601,103 +1702,28 @@ class PXEInterfacesTestCase(db_base.DbTestCase):
self._test_build_pxe_config_options_ipxe(mode='rescue',
ipxe_timeout=120)
@mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
def test__cache_tftp_images_master_path(self, mock_fetch_image):
temp_dir = tempfile.mkdtemp()
self.config(tftp_root=temp_dir, group='pxe')
self.config(tftp_master_path=os.path.join(temp_dir,
'tftp_master_path'),
group='pxe')
image_path = os.path.join(temp_dir, self.node.uuid,
'deploy_kernel')
image_info = {'deploy_kernel': ('deploy_kernel', image_path)}
fileutils.ensure_tree(CONF.pxe.tftp_master_path)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
pxe_utils.cache_ramdisk_kernel(task, image_info)
mock_fetch_image.assert_called_once_with(self.context,
mock.ANY,
[('deploy_kernel',
image_path)],
True)
@mock.patch.object(pxe_utils, 'TFTPImageCache', lambda: None)
@mock.patch.object(fileutils, 'ensure_tree', autospec=True)
@mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
def test_cache_ramdisk_kernel(self, mock_fetch_image, mock_ensure_tree):
self.config(ipxe_enabled=False, group='pxe')
fake_pxe_info = {'foo': 'bar'}
expected_path = os.path.join(CONF.pxe.tftp_root, self.node.uuid)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
pxe_utils.cache_ramdisk_kernel(task, fake_pxe_info)
mock_ensure_tree.assert_called_with(expected_path)
mock_fetch_image.assert_called_once_with(
self.context, mock.ANY, list(fake_pxe_info.values()), True)
@mock.patch.object(pxe_utils, 'TFTPImageCache', lambda: None)
@mock.patch.object(fileutils, 'ensure_tree', autospec=True)
@mock.patch.object(deploy_utils, 'fetch_images', autospec=True)
def test_cache_ramdisk_kernel_ipxe(self, mock_fetch_image,
mock_ensure_tree):
fake_pxe_info = {'foo': 'bar'}
expected_path = os.path.join(CONF.deploy.http_root,
self.node.uuid)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
pxe_utils.cache_ramdisk_kernel(task, fake_pxe_info,
ipxe_enabled=True)
mock_ensure_tree.assert_called_with(expected_path)
mock_fetch_image.assert_called_once_with(self.context, mock.ANY,
list(fake_pxe_info.values()),
True)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_one(self, mock_log):
@mock.patch('ironic.common.utils.rmtree_without_raise', autospec=True)
@mock.patch('ironic_lib.utils.unlink_without_raise', autospec=True)
def test_clean_up_ipxe_config_uefi(self, unlink_mock, rmtree_mock):
self.config(http_root='/httpboot', group='deploy')
address = "aa:aa:aa:aa:aa:aa"
properties = {'capabilities': 'boot_mode:uefi'}
instance_info = {"boot_option": "netboot"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = False
self.assertRaises(exception.InvalidParameterValue,
pxe.validate_boot_parameters_for_trusted_boot,
self.node)
self.assertTrue(mock_log.called)
object_utils.create_test_port(self.context, node_id=self.node.id,
address=address)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_two(self, mock_log):
properties = {'capabilities': 'boot_mode:bios'}
instance_info = {"boot_option": "local"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = False
self.assertRaises(exception.InvalidParameterValue,
pxe.validate_boot_parameters_for_trusted_boot,
self.node)
self.assertTrue(mock_log.called)
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.properties = properties
pxe_utils.clean_up_pxe_config(task, ipxe_enabled=True)
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_three(self, mock_log):
properties = {'capabilities': 'boot_mode:bios'}
instance_info = {"boot_option": "netboot"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = True
self.assertRaises(exception.InvalidParameterValue,
pxe.validate_boot_parameters_for_trusted_boot,
self.node)
self.assertTrue(mock_log.called)
ensure_calls = [
mock.call("/httpboot/pxelinux.cfg/%s"
% address.replace(':', '-')),
mock.call("/httpboot/%s.conf" % address)
]
@mock.patch.object(pxe_utils.LOG, 'error', autospec=True)
def test_validate_boot_parameters_for_trusted_boot_pass(self, mock_log):
properties = {'capabilities': 'boot_mode:bios'}
instance_info = {"boot_option": "netboot"}
self.node.properties = properties
self.node.instance_info['capabilities'] = instance_info
self.node.driver_internal_info['is_whole_disk_image'] = False
pxe.validate_boot_parameters_for_trusted_boot(self.node)
self.assertFalse(mock_log.called)
unlink_mock.assert_has_calls(ensure_calls)
rmtree_mock.assert_called_once_with(
os.path.join(CONF.deploy.http_root, self.node.uuid))
@mock.patch.object(ironic_utils, 'unlink_without_raise', autospec=True)

3
ironic/tests/unit/drivers/modules/irmc/test_boot.py
View File

@ -40,6 +40,7 @@ from ironic.drivers.modules.irmc import boot as irmc_boot
from ironic.drivers.modules.irmc import common as irmc_common
from ironic.drivers.modules.irmc import management as irmc_management
from ironic.drivers.modules import pxe
from ironic.drivers.modules import pxe_base
from ironic.tests import base
from ironic.tests.unit.db import utils as db_utils
from ironic.tests.unit.drivers.modules.irmc import test_common
@ -1889,7 +1890,7 @@ class IRMCPXEBootBasicTestCase(test_pxe.PXEBootTestCase):
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
properties = task.driver.get_properties()
for p in pxe.COMMON_PROPERTIES:
for p in pxe_base.COMMON_PROPERTIES:
self.assertIn(p, properties)

12
ironic/tests/unit/drivers/modules/storage/test_external.py
View File

@ -25,17 +25,15 @@ class ExternalInterfaceTestCase(db_base.DbTestCase):
def setUp(self):
super(ExternalInterfaceTestCase, self).setUp()
self.config(ipxe_enabled=True,
group='pxe')
self.config(enabled_storage_interfaces=['noop', 'external'])
self.config(enabled_storage_interfaces=['noop', 'external'],
enabled_boot_interfaces=['fake', 'pxe'])
self.interface = external.ExternalStorage()
@mock.patch.object(external, 'LOG', autospec=True)
def test_validate_fails_with_ipxe_not_enabled(self, mock_log):
"""Ensure a validation failure is raised when iPXE not enabled."""
self.config(ipxe_enabled=False, group='pxe')
self.node = object_utils.create_test_node(
self.context, storage_interface='external')
self.context, storage_interface='external', boot_interface='pxe')
object_utils.create_test_volume_connector(
self.context, node_id=self.node.id, type='iqn',
connector_id='foo.address')
@ -48,8 +46,8 @@ class ExternalInterfaceTestCase(db_base.DbTestCase):
task)
self.assertTrue(mock_log.error.called)
# Prevent /httpboot validation on creating the node
@mock.patch('ironic.drivers.modules.pxe.PXEBoot.__init__',
# Prevents creating iPXE boot script
@mock.patch('ironic.drivers.modules.ipxe.iPXEBoot.__init__',
lambda self: None)
def test_should_write_image(self):
self.node = object_utils.create_test_node(

8
ironic/tests/unit/drivers/modules/test_ipxe.py
View File

@ -86,7 +86,7 @@ class iPXEBootTestCase(db_base.DbTestCase):
self.config(group='conductor', api_url='http://127.0.0.1:1234/')
def test_get_properties(self):
expected = ipxe.COMMON_PROPERTIES
expected = pxe_base.COMMON_PROPERTIES
expected.update(agent_base.VENDOR_PROPERTIES)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
@ -415,7 +415,6 @@ class iPXEBootTestCase(db_base.DbTestCase):
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(
@ -435,7 +434,6 @@ class iPXEBootTestCase(db_base.DbTestCase):
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)
@ -456,7 +454,6 @@ class iPXEBootTestCase(db_base.DbTestCase):
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)
@ -465,7 +462,6 @@ class iPXEBootTestCase(db_base.DbTestCase):
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(
@ -480,7 +476,6 @@ class iPXEBootTestCase(db_base.DbTestCase):
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(
@ -774,7 +769,6 @@ class iPXEBootTestCase(db_base.DbTestCase):
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

14
ironic/tests/unit/drivers/modules/test_pxe.py
View File

@ -88,7 +88,7 @@ class PXEBootTestCase(db_base.DbTestCase):
self.config(group='conductor', api_url='http://127.0.0.1:1234/')
def test_get_properties(self):
expected = pxe.COMMON_PROPERTIES
expected = pxe_base.COMMON_PROPERTIES
expected.update(agent_base.VENDOR_PROPERTIES)
with task_manager.acquire(self.context, self.node.uuid,
shared=True) as task:
@ -266,7 +266,7 @@ class PXEBootTestCase(db_base.DbTestCase):
self.node.save()
with task_manager.acquire(self.context, self.node.uuid) as task:
dhcp_opts = pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=CONF.pxe.ipxe_enabled)
task, ipxe_enabled=False)
task.driver.boot.prepare_ramdisk(task, {'foo': 'bar'})
mock_deploy_img_info.assert_called_once_with(task.node,
mode=mode,
@ -594,7 +594,7 @@ class PXEBootTestCase(db_base.DbTestCase):
self.node.save()
with task_manager.acquire(self.context, self.node.uuid) as task:
dhcp_opts = pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=CONF.pxe.ipxe_enabled)
task, ipxe_enabled=False)
pxe_config_path = pxe_utils.get_pxe_config_file_path(
task.node.uuid)
task.node.properties['capabilities'] = 'boot_mode:bios'
@ -688,7 +688,7 @@ class PXEBootTestCase(db_base.DbTestCase):
task.node.save()
task.driver.boot.prepare_instance(task)
clean_up_pxe_config_mock.assert_called_once_with(
task, ipxe_enabled=CONF.pxe.ipxe_enabled)
task, ipxe_enabled=False)
set_boot_device_mock.assert_called_once_with(task,
boot_devices.DISK,
persistent=True)
@ -706,7 +706,7 @@ class PXEBootTestCase(db_base.DbTestCase):
task.node.save()
task.driver.boot.prepare_instance(task)
clean_up_pxe_config_mock.assert_called_once_with(
task, ipxe_enabled=CONF.pxe.ipxe_enabled)
task, ipxe_enabled=False)
self.assertFalse(set_boot_device_mock.called)
@mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True)
@ -733,7 +733,7 @@ class PXEBootTestCase(db_base.DbTestCase):
task.node.instance_info = instance_info
task.node.save()
dhcp_opts = pxe_utils.dhcp_options_for_instance(
task, ipxe_enabled=CONF.pxe.ipxe_enabled)
task, ipxe_enabled=False)
pxe_config_path = pxe_utils.get_pxe_config_file_path(
task.node.uuid)
task.driver.boot.prepare_instance(task)
@ -741,7 +741,7 @@ class PXEBootTestCase(db_base.DbTestCase):
get_image_info_mock.assert_called_once_with(task,
ipxe_enabled=False)
cache_mock.assert_called_once_with(
task, image_info, CONF.pxe.ipxe_enabled)
task, image_info, False)
provider_mock.update_dhcp.assert_called_once_with(task, dhcp_opts)
if config_file_exits:
self.assertFalse(create_pxe_config_mock.called)