# Copyright (c) 2012 NTT DOCOMO, INC. # Copyright 2011 OpenStack Foundation # Copyright 2011 Ilya Alekseyev # # 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. import os import tempfile from unittest import mock import fixtures from oslo_config import cfg from oslo_utils import fileutils from oslo_utils import uuidutils from ironic.common import boot_devices from ironic.common import exception from ironic.common import faults from ironic.common import image_service from ironic.common import states from ironic.common import utils as common_utils from ironic.conductor import task_manager from ironic.conductor import utils as manager_utils from ironic.drivers.modules import deploy_utils as utils from ironic.drivers.modules import fake from ironic.drivers.modules import image_cache from ironic.drivers.modules import pxe from ironic.drivers.modules.storage import cinder from ironic.drivers import utils as driver_utils from ironic.tests import base as tests_base 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 INST_INFO_DICT = db_utils.get_test_pxe_instance_info() DRV_INFO_DICT = db_utils.get_test_pxe_driver_info() DRV_INTERNAL_INFO_DICT = db_utils.get_test_pxe_driver_internal_info() _PXECONF_DEPLOY = b""" default deploy label deploy kernel deploy_kernel append initrd=deploy_ramdisk ipappend 3 label boot_partition kernel kernel append initrd=ramdisk root={{ ROOT }} label boot_whole_disk COM32 chain.c32 append mbr:{{ DISK_IDENTIFIER }} label trusted_boot kernel mboot append tboot.gz --- kernel root={{ ROOT }} --- ramdisk """ _PXECONF_BOOT_PARTITION = """ default boot_partition label deploy kernel deploy_kernel append initrd=deploy_ramdisk ipappend 3 label boot_partition kernel kernel append initrd=ramdisk root=UUID=12345678-1234-1234-1234-1234567890abcdef label boot_whole_disk COM32 chain.c32 append mbr:{{ DISK_IDENTIFIER }} label trusted_boot kernel mboot append tboot.gz --- kernel root=UUID=12345678-1234-1234-1234-1234567890abcdef \ --- ramdisk """ _PXECONF_BOOT_WHOLE_DISK = """ default boot_whole_disk label deploy kernel deploy_kernel append initrd=deploy_ramdisk ipappend 3 label boot_partition kernel kernel append initrd=ramdisk root={{ ROOT }} label boot_whole_disk COM32 chain.c32 append mbr:0x12345678 label trusted_boot kernel mboot append tboot.gz --- kernel root={{ ROOT }} --- ramdisk """ _PXECONF_TRUSTED_BOOT = """ default trusted_boot label deploy kernel deploy_kernel append initrd=deploy_ramdisk ipappend 3 label boot_partition kernel kernel append initrd=ramdisk root=UUID=12345678-1234-1234-1234-1234567890abcdef label boot_whole_disk COM32 chain.c32 append mbr:{{ DISK_IDENTIFIER }} label trusted_boot kernel mboot append tboot.gz --- kernel root=UUID=12345678-1234-1234-1234-1234567890abcdef \ --- ramdisk """ _IPXECONF_DEPLOY = b""" #!ipxe dhcp goto deploy :deploy kernel deploy_kernel initrd deploy_ramdisk boot :boot_partition kernel kernel append initrd=ramdisk root={{ ROOT }} boot :boot_whole_disk kernel chain.c32 append mbr:{{ DISK_IDENTIFIER }} boot """ _IPXECONF_BOOT_PARTITION = """ #!ipxe dhcp goto boot_partition :deploy kernel deploy_kernel initrd deploy_ramdisk boot :boot_partition kernel kernel append initrd=ramdisk root=UUID=12345678-1234-1234-1234-1234567890abcdef boot :boot_whole_disk kernel chain.c32 append mbr:{{ DISK_IDENTIFIER }} boot """ _IPXECONF_BOOT_WHOLE_DISK = """ #!ipxe dhcp goto boot_whole_disk :deploy kernel deploy_kernel initrd deploy_ramdisk boot :boot_partition kernel kernel append initrd=ramdisk root={{ ROOT }} boot :boot_whole_disk kernel chain.c32 append mbr:0x12345678 boot """ _IPXECONF_BOOT_ISCSI_NO_CONFIG = """ #!ipxe dhcp goto boot_iscsi :deploy kernel deploy_kernel initrd deploy_ramdisk boot :boot_partition kernel kernel append initrd=ramdisk root=UUID=0x12345678 boot :boot_whole_disk kernel chain.c32 append mbr:{{ DISK_IDENTIFIER }} boot """ _UEFI_PXECONF_DEPLOY = b""" default=deploy image=deploy_kernel label=deploy initrd=deploy_ramdisk append="ro text" image=kernel label=boot_partition initrd=ramdisk append="root={{ ROOT }}" image=chain.c32 label=boot_whole_disk append="mbr:{{ DISK_IDENTIFIER }}" """ _UEFI_PXECONF_BOOT_PARTITION = """ default=boot_partition image=deploy_kernel label=deploy initrd=deploy_ramdisk append="ro text" image=kernel label=boot_partition initrd=ramdisk append="root=UUID=12345678-1234-1234-1234-1234567890abcdef" image=chain.c32 label=boot_whole_disk append="mbr:{{ DISK_IDENTIFIER }}" """ _UEFI_PXECONF_BOOT_WHOLE_DISK = """ default=boot_whole_disk image=deploy_kernel label=deploy initrd=deploy_ramdisk append="ro text" image=kernel label=boot_partition initrd=ramdisk append="root={{ ROOT }}" image=chain.c32 label=boot_whole_disk append="mbr:0x12345678" """ _UEFI_PXECONF_DEPLOY_GRUB = b""" set default=deploy set timeout=5 set hidden_timeout_quiet=false menuentry "deploy" { linuxefi deploy_kernel "ro text" initrdefi deploy_ramdisk } menuentry "boot_partition" { linuxefi kernel "root=(( ROOT ))" initrdefi ramdisk } menuentry "boot_whole_disk" { linuxefi chain.c32 mbr:(( DISK_IDENTIFIER )) } """ _UEFI_PXECONF_BOOT_PARTITION_GRUB = """ set default=boot_partition set timeout=5 set hidden_timeout_quiet=false menuentry "deploy" { linuxefi deploy_kernel "ro text" initrdefi deploy_ramdisk } menuentry "boot_partition" { linuxefi kernel "root=UUID=12345678-1234-1234-1234-1234567890abcdef" initrdefi ramdisk } menuentry "boot_whole_disk" { linuxefi chain.c32 mbr:(( DISK_IDENTIFIER )) } """ _UEFI_PXECONF_BOOT_WHOLE_DISK_GRUB = """ set default=boot_whole_disk set timeout=5 set hidden_timeout_quiet=false menuentry "deploy" { linuxefi deploy_kernel "ro text" initrdefi deploy_ramdisk } menuentry "boot_partition" { linuxefi kernel "root=(( ROOT ))" initrdefi ramdisk } menuentry "boot_whole_disk" { linuxefi chain.c32 mbr:0x12345678 } """ class SwitchPxeConfigTestCase(tests_base.TestCase): # NOTE(TheJulia): Remove elilo support after the deprecation period, # in the Queens release. def _create_config(self, ipxe=False, boot_mode=None, boot_loader='elilo'): (fd, fname) = tempfile.mkstemp() if boot_mode == 'uefi' and not ipxe: if boot_loader == 'grub': pxe_cfg = _UEFI_PXECONF_DEPLOY_GRUB else: pxe_cfg = _UEFI_PXECONF_DEPLOY else: pxe_cfg = _IPXECONF_DEPLOY if ipxe else _PXECONF_DEPLOY os.write(fd, pxe_cfg) os.close(fd) self.addCleanup(os.unlink, fname) return fname def test_switch_pxe_config_partition_image(self): boot_mode = 'bios' fname = self._create_config() utils.switch_pxe_config(fname, '12345678-1234-1234-1234-1234567890abcdef', boot_mode, False) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_PXECONF_BOOT_PARTITION, pxeconf) def test_switch_pxe_config_whole_disk_image(self): boot_mode = 'bios' fname = self._create_config() utils.switch_pxe_config(fname, '0x12345678', boot_mode, True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_PXECONF_BOOT_WHOLE_DISK, pxeconf) def test_switch_pxe_config_trusted_boot(self): boot_mode = 'bios' fname = self._create_config() utils.switch_pxe_config(fname, '12345678-1234-1234-1234-1234567890abcdef', boot_mode, False, True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_PXECONF_TRUSTED_BOOT, pxeconf) def test_switch_ipxe_config_partition_image(self): boot_mode = 'bios' fname = self._create_config(ipxe=True) utils.switch_pxe_config(fname, '12345678-1234-1234-1234-1234567890abcdef', boot_mode, False, ipxe_enabled=True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_IPXECONF_BOOT_PARTITION, pxeconf) def test_switch_ipxe_config_whole_disk_image(self): boot_mode = 'bios' fname = self._create_config(ipxe=True) utils.switch_pxe_config(fname, '0x12345678', boot_mode, True, ipxe_enabled=True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_IPXECONF_BOOT_WHOLE_DISK, pxeconf) # NOTE(TheJulia): Remove elilo support after the deprecation period, # in the Queens release. def test_switch_uefi_elilo_pxe_config_partition_image(self): boot_mode = 'uefi' fname = self._create_config(boot_mode=boot_mode) utils.switch_pxe_config(fname, '12345678-1234-1234-1234-1234567890abcdef', boot_mode, False) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_UEFI_PXECONF_BOOT_PARTITION, pxeconf) # NOTE(TheJulia): Remove elilo support after the deprecation period, # in the Queens release. def test_switch_uefi_elilo_config_whole_disk_image(self): boot_mode = 'uefi' fname = self._create_config(boot_mode=boot_mode) utils.switch_pxe_config(fname, '0x12345678', boot_mode, True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_UEFI_PXECONF_BOOT_WHOLE_DISK, pxeconf) def test_switch_uefi_grub_pxe_config_partition_image(self): boot_mode = 'uefi' fname = self._create_config(boot_mode=boot_mode, boot_loader='grub') utils.switch_pxe_config(fname, '12345678-1234-1234-1234-1234567890abcdef', boot_mode, False) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_UEFI_PXECONF_BOOT_PARTITION_GRUB, pxeconf) def test_switch_uefi_grub_config_whole_disk_image(self): boot_mode = 'uefi' fname = self._create_config(boot_mode=boot_mode, boot_loader='grub') utils.switch_pxe_config(fname, '0x12345678', boot_mode, True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_UEFI_PXECONF_BOOT_WHOLE_DISK_GRUB, pxeconf) def test_switch_uefi_ipxe_config_partition_image(self): boot_mode = 'uefi' fname = self._create_config(boot_mode=boot_mode, ipxe=True) utils.switch_pxe_config(fname, '12345678-1234-1234-1234-1234567890abcdef', boot_mode, False, ipxe_enabled=True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_IPXECONF_BOOT_PARTITION, pxeconf) def test_switch_uefi_ipxe_config_whole_disk_image(self): boot_mode = 'uefi' fname = self._create_config(boot_mode=boot_mode, ipxe=True) utils.switch_pxe_config(fname, '0x12345678', boot_mode, True, ipxe_enabled=True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_IPXECONF_BOOT_WHOLE_DISK, pxeconf) def test_switch_ipxe_iscsi_boot(self): boot_mode = 'iscsi' fname = self._create_config(boot_mode=boot_mode, ipxe=True) utils.switch_pxe_config(fname, '0x12345678', boot_mode, False, False, True, ipxe_enabled=True) with open(fname, 'r') as f: pxeconf = f.read() self.assertEqual(_IPXECONF_BOOT_ISCSI_NO_CONFIG, pxeconf) class GetPxeBootConfigTestCase(db_base.DbTestCase): def setUp(self): super(GetPxeBootConfigTestCase, self).setUp() self.node = obj_utils.get_test_node(self.context, driver='fake-hardware') self.config(pxe_bootfile_name='bios-bootfile', group='pxe') self.config(uefi_pxe_bootfile_name='uefi-bootfile', group='pxe') self.config(pxe_config_template='bios-template', group='pxe') self.config(uefi_pxe_config_template='uefi-template', group='pxe') self.bootfile_by_arch = {'aarch64': 'aarch64-bootfile', 'ppc64': 'ppc64-bootfile'} self.template_by_arch = {'aarch64': 'aarch64-template', 'ppc64': 'ppc64-template'} def test_get_pxe_boot_file_bios_without_by_arch(self): properties = {'cpu_arch': 'x86', 'capabilities': 'boot_mode:bios'} self.node.properties = properties self.config(pxe_bootfile_name_by_arch={}, group='pxe') result = utils.get_pxe_boot_file(self.node) self.assertEqual('bios-bootfile', result) def test_get_pxe_config_template_bios_without_by_arch(self): properties = {'cpu_arch': 'x86', 'capabilities': 'boot_mode:bios'} self.node.properties = properties self.config(pxe_config_template_by_arch={}, group='pxe') result = utils.get_pxe_config_template(self.node) self.assertEqual('bios-template', result) def test_get_pxe_boot_file_uefi_without_by_arch(self): properties = {'cpu_arch': 'x86_64', 'capabilities': 'boot_mode:uefi'} self.node.properties = properties self.config(pxe_bootfile_name_by_arch={}, group='pxe') result = utils.get_pxe_boot_file(self.node) self.assertEqual('uefi-bootfile', result) def test_get_pxe_config_template_uefi_without_by_arch(self): properties = {'cpu_arch': 'x86_64', 'capabilities': 'boot_mode:uefi'} self.node.properties = properties self.config(pxe_config_template_by_arch={}, group='pxe') result = utils.get_pxe_config_template(self.node) self.assertEqual('uefi-template', result) def test_get_pxe_boot_file_cpu_not_in_by_arch(self): properties = {'cpu_arch': 'x86', 'capabilities': 'boot_mode:bios'} self.node.properties = properties self.config(pxe_bootfile_name_by_arch=self.bootfile_by_arch, group='pxe') result = utils.get_pxe_boot_file(self.node) self.assertEqual('bios-bootfile', result) def test_get_pxe_config_template_cpu_not_in_by_arch(self): properties = {'cpu_arch': 'x86', 'capabilities': 'boot_mode:bios'} self.node.properties = properties self.config(pxe_config_template_by_arch=self.template_by_arch, group='pxe') result = utils.get_pxe_config_template(self.node) self.assertEqual('bios-template', result) def test_get_pxe_boot_file_cpu_in_by_arch(self): properties = {'cpu_arch': 'aarch64', 'capabilities': 'boot_mode:uefi'} self.node.properties = properties self.config(pxe_bootfile_name_by_arch=self.bootfile_by_arch, group='pxe') result = utils.get_pxe_boot_file(self.node) self.assertEqual('aarch64-bootfile', result) def test_get_pxe_config_template_cpu_in_by_arch(self): properties = {'cpu_arch': 'aarch64', 'capabilities': 'boot_mode:uefi'} self.node.properties = properties self.config(pxe_config_template_by_arch=self.template_by_arch, group='pxe') result = utils.get_pxe_config_template(self.node) self.assertEqual('aarch64-template', result) def test_get_pxe_boot_file_emtpy_property(self): self.node.properties = {} self.config(pxe_bootfile_name_by_arch=self.bootfile_by_arch, group='pxe') result = utils.get_pxe_boot_file(self.node) self.assertEqual('uefi-bootfile', result) def test_get_pxe_boot_file_emtpy_property_bios_default(self): self.config(default_boot_mode='bios', group='deploy') self.node.properties = {} self.config(pxe_bootfile_name_by_arch=self.bootfile_by_arch, group='pxe') result = utils.get_pxe_boot_file(self.node) self.assertEqual('bios-bootfile', result) def test_get_ipxe_boot_uefi(self): self.config(uefi_ipxe_bootfile_name='meow', group='pxe') result = utils.get_ipxe_boot_file(self.node) self.assertEqual('meow', result) def test_get_ipxe_boot_bios(self): self.config(default_boot_mode='bios', group='deploy') self.config(ipxe_bootfile_name='meow', group='pxe') result = utils.get_ipxe_boot_file(self.node) self.assertEqual('meow', result) def test_get_ipxe_boot_file_uefi(self): self.config(uefi_ipxe_bootfile_name='ipxe-uefi-bootfile', group='pxe') properties = {'capabilities': 'boot_mode:uefi'} self.node.properties = properties result = utils.get_ipxe_boot_file(self.node) self.assertEqual('ipxe-uefi-bootfile', result) def test_get_ipxe_boot_file_other_arch(self): arch_names = {'aarch64': 'ipxe-aa64.efi', 'x86_64': 'ipxe.kpxe'} self.config(ipxe_bootfile_name_by_arch=arch_names, group='pxe') properties = {'cpu_arch': 'aarch64', 'capabilities': 'boot_mode:uefi'} self.node.properties = properties result = utils.get_ipxe_boot_file(self.node) self.assertEqual('ipxe-aa64.efi', result) def test_get_ipxe_boot_file_fallback(self): self.config(ipxe_bootfile_name=None, group='pxe') self.config(uefi_ipxe_bootfile_name=None, group='pxe') self.config(pxe_bootfile_name=None, group='pxe') self.config(uefi_pxe_bootfile_name='lolcat', group='pxe') result = utils.get_ipxe_boot_file(self.node) self.assertEqual('lolcat', result) def test_get_ipxe_boot_file_fallback_bios(self): self.config(default_boot_mode='bios', group='deploy') self.config(ipxe_bootfile_name=None, group='pxe') self.config(uefi_ipxe_bootfile_name=None, group='pxe') self.config(pxe_bootfile_name='lolcat', group='pxe') result = utils.get_ipxe_boot_file(self.node) self.assertEqual('lolcat', result) def test_get_pxe_config_template_emtpy_property(self): self.node.properties = {} self.config(pxe_config_template_by_arch=self.template_by_arch, group='pxe') result = utils.get_pxe_config_template(self.node) self.assertEqual('uefi-template', result) def test_get_pxe_config_template_emtpy_property_bios(self): self.config(default_boot_mode='bios', group='deploy') self.node.properties = {} self.config(pxe_config_template_by_arch=self.template_by_arch, group='pxe') result = utils.get_pxe_config_template(self.node) self.assertEqual('bios-template', result) def test_get_pxe_config_template_per_node(self): node = obj_utils.create_test_node( self.context, driver='fake-hardware', driver_info={"pxe_template": "fake-template"}, ) result = utils.get_pxe_config_template(node) self.assertEqual('fake-template', result) def test_get_ipxe_config_template(self): node = obj_utils.create_test_node( self.context, driver='fake-hardware') self.assertIn('ipxe_config.template', utils.get_ipxe_config_template(node)) def test_get_ipxe_config_template_none(self): self.config(ipxe_config_template=None, group='pxe') self.config(uefi_pxe_config_template='magical_bootloader', group='pxe') node = obj_utils.create_test_node( self.context, driver='fake-hardware') self.assertEqual('magical_bootloader', utils.get_ipxe_config_template(node)) def test_get_ipxe_config_template_none_bios(self): self.config(default_boot_mode='bios', group='deploy') self.config(ipxe_config_template=None, group='pxe') self.config(pxe_config_template='magical_bootloader', group='pxe') node = obj_utils.create_test_node( self.context, driver='fake-hardware') self.assertEqual('magical_bootloader', utils.get_ipxe_config_template(node)) def test_get_ipxe_config_template_override_pxe_fallback(self): node = obj_utils.create_test_node( self.context, driver='fake-hardware', driver_info={'pxe_template': 'magical'}) self.assertEqual('magical', utils.get_ipxe_config_template(node)) @mock.patch('time.sleep', lambda sec: None) class OtherFunctionTestCase(db_base.DbTestCase): def setUp(self): super(OtherFunctionTestCase, self).setUp() self.node = obj_utils.create_test_node(self.context, boot_interface='pxe') @mock.patch.object(utils, 'LOG', autospec=True) @mock.patch.object(manager_utils, 'node_power_action', autospec=True) @mock.patch.object(manager_utils, 'deploying_error_handler', autospec=True) def _test_set_failed_state(self, mock_error, mock_power, mock_log, event_value=None, power_value=None, log_calls=None, poweroff=True, collect_logs=True): err_msg = 'some failure' mock_error.side_effect = event_value mock_power.side_effect = power_value with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: if collect_logs: utils.set_failed_state(task, err_msg) else: utils.set_failed_state(task, err_msg, collect_logs=collect_logs) mock_error.assert_called_once_with(task, err_msg, err_msg, clean_up=False) if poweroff: mock_power.assert_called_once_with(task, states.POWER_OFF) else: self.assertFalse(mock_power.called) self.assertEqual(err_msg, task.node.last_error) if (log_calls and poweroff): mock_log.exception.assert_has_calls(log_calls) else: self.assertFalse(mock_log.called) @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True) def test_set_failed_state(self, mock_collect): exc_state = exception.InvalidState('invalid state') exc_param = exception.InvalidParameterValue('invalid parameter') mock_call = mock.call(mock.ANY) self._test_set_failed_state() calls = [mock_call] self._test_set_failed_state(event_value=iter([exc_state] * len(calls)), log_calls=calls) calls = [mock_call] self._test_set_failed_state(power_value=iter([exc_param] * len(calls)), log_calls=calls) calls = [mock_call, mock_call] self._test_set_failed_state(event_value=iter([exc_state] * len(calls)), power_value=iter([exc_param] * len(calls)), log_calls=calls) self.assertEqual(4, mock_collect.call_count) @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True) def test_set_failed_state_no_poweroff(self, mock_collect): cfg.CONF.set_override('power_off_after_deploy_failure', False, 'deploy') exc_state = exception.InvalidState('invalid state') exc_param = exception.InvalidParameterValue('invalid parameter') mock_call = mock.call(mock.ANY) self._test_set_failed_state(poweroff=False) calls = [mock_call] self._test_set_failed_state(event_value=iter([exc_state] * len(calls)), log_calls=calls, poweroff=False) calls = [mock_call] self._test_set_failed_state(power_value=iter([exc_param] * len(calls)), log_calls=calls, poweroff=False) calls = [mock_call, mock_call] self._test_set_failed_state(event_value=iter([exc_state] * len(calls)), power_value=iter([exc_param] * len(calls)), log_calls=calls, poweroff=False) self.assertEqual(4, mock_collect.call_count) @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True) def test_set_failed_state_collect_deploy_logs(self, mock_collect): for opt in ('always', 'on_failure'): cfg.CONF.set_override('deploy_logs_collect', opt, 'agent') self._test_set_failed_state() mock_collect.assert_called_once_with(mock.ANY) mock_collect.reset_mock() @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True) def test_set_failed_state_collect_deploy_logs_never(self, mock_collect): cfg.CONF.set_override('deploy_logs_collect', 'never', 'agent') self._test_set_failed_state() self.assertFalse(mock_collect.called) @mock.patch.object(driver_utils, 'collect_ramdisk_logs', autospec=True) def test_set_failed_state_collect_deploy_logs_overide(self, mock_collect): cfg.CONF.set_override('deploy_logs_collect', 'always', 'agent') self._test_set_failed_state(collect_logs=False) self.assertFalse(mock_collect.called) def test_get_boot_option(self): self.node.instance_info = {'capabilities': '{"boot_option": "local"}'} result = utils.get_boot_option(self.node) self.assertEqual("local", result) def test_get_boot_option_default_value(self): self.node.instance_info = {} result = utils.get_boot_option(self.node) self.assertEqual("local", result) def test_get_boot_option_overridden_default_value(self): cfg.CONF.set_override('default_boot_option', 'local', 'deploy') self.node.instance_info = {} result = utils.get_boot_option(self.node) self.assertEqual("local", result) def test_get_boot_option_instance_info_priority(self): cfg.CONF.set_override('default_boot_option', 'local', 'deploy') self.node.instance_info = {'capabilities': '{"boot_option": "netboot"}'} result = utils.get_boot_option(self.node) self.assertEqual("netboot", result) @mock.patch.object(utils, 'is_software_raid', autospec=True) def test_get_boot_option_software_raid(self, mock_is_software_raid): mock_is_software_raid.return_value = True cfg.CONF.set_override('default_boot_option', 'netboot', 'deploy') result = utils.get_boot_option(self.node) self.assertEqual("local", result) @mock.patch.object(utils, 'is_anaconda_deploy', autospec=True) def test_get_boot_option_anaconda_deploy(self, mock_is_anaconda_deploy): mock_is_anaconda_deploy.return_value = True result = utils.get_boot_option(self.node) self.assertEqual("kickstart", result) def test_is_anaconda_deploy(self): self.node.deploy_interface = 'anaconda' result = utils.is_anaconda_deploy(self.node) self.assertTrue(result) def test_is_anaconda_deploy_false(self): result = utils.is_anaconda_deploy(self.node) self.assertFalse(result) def test_is_software_raid(self): self.node.target_raid_config = { "logical_disks": [ { "size_gb": 100, "raid_level": "1", "controller": "software", } ] } result = utils.is_software_raid(self.node) self.assertTrue(result) def test_is_software_raid_false(self): self.node.target_raid_config = {} result = utils.is_software_raid(self.node) self.assertFalse(result) @mock.patch.object(image_cache, 'clean_up_caches', autospec=True) def test_fetch_images(self, mock_clean_up_caches): mock_cache = mock.MagicMock( spec_set=['fetch_image', 'master_dir'], master_dir='master_dir') utils.fetch_images(None, mock_cache, [('uuid', 'path')]) mock_clean_up_caches.assert_called_once_with(None, 'master_dir', [('uuid', 'path')]) mock_cache.fetch_image.assert_called_once_with('uuid', 'path', ctx=None, force_raw=True) @mock.patch.object(image_cache, 'clean_up_caches', autospec=True) def test_fetch_images_fail(self, mock_clean_up_caches): exc = exception.InsufficientDiskSpace(path='a', required=2, actual=1) mock_cache = mock.MagicMock( spec_set=['master_dir'], master_dir='master_dir') mock_clean_up_caches.side_effect = [exc] self.assertRaises(exception.InstanceDeployFailure, utils.fetch_images, None, mock_cache, [('uuid', 'path')]) mock_clean_up_caches.assert_called_once_with(None, 'master_dir', [('uuid', 'path')]) @mock.patch('ironic.common.keystone.get_auth', autospec=True) @mock.patch.object(utils, '_get_ironic_session', autospec=True) @mock.patch('ironic.common.keystone.get_adapter', autospec=True) def test_get_ironic_api_url_from_keystone(self, mock_ka, mock_ks, mock_auth): mock_sess = mock.Mock() mock_ks.return_value = mock_sess fake_api_url = 'http://foo/' mock_ka.return_value.get_endpoint.return_value = fake_api_url # NOTE(pas-ha) endpoint_override is None by default url = utils.get_ironic_api_url() # also checking for stripped trailing slash self.assertEqual(fake_api_url[:-1], url) mock_ka.assert_called_with('service_catalog', session=mock_sess, auth=mock_auth.return_value) mock_ka.return_value.get_endpoint.assert_called_once_with() @mock.patch('ironic.common.keystone.get_auth', autospec=True) @mock.patch.object(utils, '_get_ironic_session', autospec=True) @mock.patch('ironic.common.keystone.get_adapter', autospec=True) def test_get_ironic_api_url_fail(self, mock_ka, mock_ks, mock_auth): mock_sess = mock.Mock() mock_ks.return_value = mock_sess mock_ka.return_value.get_endpoint.side_effect = ( exception.KeystoneFailure()) self.assertRaises(exception.InvalidParameterValue, utils.get_ironic_api_url) @mock.patch('ironic.common.keystone.get_auth', autospec=True) @mock.patch.object(utils, '_get_ironic_session', autospec=True) @mock.patch('ironic.common.keystone.get_adapter', autospec=True) def test_get_ironic_api_url_none(self, mock_ka, mock_ks, mock_auth): mock_sess = mock.Mock() mock_ks.return_value = mock_sess mock_ka.return_value.get_endpoint.return_value = None self.assertRaises(exception.InvalidParameterValue, utils.get_ironic_api_url) class GetSingleNicTestCase(db_base.DbTestCase): def setUp(self): super(GetSingleNicTestCase, self).setUp() self.node = obj_utils.create_test_node(self.context) def test_get_single_nic_with_vif_port_id(self): obj_utils.create_test_port( self.context, node_id=self.node.id, address='aa:bb:cc:dd:ee:ff', uuid=uuidutils.generate_uuid(), internal_info={'tenant_vif_port_id': 'test-vif-A'}) with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: address = utils.get_single_nic_with_vif_port_id(task) self.assertEqual('aa:bb:cc:dd:ee:ff', address) def test_get_single_nic_with_cleaning_vif_port_id(self): obj_utils.create_test_port( self.context, node_id=self.node.id, address='aa:bb:cc:dd:ee:ff', uuid=uuidutils.generate_uuid(), internal_info={'cleaning_vif_port_id': 'test-vif-A'}) with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: address = utils.get_single_nic_with_vif_port_id(task) self.assertEqual('aa:bb:cc:dd:ee:ff', address) def test_get_single_nic_with_provisioning_vif_port_id(self): obj_utils.create_test_port( self.context, node_id=self.node.id, address='aa:bb:cc:dd:ee:ff', uuid=uuidutils.generate_uuid(), internal_info={'provisioning_vif_port_id': 'test-vif-A'}) with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: address = utils.get_single_nic_with_vif_port_id(task) self.assertEqual('aa:bb:cc:dd:ee:ff', address) class ParseInstanceInfoCapabilitiesTestCase(tests_base.TestCase): def setUp(self): super(ParseInstanceInfoCapabilitiesTestCase, self).setUp() self.node = obj_utils.get_test_node(self.context, driver='fake-hardware') def test_parse_instance_info_capabilities_string(self): self.node.instance_info = {'capabilities': '{"cat": "meow"}'} expected_result = {"cat": "meow"} result = utils.parse_instance_info_capabilities(self.node) self.assertEqual(expected_result, result) def test_parse_instance_info_capabilities(self): self.node.instance_info = {'capabilities': {"dog": "wuff"}} expected_result = {"dog": "wuff"} result = utils.parse_instance_info_capabilities(self.node) self.assertEqual(expected_result, result) def test_parse_instance_info_invalid_type(self): self.node.instance_info = {'capabilities': 'not-a-dict'} self.assertRaises(exception.InvalidParameterValue, utils.parse_instance_info_capabilities, self.node) def test_is_secure_boot_requested_true(self): self.node.instance_info = {'capabilities': {"secure_boot": "tRue"}} self.assertTrue(utils.is_secure_boot_requested(self.node)) def test_is_secure_boot_requested_false(self): self.node.instance_info = {'capabilities': {"secure_boot": "false"}} self.assertFalse(utils.is_secure_boot_requested(self.node)) def test_is_secure_boot_requested_invalid(self): self.node.instance_info = {'capabilities': {"secure_boot": "invalid"}} self.assertFalse(utils.is_secure_boot_requested(self.node)) def test_is_trusted_boot_requested_true(self): self.node.instance_info = {'capabilities': {"trusted_boot": "true"}} self.assertTrue(utils.is_trusted_boot_requested(self.node)) def test_is_trusted_boot_requested_false(self): self.node.instance_info = {'capabilities': {"trusted_boot": "false"}} self.assertFalse(utils.is_trusted_boot_requested(self.node)) def test_is_trusted_boot_requested_invalid(self): self.node.instance_info = {'capabilities': {"trusted_boot": "invalid"}} self.assertFalse(utils.is_trusted_boot_requested(self.node)) def test_validate_boot_mode_capability(self): prop = {'capabilities': 'boot_mode:uefi,cap2:value2'} self.node.properties = prop result = utils.validate_capabilities(self.node) self.assertIsNone(result) def test_validate_boot_mode_capability_with_exc(self): prop = {'capabilities': 'boot_mode:UEFI,cap2:value2'} self.node.properties = prop self.assertRaises(exception.InvalidParameterValue, utils.validate_capabilities, self.node) def test_validate_boot_mode_capability_instance_info(self): inst_info = {'capabilities': {"boot_mode": "uefi", "cap2": "value2"}} self.node.instance_info = inst_info result = utils.validate_capabilities(self.node) self.assertIsNone(result) def test_validate_boot_mode_capability_instance_info_with_exc(self): inst_info = {'capabilities': {"boot_mode": "UEFI", "cap2": "value2"}} self.node.instance_info = inst_info self.assertRaises(exception.InvalidParameterValue, utils.validate_capabilities, self.node) def test_validate_trusted_boot_capability(self): properties = {'capabilities': 'trusted_boot:value'} self.node.properties = properties self.assertRaises(exception.InvalidParameterValue, utils.validate_capabilities, self.node) def test_all_supported_capabilities(self): self.assertEqual(('local', 'netboot', 'ramdisk', 'kickstart'), utils.SUPPORTED_CAPABILITIES['boot_option']) self.assertEqual(('bios', 'uefi'), utils.SUPPORTED_CAPABILITIES['boot_mode']) self.assertEqual(('true', 'false'), utils.SUPPORTED_CAPABILITIES['secure_boot']) self.assertEqual(('true', 'false'), utils.SUPPORTED_CAPABILITIES['trusted_boot']) def test_get_disk_label(self): inst_info = {'capabilities': {'disk_label': 'gpt', 'foo': 'bar'}} self.node.instance_info = inst_info result = utils.get_disk_label(self.node) self.assertEqual('gpt', result) def test_get_disk_label_nothing_set(self): inst_info = {'capabilities': {'cat': 'meows'}} self.node.instance_info = inst_info result = utils.get_disk_label(self.node) self.assertEqual('gpt', result) def test_get_disk_label_nothing_set_bios_mode(self): self.config(default_boot_mode='bios', group='deploy') inst_info = {'capabilities': {'cat': 'meows'}} self.node.instance_info = inst_info result = utils.get_disk_label(self.node) self.assertIsNone(result) def test_get_disk_label_uefi_mode(self): inst_info = {'capabilities': {'cat': 'meows'}} properties = {'capabilities': 'boot_mode:uefi'} self.node.instance_info = inst_info self.node.properties = properties result = utils.get_disk_label(self.node) self.assertEqual('gpt', result) class TrySetBootDeviceTestCase(db_base.DbTestCase): def setUp(self): super(TrySetBootDeviceTestCase, self).setUp() self.node = obj_utils.create_test_node(self.context, driver="fake-hardware") @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True) def test_try_set_boot_device_okay(self, node_set_boot_device_mock): with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: utils.try_set_boot_device(task, boot_devices.DISK, persistent=True) node_set_boot_device_mock.assert_called_once_with( task, boot_devices.DISK, persistent=True) @mock.patch.object(utils, 'LOG', autospec=True) @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True) def test_try_set_boot_device_ipmifailure_uefi( self, node_set_boot_device_mock, log_mock): self.node.properties = {'capabilities': 'boot_mode:uefi'} self.node.save() node_set_boot_device_mock.side_effect = exception.IPMIFailure(cmd='a') with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: utils.try_set_boot_device(task, boot_devices.DISK, persistent=True) node_set_boot_device_mock.assert_called_once_with( task, boot_devices.DISK, persistent=True) log_mock.warning.assert_called_once_with(mock.ANY, self.node.uuid) @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True) def test_try_set_boot_device_ipmifailure_bios( self, node_set_boot_device_mock): self.config(default_boot_mode='bios', group='deploy') node_set_boot_device_mock.side_effect = exception.IPMIFailure(cmd='a') with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: self.assertRaises(exception.IPMIFailure, utils.try_set_boot_device, task, boot_devices.DISK, persistent=True) node_set_boot_device_mock.assert_called_once_with( task, boot_devices.DISK, persistent=True) @mock.patch.object(manager_utils, 'node_set_boot_device', autospec=True) def test_try_set_boot_device_some_other_exception( self, node_set_boot_device_mock): exc = exception.IloOperationError(operation="qwe", error="error") node_set_boot_device_mock.side_effect = exc with task_manager.acquire(self.context, self.node.uuid, shared=False) as task: self.assertRaises(exception.IloOperationError, utils.try_set_boot_device, task, boot_devices.DISK, persistent=True) node_set_boot_device_mock.assert_called_once_with( task, boot_devices.DISK, persistent=True) class AgentMethodsTestCase(db_base.DbTestCase): def setUp(self): super(AgentMethodsTestCase, self).setUp() self.clean_steps = { 'deploy': [ {'interface': 'deploy', 'step': 'erase_devices', 'priority': 20}, {'interface': 'deploy', 'step': 'update_firmware', 'priority': 30} ], 'raid': [ {'interface': 'raid', 'step': 'create_configuration', 'priority': 10} ] } n = {'boot_interface': 'pxe', 'deploy_interface': 'direct', 'driver_internal_info': { 'agent_cached_clean_steps': self.clean_steps}} self.node = obj_utils.create_test_node(self.context, **n) self.ports = [obj_utils.create_test_port(self.context, node_id=self.node.id)] def test_agent_add_clean_params(self): cfg.CONF.set_override('shred_random_overwrite_iterations', 2, 'deploy') cfg.CONF.set_override('shred_final_overwrite_with_zeros', False, 'deploy') cfg.CONF.set_override('continue_if_disk_secure_erase_fails', True, 'deploy') cfg.CONF.set_override('enable_ata_secure_erase', False, 'deploy') cfg.CONF.set_override('disk_erasure_concurrency', 8, 'deploy') cfg.CONF.set_override('enable_nvme_secure_erase', False, 'deploy') with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: utils.agent_add_clean_params(task) self.assertEqual(2, task.node.driver_internal_info[ 'agent_erase_devices_iterations']) self.assertIs(False, task.node.driver_internal_info[ 'agent_erase_devices_zeroize']) self.assertIs(True, task.node.driver_internal_info[ 'agent_continue_if_secure_erase_failed']) self.assertIs(False, task.node.driver_internal_info[ 'agent_enable_ata_secure_erase']) self.assertEqual(8, task.node.driver_internal_info[ 'disk_erasure_concurrency']) self.assertIs(False, task.node.driver_internal_info[ 'agent_enable_nvme_secure_erase']) @mock.patch('ironic.conductor.utils.is_fast_track', autospec=True) @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk', autospec=True) @mock.patch('ironic.conductor.utils.node_power_action', autospec=True) @mock.patch.object(utils, 'build_agent_options', autospec=True) @mock.patch('ironic.drivers.modules.network.flat.FlatNetwork.' 'add_cleaning_network', autospec=True) def _test_prepare_inband_cleaning( self, add_cleaning_network_mock, build_options_mock, power_mock, prepare_ramdisk_mock, is_fast_track_mock, manage_boot=True, fast_track=False): build_options_mock.return_value = {'a': 'b'} is_fast_track_mock.return_value = fast_track with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: result = utils.prepare_inband_cleaning(task, manage_boot=manage_boot) add_cleaning_network_mock.assert_called_once_with( task.driver.network, task) if not fast_track: self.assertEqual(states.CLEANWAIT, result) power_mock.assert_called_once_with(task, states.REBOOT) else: self.assertFalse(power_mock.called) self.assertIsNone(result) self.assertEqual(1, task.node.driver_internal_info[ 'agent_erase_devices_iterations']) self.assertIs(True, task.node.driver_internal_info[ 'agent_erase_devices_zeroize']) if manage_boot: prepare_ramdisk_mock.assert_called_once_with( mock.ANY, mock.ANY, {'a': 'b'}) build_options_mock.assert_called_once_with(task.node) else: self.assertFalse(prepare_ramdisk_mock.called) self.assertFalse(build_options_mock.called) def test_prepare_inband_cleaning(self): self._test_prepare_inband_cleaning() def test_prepare_inband_cleaning_manage_boot_false(self): self._test_prepare_inband_cleaning(manage_boot=False) def test_prepare_inband_cleaning_fast_track(self): self._test_prepare_inband_cleaning(fast_track=True) @mock.patch('ironic.conductor.utils.power_on_node_if_needed', autospec=True) @mock.patch('ironic.conductor.utils.is_fast_track', autospec=True) @mock.patch.object(pxe.PXEBoot, 'prepare_ramdisk', autospec=True) @mock.patch('ironic.conductor.utils.node_power_action', autospec=True) @mock.patch.object(utils, 'build_agent_options', autospec=True) @mock.patch('ironic.drivers.modules.network.flat.FlatNetwork.' 'add_cleaning_network', autospec=True) def test_prepare_inband_cleaning_broken_fast_track( self, add_cleaning_network_mock, build_options_mock, power_mock, prepare_ramdisk_mock, is_fast_track_mock, power_on_if_needed_mock): build_options_mock.return_value = {'a': 'b'} is_fast_track_mock.side_effect = [True, False] with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertEqual( states.CLEANWAIT, utils.prepare_inband_cleaning(task)) add_cleaning_network_mock.assert_called_once_with( task.driver.network, task) power_mock.assert_called_once_with(task, states.REBOOT) self.assertEqual(1, task.node.driver_internal_info[ 'agent_erase_devices_iterations']) self.assertIs(True, task.node.driver_internal_info[ 'agent_erase_devices_zeroize']) prepare_ramdisk_mock.assert_called_once_with( mock.ANY, mock.ANY, {'a': 'b'}) build_options_mock.assert_called_once_with(task.node) self.assertFalse(power_on_if_needed_mock.called) @mock.patch('ironic.conductor.utils.is_fast_track', autospec=True) @mock.patch.object(pxe.PXEBoot, 'clean_up_ramdisk', autospec=True) @mock.patch('ironic.drivers.modules.network.flat.FlatNetwork.' 'remove_cleaning_network', autospec=True) @mock.patch('ironic.conductor.utils.node_power_action', autospec=True) def _test_tear_down_inband_cleaning( self, power_mock, remove_cleaning_network_mock, clean_up_ramdisk_mock, is_fast_track_mock, manage_boot=True, fast_track=False, cleaning_error=False): is_fast_track_mock.return_value = fast_track with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: if cleaning_error: task.node.fault = faults.CLEAN_FAILURE utils.tear_down_inband_cleaning(task, manage_boot=manage_boot) if not (fast_track or cleaning_error): power_mock.assert_called_once_with(task, states.POWER_OFF) else: self.assertFalse(power_mock.called) remove_cleaning_network_mock.assert_called_once_with( task.driver.network, task) if manage_boot: clean_up_ramdisk_mock.assert_called_once_with( task.driver.boot, task) else: self.assertFalse(clean_up_ramdisk_mock.called) def test_tear_down_inband_cleaning(self): self._test_tear_down_inband_cleaning(manage_boot=True) def test_tear_down_inband_cleaning_manage_boot_false(self): self._test_tear_down_inband_cleaning(manage_boot=False) def test_tear_down_inband_cleaning_fast_track(self): self._test_tear_down_inband_cleaning(fast_track=True) def test_tear_down_inband_cleaning_cleaning_error(self): self._test_tear_down_inband_cleaning(cleaning_error=True) def test_build_agent_options_conf(self): self.config(endpoint_override='https://api-url', group='service_catalog') options = utils.build_agent_options(self.node) self.assertEqual('https://api-url', options['ipa-api-url']) @mock.patch.object(utils, '_get_ironic_session', autospec=True) def test_build_agent_options_keystone(self, session_mock): sess = mock.Mock() sess.get_endpoint.return_value = 'https://api-url' session_mock.return_value = sess options = utils.build_agent_options(self.node) self.assertEqual('https://api-url', options['ipa-api-url']) def test_direct_deploy_should_convert_raw_image_true(self): cfg.CONF.set_override('force_raw_images', True) cfg.CONF.set_override('stream_raw_images', True, group='agent') internal_info = self.node.driver_internal_info internal_info['is_whole_disk_image'] = True self.node.driver_internal_info = internal_info self.assertTrue( utils.direct_deploy_should_convert_raw_image(self.node)) def test_direct_deploy_should_convert_raw_image_no_force_raw(self): cfg.CONF.set_override('force_raw_images', False) cfg.CONF.set_override('stream_raw_images', True, group='agent') internal_info = self.node.driver_internal_info internal_info['is_whole_disk_image'] = True self.node.driver_internal_info = internal_info self.assertFalse( utils.direct_deploy_should_convert_raw_image(self.node)) def test_direct_deploy_should_convert_raw_image_no_stream(self): cfg.CONF.set_override('force_raw_images', True) cfg.CONF.set_override('stream_raw_images', False, group='agent') internal_info = self.node.driver_internal_info internal_info['is_whole_disk_image'] = True self.node.driver_internal_info = internal_info self.assertFalse( utils.direct_deploy_should_convert_raw_image(self.node)) def test_direct_deploy_should_convert_raw_image_partition(self): cfg.CONF.set_override('force_raw_images', True) cfg.CONF.set_override('stream_raw_images', True, group='agent') internal_info = self.node.driver_internal_info internal_info['is_whole_disk_image'] = False self.node.driver_internal_info = internal_info self.assertTrue( utils.direct_deploy_should_convert_raw_image(self.node)) class ValidateImagePropertiesTestCase(db_base.DbTestCase): def setUp(self): super().setUp() self.node = obj_utils.create_test_node( self.context, boot_interface='pxe', instance_info=INST_INFO_DICT, driver_info=DRV_INFO_DICT, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.task = mock.Mock(context=self.context, node=self.node, spec=['context', 'node']) @mock.patch.object(image_service, 'get_image_service', autospec=True) def test_validate_image_properties_glance_image(self, image_service_mock): inst_info = utils.get_image_instance_info(self.node) image_service_mock.return_value.show.return_value = { 'properties': {'kernel_id': '1111', 'ramdisk_id': '2222'}, } utils.validate_image_properties(self.task, inst_info) image_service_mock.assert_called_once_with( self.node.instance_info['image_source'], context=self.context ) @mock.patch.object(image_service, 'get_image_service', autospec=True) def test_validate_image_properties_glance_image_missing_prop( self, image_service_mock): inst_info = utils.get_image_instance_info(self.node) image_service_mock.return_value.show.return_value = { 'properties': {'kernel_id': '1111'}, } self.assertRaises(exception.MissingParameterValue, utils.validate_image_properties, self.task, inst_info) image_service_mock.assert_called_once_with( self.node.instance_info['image_source'], context=self.context ) @mock.patch.object(utils, 'get_boot_option', autospec=True, return_value='kickstart') @mock.patch.object(image_service, 'get_image_service', autospec=True) def test_validate_image_properties_glance_image_missing_squashfs_id( self, image_service_mock, boot_options_mock): inst_info = utils.get_image_instance_info(self.node) image_service_mock.return_value.show.return_value = { 'properties': {'kernel_id': '1111', 'ramdisk_id': '2222'}, } self.assertRaises(exception.MissingParameterValue, utils.validate_image_properties, self.task, inst_info) image_service_mock.assert_called_once_with( self.node.instance_info['image_source'], context=self.context ) @mock.patch.object(image_service, 'get_image_service', autospec=True) def test_validate_image_properties_glance_image_not_authorized( self, image_service_mock): inst_info = {'image_source': 'uuid'} show_mock = image_service_mock.return_value.show show_mock.side_effect = exception.ImageNotAuthorized(image_id='uuid') self.assertRaises(exception.InvalidParameterValue, utils.validate_image_properties, self.task, inst_info) @mock.patch.object(image_service, 'get_image_service', autospec=True) def test_validate_image_properties_glance_image_not_found( self, image_service_mock): inst_info = {'image_source': 'uuid'} show_mock = image_service_mock.return_value.show show_mock.side_effect = exception.ImageNotFound(image_id='uuid') self.assertRaises(exception.InvalidParameterValue, utils.validate_image_properties, self.task, inst_info) def test_validate_image_properties_invalid_image_href(self): inst_info = {'image_source': 'emule://uuid'} self.assertRaises(exception.InvalidParameterValue, utils.validate_image_properties, self.task, inst_info) @mock.patch.object(image_service.HttpImageService, 'show', autospec=True) def test_validate_image_properties_nonglance_image( self, image_service_show_mock): instance_info = { 'image_source': 'http://ubuntu', 'kernel': 'kernel_uuid', 'ramdisk': 'file://initrd', 'root_gb': 100, } image_service_show_mock.return_value = {'size': 1, 'properties': {}} self.node.instance_info = instance_info inst_info = utils.get_image_instance_info(self.node) utils.validate_image_properties(self.task, inst_info) image_service_show_mock.assert_called_once_with( mock.ANY, instance_info['image_source']) @mock.patch.object(image_service.HttpImageService, 'show', autospec=True) def test_validate_image_properties_nonglance_image_validation_fail( self, img_service_show_mock): instance_info = { 'image_source': 'http://ubuntu', 'kernel': 'kernel_uuid', 'ramdisk': 'file://initrd', 'root_gb': 100, } img_service_show_mock.side_effect = exception.ImageRefValidationFailed( image_href='http://ubuntu', reason='HTTPError') self.node.instance_info = instance_info inst_info = utils.get_image_instance_info(self.node) expected_error = ('Validation of image href http://ubuntu ' 'failed, reason: HTTPError') error = self.assertRaises(exception.InvalidParameterValue, utils.validate_image_properties, self.task, inst_info) self.assertEqual(expected_error, str(error)) def test_validate_image_properties_boot_iso_conflict(self): instance_info = { 'image_source': 'http://ubuntu', 'boot_iso': 'http://ubuntu.iso', } expected_error = ("An 'image_source' and 'boot_iso' " "parameter may not be specified at " "the same time.") error = self.assertRaises(exception.InvalidParameterValue, utils.validate_image_properties, self.task, instance_info) self.assertEqual(expected_error, str(error)) @mock.patch.object(utils, 'get_boot_option', autospec=True, return_value='ramdisk') @mock.patch.object(image_service.HttpImageService, 'show', autospec=True) def test_validate_image_properties_ramdisk_deploy( self, image_service_show_mock, boot_options_mock): instance_info = { 'kernel': 'file://kernel', 'ramdisk': 'file://initrd', } self.node.instance_info = instance_info inst_info = utils.get_image_instance_info(self.node) utils.validate_image_properties(self.task, inst_info) image_service_show_mock.assert_not_called() class ValidateParametersTestCase(db_base.DbTestCase): def _test__get_img_instance_info( self, instance_info=INST_INFO_DICT, driver_info=DRV_INFO_DICT, driver_internal_info=DRV_INTERNAL_INFO_DICT): # make sure we get back the expected things node = obj_utils.create_test_node( self.context, boot_interface='pxe', instance_info=instance_info, driver_info=driver_info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) return utils.get_image_instance_info(node) def test__get_img_instance_info_good(self): info = self._test__get_img_instance_info() self.assertIsNotNone(info['image_source']) def test__get_img_instance_info_good_non_glance_image(self): instance_info = INST_INFO_DICT.copy() instance_info['image_source'] = 'http://image' instance_info['kernel'] = 'http://kernel' instance_info['ramdisk'] = 'http://ramdisk' info = self._test__get_img_instance_info(instance_info=instance_info) self.assertIsNotNone(info['image_source']) self.assertIsNotNone(info['ramdisk']) self.assertIsNotNone(info['kernel']) def test__get_img_instance_info_non_glance_image_missing_kernel(self): instance_info = INST_INFO_DICT.copy() instance_info['image_source'] = 'http://image' instance_info['ramdisk'] = 'http://ramdisk' self.assertRaises( exception.MissingParameterValue, self._test__get_img_instance_info, instance_info=instance_info) def test__get_img_instance_info_non_glance_image_missing_ramdisk(self): instance_info = INST_INFO_DICT.copy() instance_info['image_source'] = 'http://image' instance_info['kernel'] = 'http://kernel' self.assertRaises( exception.MissingParameterValue, self._test__get_img_instance_info, instance_info=instance_info) def test__get_img_instance_info_missing_image_source(self): instance_info = INST_INFO_DICT.copy() del instance_info['image_source'] self.assertRaises( exception.MissingParameterValue, self._test__get_img_instance_info, instance_info=instance_info) def test__get_img_instance_info_whole_disk_image(self): driver_internal_info = DRV_INTERNAL_INFO_DICT.copy() driver_internal_info['is_whole_disk_image'] = True info = self._test__get_img_instance_info( driver_internal_info=driver_internal_info) self.assertIsNotNone(info['image_source']) def test__get_img_instance_info_boot_iso_only(self): instance_info = { 'boot_iso': 'http://iso' } info = self._test__get_img_instance_info(instance_info=instance_info) self.assertIsNotNone(info['boot_iso']) self.assertNotIn('image_source', info) @mock.patch.object(utils, 'get_boot_option', autospec=True, return_value='ramdisk') def test__get_img_instance_info_ramdisk_deploy(self, mock_boot_opt): instance_info = { 'kernel': 'http://kernel', 'ramdisk': 'http://ramdisk', } info = self._test__get_img_instance_info(instance_info=instance_info) self.assertIsNotNone(info['kernel']) self.assertIsNotNone(info['ramdisk']) self.assertNotIn('image_source', info) class InstanceInfoTestCase(db_base.DbTestCase): def test_parse_instance_info_good(self): # make sure we get back the expected things node = obj_utils.create_test_node( self.context, boot_interface='pxe', instance_info=INST_INFO_DICT, driver_internal_info=DRV_INTERNAL_INFO_DICT ) info = utils.parse_instance_info(node) self.assertIsNotNone(info['image_source']) self.assertIsNotNone(info['root_gb']) self.assertEqual(0, info['ephemeral_gb']) self.assertIsNone(info['configdrive']) def test_parse_instance_info_missing_instance_source(self): # make sure error is raised when info is missing info = dict(INST_INFO_DICT) del info['image_source'] node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.assertRaises(exception.MissingParameterValue, utils.parse_instance_info, node) def test_parse_instance_info_missing_root_gb(self): # make sure error is raised when info is missing info = dict(INST_INFO_DICT) del info['root_gb'] node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.assertRaises(exception.MissingParameterValue, utils.parse_instance_info, node) def test_parse_instance_info_invalid_root_gb(self): info = dict(INST_INFO_DICT) info['root_gb'] = 'foobar' node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.assertRaises(exception.InvalidParameterValue, utils.parse_instance_info, node) def test_parse_instance_info_valid_ephemeral_gb(self): ephemeral_gb = 10 ephemeral_mb = 1024 * ephemeral_gb ephemeral_fmt = 'test-fmt' info = dict(INST_INFO_DICT) info['ephemeral_gb'] = ephemeral_gb info['ephemeral_format'] = ephemeral_fmt node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) data = utils.parse_instance_info(node) self.assertEqual(ephemeral_mb, data['ephemeral_mb']) self.assertEqual(ephemeral_fmt, data['ephemeral_format']) def test_parse_instance_info_unicode_swap_mb(self): swap_mb = u'10' swap_mb_int = 10 info = dict(INST_INFO_DICT) info['swap_mb'] = swap_mb node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) data = utils.parse_instance_info(node) self.assertEqual(swap_mb_int, data['swap_mb']) def test_parse_instance_info_invalid_ephemeral_gb(self): info = dict(INST_INFO_DICT) info['ephemeral_gb'] = 'foobar' info['ephemeral_format'] = 'exttest' node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.assertRaises(exception.InvalidParameterValue, utils.parse_instance_info, node) def test_parse_instance_info_valid_ephemeral_missing_format(self): ephemeral_gb = 10 ephemeral_fmt = 'test-fmt' info = dict(INST_INFO_DICT) info['ephemeral_gb'] = ephemeral_gb info['ephemeral_format'] = None self.config(default_ephemeral_format=ephemeral_fmt, group='pxe') node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) instance_info = utils.parse_instance_info(node) self.assertEqual(ephemeral_fmt, instance_info['ephemeral_format']) def test_parse_instance_info_valid_preserve_ephemeral_true(self): info = dict(INST_INFO_DICT) for opt in ['true', 'TRUE', 'True', 't', 'on', 'yes', 'y', '1']: info['preserve_ephemeral'] = opt node = obj_utils.create_test_node( self.context, uuid=uuidutils.generate_uuid(), instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) data = utils.parse_instance_info(node) self.assertTrue(data['preserve_ephemeral']) def test_parse_instance_info_valid_preserve_ephemeral_false(self): info = dict(INST_INFO_DICT) for opt in ['false', 'FALSE', 'False', 'f', 'off', 'no', 'n', '0']: info['preserve_ephemeral'] = opt node = obj_utils.create_test_node( self.context, uuid=uuidutils.generate_uuid(), instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) data = utils.parse_instance_info(node) self.assertFalse(data['preserve_ephemeral']) def test_parse_instance_info_invalid_preserve_ephemeral(self): info = dict(INST_INFO_DICT) info['preserve_ephemeral'] = 'foobar' node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.assertRaises(exception.InvalidParameterValue, utils.parse_instance_info, node) def test_parse_instance_info_invalid_ephemeral_disk(self): info = dict(INST_INFO_DICT) info['ephemeral_gb'] = 10 info['swap_mb'] = 0 info['root_gb'] = 20 info['preserve_ephemeral'] = True drv_internal_dict = {'instance': {'ephemeral_gb': 9, 'swap_mb': 0, 'root_gb': 20}} drv_internal_dict.update(DRV_INTERNAL_INFO_DICT) node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=drv_internal_dict, ) self.assertRaises(exception.InvalidParameterValue, utils.parse_instance_info, node) def test__check_disk_layout_unchanged_fails(self): info = dict(INST_INFO_DICT) info['ephemeral_gb'] = 10 info['swap_mb'] = 0 info['root_gb'] = 20 info['preserve_ephemeral'] = True drv_internal_dict = {'instance': {'ephemeral_gb': 20, 'swap_mb': 0, 'root_gb': 20}} drv_internal_dict.update(DRV_INTERNAL_INFO_DICT) node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=drv_internal_dict, ) self.assertRaises(exception.InvalidParameterValue, utils._check_disk_layout_unchanged, node, info) def test__check_disk_layout_unchanged(self): info = dict(INST_INFO_DICT) info['ephemeral_gb'] = 10 info['swap_mb'] = 0 info['root_gb'] = 20 info['preserve_ephemeral'] = True drv_internal_dict = {'instance': {'ephemeral_gb': 10, 'swap_mb': 0, 'root_gb': 20}} drv_internal_dict.update(DRV_INTERNAL_INFO_DICT) node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=drv_internal_dict, ) self.assertIsNone(utils._check_disk_layout_unchanged(node, info)) def test_parse_instance_info_configdrive(self): info = dict(INST_INFO_DICT) info['configdrive'] = 'http://1.2.3.4/cd' node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) instance_info = utils.parse_instance_info(node) self.assertEqual('http://1.2.3.4/cd', instance_info['configdrive']) def test_parse_instance_info_nonglance_image(self): info = INST_INFO_DICT.copy() info['image_source'] = 'file:///image.qcow2' info['kernel'] = 'file:///image.vmlinuz' info['ramdisk'] = 'file:///image.initrd' node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) utils.parse_instance_info(node) def test_parse_instance_info_nonglance_image_no_kernel(self): info = INST_INFO_DICT.copy() info['image_source'] = 'file:///image.qcow2' info['ramdisk'] = 'file:///image.initrd' node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=DRV_INTERNAL_INFO_DICT, ) self.assertRaises(exception.MissingParameterValue, utils.parse_instance_info, node) def test_parse_instance_info_whole_disk_image(self): driver_internal_info = dict(DRV_INTERNAL_INFO_DICT) driver_internal_info['is_whole_disk_image'] = True node = obj_utils.create_test_node( self.context, instance_info=INST_INFO_DICT, driver_internal_info=driver_internal_info, ) instance_info = utils.parse_instance_info(node) self.assertIsNotNone(instance_info['image_source']) self.assertNotIn('root_mb', instance_info) self.assertNotIn('ephemeral_mb', instance_info) self.assertNotIn('swap_mb', instance_info) self.assertIsNone(instance_info['configdrive']) def test_parse_instance_info_whole_disk_image_missing_root(self): driver_internal_info = dict(DRV_INTERNAL_INFO_DICT) driver_internal_info['is_whole_disk_image'] = True info = dict(INST_INFO_DICT) del info['root_gb'] node = obj_utils.create_test_node( self.context, instance_info=info, driver_internal_info=driver_internal_info ) instance_info = utils.parse_instance_info(node) self.assertIsNotNone(instance_info['image_source']) self.assertNotIn('root_mb', instance_info) self.assertNotIn('ephemeral_mb', instance_info) self.assertNotIn('swap_mb', instance_info) class TestBuildInstanceInfoForDeploy(db_base.DbTestCase): def setUp(self): super(TestBuildInstanceInfoForDeploy, self).setUp() self.node = obj_utils.create_test_node(self.context, boot_interface='pxe', deploy_interface='direct') cfg.CONF.set_override('image_download_source', 'swift', group='agent') @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) @mock.patch.object(image_service, 'GlanceImageService', autospec=True) def test_build_instance_info_for_deploy_glance_image(self, glance_mock, validate_mock): i_info = self.node.instance_info i_info['image_source'] = '733d1c44-a2ea-414b-aca7-69decf20d810' 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.instance_info = i_info self.node.save() image_info = {'checksum': 'aa', 'disk_format': 'qcow2', 'os_hash_algo': 'sha512', 'os_hash_value': 'fake-sha512', 'container_format': 'bare', 'properties': {}} glance_mock.return_value.show = mock.MagicMock(spec_set=[], return_value=image_info) glance_mock.return_value.swift_temp_url.return_value = ( 'http://temp-url') with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: utils.build_instance_info_for_deploy(task) glance_mock.assert_called_once_with(context=task.context) glance_mock.return_value.show.assert_called_once_with( self.node.instance_info['image_source']) glance_mock.return_value.swift_temp_url.assert_called_once_with( image_info) validate_mock.assert_called_once_with(mock.ANY, 'http://temp-url', secret=True) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) @mock.patch.object(utils, 'parse_instance_info', autospec=True) @mock.patch.object(image_service, 'GlanceImageService', autospec=True) def test_build_instance_info_for_deploy_glance_partition_image( self, glance_mock, parse_instance_info_mock, validate_mock): i_info = {} i_info['image_source'] = '733d1c44-a2ea-414b-aca7-69decf20d810' i_info['kernel'] = '13ce5a56-1de3-4916-b8b2-be778645d003' i_info['ramdisk'] = 'a5a370a8-1b39-433f-be63-2c7d708e4b4e' i_info['root_gb'] = 5 i_info['swap_mb'] = 4 i_info['ephemeral_gb'] = 0 i_info['ephemeral_format'] = None i_info['configdrive'] = 'configdrive' driver_internal_info = self.node.driver_internal_info driver_internal_info['is_whole_disk_image'] = False self.node.driver_internal_info = driver_internal_info self.node.instance_info = i_info self.node.save() image_info = {'checksum': 'aa', 'disk_format': 'qcow2', 'os_hash_algo': 'sha512', 'os_hash_value': 'fake-sha512', 'container_format': 'bare', 'properties': {'kernel_id': 'kernel', 'ramdisk_id': 'ramdisk'}} glance_mock.return_value.show = mock.MagicMock(spec_set=[], return_value=image_info) glance_obj_mock = glance_mock.return_value glance_obj_mock.swift_temp_url.return_value = 'http://temp-url' parse_instance_info_mock.return_value = {'swap_mb': 4} image_source = '733d1c44-a2ea-414b-aca7-69decf20d810' expected_i_info = {'root_gb': 5, 'swap_mb': 4, 'ephemeral_gb': 0, 'ephemeral_format': None, 'configdrive': 'configdrive', 'image_source': image_source, 'image_url': 'http://temp-url', 'kernel': 'kernel', 'ramdisk': 'ramdisk', 'image_type': 'partition', 'image_tags': [], 'image_properties': {'kernel_id': 'kernel', 'ramdisk_id': 'ramdisk'}, 'image_checksum': 'aa', 'image_os_hash_algo': 'sha512', 'image_os_hash_value': 'fake-sha512', 'image_container_format': 'bare', 'image_disk_format': 'qcow2'} with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) glance_mock.assert_called_once_with(context=task.context) glance_mock.return_value.show.assert_called_once_with( self.node.instance_info['image_source']) glance_mock.return_value.swift_temp_url.assert_called_once_with( image_info) validate_mock.assert_called_once_with( mock.ANY, 'http://temp-url', secret=True) image_type = task.node.instance_info['image_type'] self.assertEqual('partition', image_type) self.assertEqual('kernel', info['kernel']) self.assertEqual('ramdisk', info['ramdisk']) self.assertEqual(expected_i_info, info) parse_instance_info_mock.assert_called_once_with(task.node) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_for_deploy_nonglance_image( self, validate_href_mock): i_info = self.node.instance_info driver_internal_info = self.node.driver_internal_info i_info['image_source'] = 'http://image-ref' i_info['image_checksum'] = 'aa' i_info['root_gb'] = 10 i_info['image_checksum'] = 'aa' driver_internal_info['is_whole_disk_image'] = True self.node.instance_info = i_info self.node.driver_internal_info = driver_internal_info self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) self.assertEqual(self.node.instance_info['image_source'], info['image_url']) validate_href_mock.assert_called_once_with( mock.ANY, 'http://image-ref', False) @mock.patch.object(utils, 'parse_instance_info', autospec=True) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_for_deploy_nonglance_partition_image( self, validate_href_mock, parse_instance_info_mock): i_info = {} driver_internal_info = self.node.driver_internal_info i_info['image_source'] = 'http://image-ref' i_info['kernel'] = 'http://kernel-ref' i_info['ramdisk'] = 'http://ramdisk-ref' i_info['image_checksum'] = 'aa' i_info['root_gb'] = 10 i_info['configdrive'] = 'configdrive' driver_internal_info['is_whole_disk_image'] = False self.node.instance_info = i_info self.node.driver_internal_info = driver_internal_info self.node.save() validate_href_mock.side_effect = ['http://image-ref', 'http://kernel-ref', 'http://ramdisk-ref'] parse_instance_info_mock.return_value = {'swap_mb': 5} expected_i_info = {'image_source': 'http://image-ref', 'image_url': 'http://image-ref', 'image_type': 'partition', 'kernel': 'http://kernel-ref', 'ramdisk': 'http://ramdisk-ref', 'image_checksum': 'aa', 'root_gb': 10, 'swap_mb': 5, 'configdrive': 'configdrive'} with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) self.assertEqual(self.node.instance_info['image_source'], info['image_url']) validate_href_mock.assert_called_once_with( mock.ANY, 'http://image-ref', False) self.assertEqual('partition', info['image_type']) self.assertEqual(expected_i_info, info) parse_instance_info_mock.assert_called_once_with(task.node) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_for_deploy_image_not_found( self, validate_href_mock): validate_href_mock.side_effect = exception.ImageRefValidationFailed( image_href='http://img.qcow2', reason='fail') i_info = self.node.instance_info i_info['image_source'] = 'http://img.qcow2' i_info['image_checksum'] = 'aa' self.node.instance_info = i_info self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertRaises(exception.ImageRefValidationFailed, utils.build_instance_info_for_deploy, task) class TestBuildInstanceInfoForHttpProvisioning(db_base.DbTestCase): def setUp(self): super(TestBuildInstanceInfoForHttpProvisioning, self).setUp() self.node = obj_utils.create_test_node(self.context, boot_interface='pxe', deploy_interface='direct') i_info = self.node.instance_info i_info['image_source'] = '733d1c44-a2ea-414b-aca7-69decf20d810' i_info['root_gb'] = 100 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.instance_info = i_info self.node.save() self.checksum_mock = self.useFixture(fixtures.MockPatchObject( fileutils, 'compute_file_checksum')).mock self.checksum_mock.return_value = 'fake-checksum' self.cache_image_mock = self.useFixture(fixtures.MockPatchObject( utils, 'cache_instance_image', autospec=True)).mock self.fake_path = '/var/lib/ironic/images/{}/disk'.format( self.node.uuid) self.cache_image_mock.return_value = ( '733d1c44-a2ea-414b-aca7-69decf20d810', self.fake_path) self.ensure_tree_mock = self.useFixture(fixtures.MockPatchObject( utils.fileutils, 'ensure_tree', autospec=True)).mock self.create_link_mock = self.useFixture(fixtures.MockPatchObject( common_utils, 'create_link_without_raise', autospec=True)).mock cfg.CONF.set_override('http_url', 'http://172.172.24.10:8080', group='deploy') cfg.CONF.set_override('image_download_source', 'http', group='agent') self.expected_url = '/'.join([cfg.CONF.deploy.http_url, cfg.CONF.deploy.http_image_subdir, self.node.uuid]) self.image_info = {'checksum': 'aa', 'disk_format': 'qcow2', 'os_hash_algo': 'sha512', 'os_hash_value': 'fake-sha512', 'container_format': 'bare', 'properties': {}} @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) @mock.patch.object(image_service, 'GlanceImageService', autospec=True) def _test_build_instance_info(self, glance_mock, validate_mock, image_info={}, expect_raw=False): glance_mock.return_value.show = mock.MagicMock(spec_set=[], return_value=image_info) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: instance_info = utils.build_instance_info_for_deploy(task) glance_mock.assert_called_once_with(context=task.context) glance_mock.return_value.show.assert_called_once_with( self.node.instance_info['image_source']) self.cache_image_mock.assert_called_once_with(task.context, task.node, force_raw=expect_raw) symlink_dir = utils._get_http_image_symlink_dir_path() symlink_file = utils._get_http_image_symlink_file_path( self.node.uuid) image_path = utils._get_image_file_path(self.node.uuid) self.ensure_tree_mock.assert_called_once_with(symlink_dir) self.create_link_mock.assert_called_once_with(image_path, symlink_file) validate_mock.assert_called_once_with(mock.ANY, self.expected_url, secret=False) return image_path, instance_info def test_build_instance_info_no_force_raw(self): cfg.CONF.set_override('force_raw_images', False) _, instance_info = self._test_build_instance_info( image_info=self.image_info, expect_raw=False) self.assertEqual(instance_info['image_checksum'], 'aa') self.assertEqual(instance_info['image_disk_format'], 'qcow2') self.assertEqual(instance_info['image_os_hash_algo'], 'sha512') self.assertEqual(instance_info['image_os_hash_value'], 'fake-sha512') self.checksum_mock.assert_not_called() def test_build_instance_info_force_raw(self): cfg.CONF.set_override('force_raw_images', True) image_path, instance_info = self._test_build_instance_info( image_info=self.image_info, expect_raw=True) self.assertIsNone(instance_info['image_checksum']) self.assertEqual(instance_info['image_disk_format'], 'raw') calls = [mock.call(image_path, algorithm='sha512')] self.checksum_mock.assert_has_calls(calls) def test_build_instance_info_force_raw_drops_md5(self): cfg.CONF.set_override('force_raw_images', True) self.image_info['os_hash_algo'] = 'md5' image_path, instance_info = self._test_build_instance_info( image_info=self.image_info, expect_raw=True) self.assertIsNone(instance_info['image_checksum']) self.assertEqual(instance_info['image_disk_format'], 'raw') calls = [mock.call(image_path, algorithm='sha256')] self.checksum_mock.assert_has_calls(calls) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_file_image(self, validate_href_mock): i_info = self.node.instance_info driver_internal_info = self.node.driver_internal_info i_info['image_source'] = 'file://image-ref' i_info['image_checksum'] = 'aa' i_info['root_gb'] = 10 i_info['image_checksum'] = 'aa' driver_internal_info['is_whole_disk_image'] = True self.node.instance_info = i_info self.node.driver_internal_info = driver_internal_info self.node.save() expected_url = ( 'http://172.172.24.10:8080/agent_images/%s' % self.node.uuid) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) self.assertEqual(expected_url, info['image_url']) self.assertEqual('sha256', info['image_os_hash_algo']) self.assertEqual('fake-checksum', info['image_os_hash_value']) self.cache_image_mock.assert_called_once_with( task.context, task.node, force_raw=True) self.checksum_mock.assert_called_once_with( self.fake_path, algorithm='sha256') validate_href_mock.assert_called_once_with( mock.ANY, expected_url, False) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_local_image(self, validate_href_mock): cfg.CONF.set_override('image_download_source', 'local', group='agent') i_info = self.node.instance_info driver_internal_info = self.node.driver_internal_info i_info['image_source'] = 'http://image-ref' i_info['image_checksum'] = 'aa' i_info['root_gb'] = 10 i_info['image_checksum'] = 'aa' driver_internal_info['is_whole_disk_image'] = True self.node.instance_info = i_info self.node.driver_internal_info = driver_internal_info self.node.save() expected_url = ( 'http://172.172.24.10:8080/agent_images/%s' % self.node.uuid) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) self.assertEqual(expected_url, info['image_url']) self.assertEqual('sha256', info['image_os_hash_algo']) self.assertEqual('fake-checksum', info['image_os_hash_value']) self.cache_image_mock.assert_called_once_with( task.context, task.node, force_raw=True) self.checksum_mock.assert_called_once_with( self.fake_path, algorithm='sha256') validate_href_mock.assert_called_once_with( mock.ANY, expected_url, False) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_local_image_via_iinfo(self, validate_href_mock): cfg.CONF.set_override('image_download_source', 'http', group='agent') i_info = self.node.instance_info driver_internal_info = self.node.driver_internal_info i_info['image_source'] = 'http://image-ref' i_info['image_checksum'] = 'aa' i_info['root_gb'] = 10 i_info['image_checksum'] = 'aa' i_info['image_download_source'] = 'local' driver_internal_info['is_whole_disk_image'] = True self.node.instance_info = i_info self.node.driver_internal_info = driver_internal_info self.node.save() expected_url = ( 'http://172.172.24.10:8080/agent_images/%s' % self.node.uuid) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) self.assertEqual(expected_url, info['image_url']) self.assertEqual('sha256', info['image_os_hash_algo']) self.assertEqual('fake-checksum', info['image_os_hash_value']) self.cache_image_mock.assert_called_once_with( task.context, task.node, force_raw=True) self.checksum_mock.assert_called_once_with( self.fake_path, algorithm='sha256') validate_href_mock.assert_called_once_with( mock.ANY, expected_url, False) @mock.patch.object(image_service.HttpImageService, 'validate_href', autospec=True) def test_build_instance_info_local_image_via_dinfo(self, validate_href_mock): cfg.CONF.set_override('image_download_source', 'http', group='agent') i_info = self.node.instance_info d_info = self.node.driver_info driver_internal_info = self.node.driver_internal_info i_info['image_source'] = 'http://image-ref' i_info['image_checksum'] = 'aa' i_info['root_gb'] = 10 i_info['image_checksum'] = 'aa' d_info['image_download_source'] = 'local' driver_internal_info['is_whole_disk_image'] = True self.node.instance_info = i_info self.node.driver_info = d_info self.node.driver_internal_info = driver_internal_info self.node.save() expected_url = ( 'http://172.172.24.10:8080/agent_images/%s' % self.node.uuid) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: info = utils.build_instance_info_for_deploy(task) self.assertEqual(expected_url, info['image_url']) self.assertEqual('sha256', info['image_os_hash_algo']) self.assertEqual('fake-checksum', info['image_os_hash_value']) self.cache_image_mock.assert_called_once_with( task.context, task.node, force_raw=True) self.checksum_mock.assert_called_once_with( self.fake_path, algorithm='sha256') validate_href_mock.assert_called_once_with( mock.ANY, expected_url, False) class TestStorageInterfaceUtils(db_base.DbTestCase): def setUp(self): super(TestStorageInterfaceUtils, self).setUp() self.node = obj_utils.create_test_node(self.context, driver='fake-hardware') self.config(enabled_storage_interfaces=['noop', 'fake', 'cinder']) def test_check_interface_capability(self): class fake_driver(object): capabilities = ['foo', 'bar'] self.assertTrue(utils.check_interface_capability(fake_driver, 'foo')) self.assertFalse(utils.check_interface_capability(fake_driver, 'baz')) def test_get_remote_boot_volume(self): obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=1, volume_id='4321') obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=0, volume_id='1234', uuid=uuidutils.generate_uuid()) self.node.storage_interface = 'cinder' self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: volume = utils.get_remote_boot_volume(task) self.assertEqual('1234', volume['volume_id']) def test_get_remote_boot_volume_none(self): with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertIsNone(utils.get_remote_boot_volume(task)) obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=1, volume_id='4321') with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertIsNone(utils.get_remote_boot_volume(task)) @mock.patch.object(fake.FakeBoot, 'capabilities', ['iscsi_volume_boot'], create=True) @mock.patch.object(fake.FakeDeploy, 'capabilities', ['iscsi_volume_deploy'], create=True) @mock.patch.object(cinder.CinderStorage, 'should_write_image', autospec=True) def test_populate_storage_driver_internal_info_iscsi(self, mock_should_write): mock_should_write.return_value = True vol_uuid = uuidutils.generate_uuid() obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=0, volume_id='1234', uuid=vol_uuid) # NOTE(TheJulia): Since the default for the storage_interface # is a noop interface, we need to define another driver that # can be loaded by driver_manager in order to create the task # to test this method. self.node.storage_interface = "cinder" self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: driver_utils.add_node_capability(task, 'iscsi_boot', 'True') utils.populate_storage_driver_internal_info(task) self.assertEqual( vol_uuid, task.node.driver_internal_info.get('boot_from_volume', None)) self.assertEqual( vol_uuid, task.node.driver_internal_info.get('boot_from_volume_deploy', None)) @mock.patch.object(fake.FakeBoot, 'capabilities', ['fibre_channel_volume_boot'], create=True) @mock.patch.object(fake.FakeDeploy, 'capabilities', ['fibre_channel_volume_deploy'], create=True) @mock.patch.object(cinder.CinderStorage, 'should_write_image', autospec=True) def test_populate_storage_driver_internal_info_fc(self, mock_should_write): mock_should_write.return_value = True self.node.storage_interface = "cinder" self.node.save() vol_uuid = uuidutils.generate_uuid() obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='fibre_channel', boot_index=0, volume_id='1234', uuid=vol_uuid) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: driver_utils.add_node_capability(task, 'fibre_channel_boot', 'True') utils.populate_storage_driver_internal_info(task) self.assertEqual( vol_uuid, task.node.driver_internal_info.get('boot_from_volume', None)) self.assertEqual( vol_uuid, task.node.driver_internal_info.get('boot_from_volume_deploy', None)) @mock.patch.object(fake.FakeBoot, 'capabilities', ['fibre_channel_volume_boot'], create=True) @mock.patch.object(fake.FakeDeploy, 'capabilities', ['fibre_channel_volume_deploy'], create=True) def test_populate_storage_driver_internal_info_error(self): obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=0, volume_id='1234') with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertRaises(exception.StorageError, utils.populate_storage_driver_internal_info, task) def test_tear_down_storage_configuration(self): vol_uuid = uuidutils.generate_uuid() obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=0, volume_id='1234', uuid=vol_uuid) d_i_info = self.node.driver_internal_info d_i_info['boot_from_volume'] = vol_uuid d_i_info['boot_from_volume_deploy'] = vol_uuid self.node.driver_internal_info = d_i_info self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: node = task.node self.assertEqual(1, len(task.volume_targets)) self.assertEqual( vol_uuid, node.driver_internal_info.get('boot_from_volume')) self.assertEqual( vol_uuid, node.driver_internal_info.get('boot_from_volume_deploy')) utils.tear_down_storage_configuration(task) node.refresh() self.assertIsNone( node.driver_internal_info.get('boot_from_volume')) self.assertIsNone( node.driver_internal_info.get('boot_from_volume_deploy')) with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertEqual(0, len(task.volume_targets)) def test_is_iscsi_boot(self): vol_id = uuidutils.generate_uuid() obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='iscsi', boot_index=0, volume_id='1234', uuid=vol_id) self.node.driver_internal_info = {'boot_from_volume': vol_id} self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertTrue(utils.is_iscsi_boot(task)) def test_is_iscsi_boot_exception(self): self.node.driver_internal_info = { 'boot_from_volume': uuidutils.generate_uuid()} with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertFalse(utils.is_iscsi_boot(task)) def test_is_iscsi_boot_false(self): with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertFalse(utils.is_iscsi_boot(task)) def test_is_iscsi_boot_false_fc_target(self): vol_id = uuidutils.generate_uuid() obj_utils.create_test_volume_target( self.context, node_id=self.node.id, volume_type='fibre_channel', boot_index=0, volume_id='3214', uuid=vol_id) self.node.driver_internal_info.update({'boot_from_volume': vol_id}) self.node.save() with task_manager.acquire( self.context, self.node.uuid, shared=False) as task: self.assertFalse(utils.is_iscsi_boot(task)) class InstanceImageCacheTestCase(db_base.DbTestCase): @mock.patch.object(fileutils, 'ensure_tree', autospec=True) def test_with_master_path(self, mock_ensure_tree): self.config(instance_master_path='/fake/path', group='pxe') self.config(image_cache_size=500, group='pxe') self.config(image_cache_ttl=30, group='pxe') cache = utils.InstanceImageCache() mock_ensure_tree.assert_called_once_with('/fake/path') self.assertEqual(500 * 1024 * 1024, cache._cache_size) self.assertEqual(30 * 60, cache._cache_ttl) @mock.patch.object(fileutils, 'ensure_tree', autospec=True) def test_without_master_path(self, mock_ensure_tree): self.config(instance_master_path='', group='pxe') self.config(image_cache_size=500, group='pxe') self.config(image_cache_ttl=30, group='pxe') cache = utils.InstanceImageCache() mock_ensure_tree.assert_not_called() self.assertEqual(500 * 1024 * 1024, cache._cache_size) self.assertEqual(30 * 60, cache._cache_ttl) class AsyncStepTestCase(db_base.DbTestCase): def setUp(self): super(AsyncStepTestCase, self).setUp() self.node = obj_utils.create_test_node(self.context, driver="fake-hardware") def _test_get_async_step_return_state(self): result = utils.get_async_step_return_state(self.node) if self.node.clean_step: self.assertEqual(states.CLEANWAIT, result) else: self.assertEqual(states.DEPLOYWAIT, result) def test_get_async_step_return_state_cleaning(self): self.node.clean_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.save() self._test_get_async_step_return_state() def test_get_async_step_return_state_deploying(self): self.node.deploy_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.save() self._test_get_async_step_return_state() def test_set_async_step_flags_cleaning_set_all(self): self.node.clean_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.driver_internal_info = {} expected = {'cleaning_reboot': True, 'cleaning_polling': True, 'skip_current_clean_step': True} self.node.save() utils.set_async_step_flags(self.node, reboot=True, skip_current_step=True, polling=True) self.assertEqual(expected, self.node.driver_internal_info) def test_set_async_step_flags_cleaning_set_one(self): self.node.clean_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.driver_internal_info = {} self.node.save() utils.set_async_step_flags(self.node, reboot=True) self.assertEqual({'cleaning_reboot': True}, self.node.driver_internal_info) def test_set_async_step_flags_deploying_set_all(self): self.node.deploy_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.driver_internal_info = { 'agent_secret_token': 'test', 'agent_secret_token_pregenerated': True} expected = {'deployment_reboot': True, 'deployment_polling': True, 'skip_current_deploy_step': True, 'agent_secret_token': 'test', 'agent_secret_token_pregenerated': True} self.node.save() utils.set_async_step_flags(self.node, reboot=True, skip_current_step=True, polling=True) self.assertEqual(expected, self.node.driver_internal_info) def test_set_async_step_flags_deploying_set_one(self): self.node.deploy_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.driver_internal_info = {} self.node.save() utils.set_async_step_flags(self.node, reboot=True) self.assertEqual({'deployment_reboot': True}, self.node.driver_internal_info) def test_set_async_step_flags_clears_non_pregenerated_token(self): self.node.clean_step = {'step': 'create_configuration', 'interface': 'raid'} self.node.driver_internal_info = {'agent_secret_token': 'test'} expected = {'cleaning_reboot': True, 'cleaning_polling': True, 'skip_current_clean_step': True} self.node.save() utils.set_async_step_flags(self.node, reboot=True, skip_current_step=True, polling=True) self.assertEqual(expected, self.node.driver_internal_info)