# Copyright 2014 Hewlett-Packard Development Company, L.P.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Test class for common methods used by iLO modules."""
import hashlib
import os
import shutil
import tempfile
from ironic_lib import utils as ironic_utils
import mock
from oslo_config import cfg
from oslo_utils import importutils
import six
import six.moves.builtins as __builtin__
from ironic.common import boot_devices
from ironic.common import exception
from ironic.common import images
from ironic.common import swift
from ironic.conductor import task_manager
from ironic.conductor import utils as manager_utils
from ironic.drivers.modules import deploy_utils
from ironic.drivers.modules.ilo import common as ilo_common
from ironic.tests.unit.conductor import mgr_utils
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
ilo_client = importutils.try_import('proliantutils.ilo.client')
ilo_error = importutils.try_import('proliantutils.exception')
if six.PY3:
import io
file = io.BytesIO
CONF = cfg.CONF
[docs]class IloValidateParametersTestCase(db_base.DbTestCase):
[docs] def setUp(self):
super(IloValidateParametersTestCase, self).setUp()
self.node = obj_utils.create_test_node(
self.context, driver='fake_ilo',
driver_info=db_utils.get_test_ilo_info())
[docs] def test_parse_driver_info(self):
info = ilo_common.parse_driver_info(self.node)
self.assertIsNotNone(info.get('ilo_address'))
self.assertIsNotNone(info.get('ilo_username'))
self.assertIsNotNone(info.get('ilo_password'))
self.assertIsNotNone(info.get('client_timeout'))
self.assertIsNotNone(info.get('client_port'))
[docs] def test_parse_driver_info_missing_address(self):
del self.node.driver_info['ilo_address']
self.assertRaises(exception.MissingParameterValue,
ilo_common.parse_driver_info, self.node)
[docs] def test_parse_driver_info_missing_username(self):
del self.node.driver_info['ilo_username']
self.assertRaises(exception.MissingParameterValue,
ilo_common.parse_driver_info, self.node)
[docs] def test_parse_driver_info_missing_password(self):
del self.node.driver_info['ilo_password']
self.assertRaises(exception.MissingParameterValue,
ilo_common.parse_driver_info, self.node)
[docs] def test_parse_driver_info_invalid_timeout(self):
self.node.driver_info['client_timeout'] = 'qwe'
self.assertRaises(exception.InvalidParameterValue,
ilo_common.parse_driver_info, self.node)
[docs] def test_parse_driver_info_invalid_port(self):
self.node.driver_info['client_port'] = 'qwe'
self.assertRaises(exception.InvalidParameterValue,
ilo_common.parse_driver_info, self.node)
self.node.driver_info['client_port'] = '65536'
self.assertRaises(exception.InvalidParameterValue,
ilo_common.parse_driver_info, self.node)
self.node.driver_info['console_port'] = 'invalid'
self.assertRaises(exception.InvalidParameterValue,
ilo_common.parse_driver_info, self.node)
self.node.driver_info['console_port'] = '-1'
self.assertRaises(exception.InvalidParameterValue,
ilo_common.parse_driver_info, self.node)
[docs] def test_parse_driver_info_missing_multiple_params(self):
del self.node.driver_info['ilo_password']
del self.node.driver_info['ilo_address']
try:
ilo_common.parse_driver_info(self.node)
self.fail("parse_driver_info did not throw exception.")
except exception.MissingParameterValue as e:
self.assertIn('ilo_password', str(e))
self.assertIn('ilo_address', str(e))
[docs] def test_parse_driver_info_invalid_multiple_params(self):
self.node.driver_info['client_timeout'] = 'qwe'
try:
ilo_common.parse_driver_info(self.node)
self.fail("parse_driver_info did not throw exception.")
except exception.InvalidParameterValue as e:
self.assertIn('client_timeout', str(e))
[docs]class IloCommonMethodsTestCase(db_base.DbTestCase):
[docs] def setUp(self):
super(IloCommonMethodsTestCase, self).setUp()
mgr_utils.mock_the_extension_manager(driver="fake_ilo")
self.info = db_utils.get_test_ilo_info()
self.node = obj_utils.create_test_node(
self.context, driver='fake_ilo', driver_info=self.info)
@mock.patch.object(ilo_client, 'IloClient', spec_set=True,
autospec=True)
[docs] def test_get_ilo_object(self, ilo_client_mock):
self.info['client_timeout'] = 60
self.info['client_port'] = 443
ilo_client_mock.return_value = 'ilo_object'
returned_ilo_object = ilo_common.get_ilo_object(self.node)
ilo_client_mock.assert_called_with(
self.info['ilo_address'],
self.info['ilo_username'],
self.info['ilo_password'],
self.info['client_timeout'],
self.info['client_port'])
self.assertEqual('ilo_object', returned_ilo_object)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_get_ilo_license(self, get_ilo_object_mock):
ilo_advanced_license = {'LICENSE_TYPE': 'iLO 3 Advanced'}
ilo_standard_license = {'LICENSE_TYPE': 'iLO 3'}
ilo_mock_object = get_ilo_object_mock.return_value
ilo_mock_object.get_all_licenses.return_value = ilo_advanced_license
license = ilo_common.get_ilo_license(self.node)
self.assertEqual(ilo_common.ADVANCED_LICENSE, license)
ilo_mock_object.get_all_licenses.return_value = ilo_standard_license
license = ilo_common.get_ilo_license(self.node)
self.assertEqual(ilo_common.STANDARD_LICENSE, license)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_get_ilo_license_fail(self, get_ilo_object_mock):
ilo_mock_object = get_ilo_object_mock.return_value
exc = ilo_error.IloError('error')
ilo_mock_object.get_all_licenses.side_effect = exc
self.assertRaises(exception.IloOperationError,
ilo_common.get_ilo_license,
self.node)
[docs] def test_update_ipmi_properties(self):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ipmi_info = {
"ipmi_address": "1.2.3.4",
"ipmi_username": "admin",
"ipmi_password": "fake",
"ipmi_terminal_port": 60
}
self.info['console_port'] = 60
task.node.driver_info = self.info
ilo_common.update_ipmi_properties(task)
actual_info = task.node.driver_info
expected_info = dict(self.info, **ipmi_info)
self.assertEqual(expected_info, actual_info)
[docs] def test__get_floppy_image_name(self):
image_name_expected = 'image-' + self.node.uuid
image_name_actual = ilo_common._get_floppy_image_name(self.node)
self.assertEqual(image_name_expected, image_name_actual)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
@mock.patch.object(images, 'create_vfat_image', spec_set=True,
autospec=True)
@mock.patch.object(tempfile, 'NamedTemporaryFile', spec_set=True,
autospec=True)
[docs] def test__prepare_floppy_image(self, tempfile_mock, fatimage_mock,
swift_api_mock):
mock_image_file_handle = mock.MagicMock(spec=file)
mock_image_file_obj = mock.MagicMock(spec=file)
mock_image_file_obj.name = 'image-tmp-file'
mock_image_file_handle.__enter__.return_value = mock_image_file_obj
tempfile_mock.return_value = mock_image_file_handle
swift_obj_mock = swift_api_mock.return_value
self.config(swift_ilo_container='ilo_cont', group='ilo')
self.config(swift_object_expiry_timeout=1, group='ilo')
deploy_args = {'arg1': 'val1', 'arg2': 'val2'}
swift_obj_mock.get_temp_url.return_value = 'temp-url'
timeout = CONF.ilo.swift_object_expiry_timeout
object_headers = {'X-Delete-After': timeout}
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
temp_url = ilo_common._prepare_floppy_image(task, deploy_args)
node_uuid = task.node.uuid
object_name = 'image-' + node_uuid
fatimage_mock.assert_called_once_with('image-tmp-file',
parameters=deploy_args)
swift_obj_mock.create_object.assert_called_once_with(
'ilo_cont', object_name, 'image-tmp-file',
object_headers=object_headers)
swift_obj_mock.get_temp_url.assert_called_once_with(
'ilo_cont', object_name, timeout)
self.assertEqual('temp-url', temp_url)
@mock.patch.object(ilo_common, 'copy_image_to_web_server',
spec_set=True, autospec=True)
@mock.patch.object(images, 'create_vfat_image', spec_set=True,
autospec=True)
@mock.patch.object(tempfile, 'NamedTemporaryFile', spec_set=True,
autospec=True)
[docs] def test__prepare_floppy_image_use_webserver(self, tempfile_mock,
fatimage_mock,
copy_mock):
mock_image_file_handle = mock.MagicMock(spec=file)
mock_image_file_obj = mock.MagicMock(spec=file)
mock_image_file_obj.name = 'image-tmp-file'
mock_image_file_handle.__enter__.return_value = mock_image_file_obj
tempfile_mock.return_value = mock_image_file_handle
self.config(use_web_server_for_images=True, group='ilo')
deploy_args = {'arg1': 'val1', 'arg2': 'val2'}
CONF.deploy.http_url = "http://abc.com/httpboot"
CONF.deploy.http_root = "/httpboot"
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
node_uuid = task.node.uuid
object_name = 'image-' + node_uuid
http_url = CONF.deploy.http_url + '/' + object_name
copy_mock.return_value = "http://abc.com/httpboot/" + object_name
temp_url = ilo_common._prepare_floppy_image(task, deploy_args)
fatimage_mock.assert_called_once_with('image-tmp-file',
parameters=deploy_args)
copy_mock.assert_called_once_with('image-tmp-file', object_name)
self.assertEqual(http_url, temp_url)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_set_boot_mode(self, get_ilo_object_mock):
ilo_object_mock = get_ilo_object_mock.return_value
get_pending_boot_mode_mock = ilo_object_mock.get_pending_boot_mode
set_pending_boot_mode_mock = ilo_object_mock.set_pending_boot_mode
get_pending_boot_mode_mock.return_value = 'LEGACY'
ilo_common.set_boot_mode(self.node, 'uefi')
get_ilo_object_mock.assert_called_once_with(self.node)
get_pending_boot_mode_mock.assert_called_once_with()
set_pending_boot_mode_mock.assert_called_once_with('UEFI')
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_set_boot_mode_without_set_pending_boot_mode(self,
get_ilo_object_mock):
ilo_object_mock = get_ilo_object_mock.return_value
get_pending_boot_mode_mock = ilo_object_mock.get_pending_boot_mode
get_pending_boot_mode_mock.return_value = 'LEGACY'
ilo_common.set_boot_mode(self.node, 'bios')
get_ilo_object_mock.assert_called_once_with(self.node)
get_pending_boot_mode_mock.assert_called_once_with()
self.assertFalse(ilo_object_mock.set_pending_boot_mode.called)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_set_boot_mode_with_IloOperationError(self,
get_ilo_object_mock):
ilo_object_mock = get_ilo_object_mock.return_value
get_pending_boot_mode_mock = ilo_object_mock.get_pending_boot_mode
get_pending_boot_mode_mock.return_value = 'UEFI'
set_pending_boot_mode_mock = ilo_object_mock.set_pending_boot_mode
exc = ilo_error.IloError('error')
set_pending_boot_mode_mock.side_effect = exc
self.assertRaises(exception.IloOperationError,
ilo_common.set_boot_mode, self.node, 'bios')
get_ilo_object_mock.assert_called_once_with(self.node)
get_pending_boot_mode_mock.assert_called_once_with()
@mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode_instance_info_exists(self,
set_boot_mode_mock):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.node.instance_info['deploy_boot_mode'] = 'bios'
ilo_common.update_boot_mode(task)
set_boot_mode_mock.assert_called_once_with(task.node, 'bios')
@mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode_capabilities_exist(self,
set_boot_mode_mock):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.node.properties['capabilities'] = 'boot_mode:bios'
ilo_common.update_boot_mode(task)
set_boot_mode_mock.assert_called_once_with(task.node, 'bios')
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode(self, get_ilo_object_mock):
ilo_mock_obj = get_ilo_object_mock.return_value
ilo_mock_obj.get_pending_boot_mode.return_value = 'LEGACY'
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ilo_common.update_boot_mode(task)
get_ilo_object_mock.assert_called_once_with(task.node)
ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
self.assertEqual('bios',
task.node.instance_info['deploy_boot_mode'])
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode_unknown(self,
get_ilo_object_mock):
ilo_mock_obj = get_ilo_object_mock.return_value
ilo_mock_obj.get_pending_boot_mode.return_value = 'UNKNOWN'
set_pending_boot_mode_mock = ilo_mock_obj.set_pending_boot_mode
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ilo_common.update_boot_mode(task)
get_ilo_object_mock.assert_called_once_with(task.node)
ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
set_pending_boot_mode_mock.assert_called_once_with('UEFI')
self.assertEqual('uefi',
task.node.instance_info['deploy_boot_mode'])
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode_unknown_except(self,
get_ilo_object_mock):
ilo_mock_obj = get_ilo_object_mock.return_value
ilo_mock_obj.get_pending_boot_mode.return_value = 'UNKNOWN'
set_pending_boot_mode_mock = ilo_mock_obj.set_pending_boot_mode
exc = ilo_error.IloError('error')
set_pending_boot_mode_mock.side_effect = exc
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.IloOperationError,
ilo_common.update_boot_mode, task)
get_ilo_object_mock.assert_called_once_with(task.node)
ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode_legacy(self,
get_ilo_object_mock):
ilo_mock_obj = get_ilo_object_mock.return_value
exc = ilo_error.IloCommandNotSupportedError('error')
ilo_mock_obj.get_pending_boot_mode.side_effect = exc
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ilo_common.update_boot_mode(task)
get_ilo_object_mock.assert_called_once_with(task.node)
ilo_mock_obj.get_pending_boot_mode.assert_called_once_with()
self.assertEqual('bios',
task.node.instance_info['deploy_boot_mode'])
@mock.patch.object(ilo_common, 'set_boot_mode', spec_set=True,
autospec=True)
[docs] def test_update_boot_mode_prop_boot_mode_exist(self,
set_boot_mode_mock):
properties = {'capabilities': 'boot_mode:uefi'}
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
task.node.properties = properties
ilo_common.update_boot_mode(task)
set_boot_mode_mock.assert_called_once_with(task.node, 'uefi')
@mock.patch.object(images, 'get_temp_url_for_glance_image', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, '_prepare_floppy_image', spec_set=True,
autospec=True)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
@mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'attach_vmedia', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'eject_vmedia_devices',
spec_set=True, autospec=True)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
@mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common.LOG, 'exception', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'eject_vmedia_devices',
spec_set=True, autospec=True)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
@mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common.LOG, 'warning', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'eject_vmedia_devices',
spec_set=True, autospec=True)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
@mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'eject_vmedia_devices',
spec_set=True, autospec=True)
@mock.patch.object(ilo_common, 'destroy_floppy_image_from_web_server',
spec_set=True, autospec=True)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_get_secure_boot_mode(self,
get_ilo_object_mock):
ilo_object_mock = get_ilo_object_mock.return_value
ilo_object_mock.get_current_boot_mode.return_value = 'UEFI'
ilo_object_mock.get_secure_boot_mode.return_value = True
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ret = ilo_common.get_secure_boot_mode(task)
ilo_object_mock.get_current_boot_mode.assert_called_once_with()
ilo_object_mock.get_secure_boot_mode.assert_called_once_with()
self.assertTrue(ret)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_get_secure_boot_mode_bios(self,
get_ilo_object_mock):
ilo_object_mock = get_ilo_object_mock.return_value
ilo_object_mock.get_current_boot_mode.return_value = 'BIOS'
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ret = ilo_common.get_secure_boot_mode(task)
ilo_object_mock.get_current_boot_mode.assert_called_once_with()
self.assertFalse(ilo_object_mock.get_secure_boot_mode.called)
self.assertFalse(ret)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_get_secure_boot_mode_fail(self,
get_ilo_object_mock):
ilo_mock_object = get_ilo_object_mock.return_value
exc = ilo_error.IloError('error')
ilo_mock_object.get_current_boot_mode.side_effect = exc
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.IloOperationError,
ilo_common.get_secure_boot_mode,
task)
ilo_mock_object.get_current_boot_mode.assert_called_once_with()
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_get_secure_boot_mode_not_supported(self,
ilo_object_mock):
ilo_mock_object = ilo_object_mock.return_value
exc = ilo_error.IloCommandNotSupportedError('error')
ilo_mock_object.get_current_boot_mode.return_value = 'UEFI'
ilo_mock_object.get_secure_boot_mode.side_effect = exc
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.IloOperationNotSupported,
ilo_common.get_secure_boot_mode,
task)
ilo_mock_object.get_current_boot_mode.assert_called_once_with()
ilo_mock_object.get_secure_boot_mode.assert_called_once_with()
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_set_secure_boot_mode(self,
get_ilo_object_mock):
ilo_object_mock = get_ilo_object_mock.return_value
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ilo_common.set_secure_boot_mode(task, True)
ilo_object_mock.set_secure_boot_mode.assert_called_once_with(True)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_set_secure_boot_mode_fail(self,
get_ilo_object_mock):
ilo_mock_object = get_ilo_object_mock.return_value
exc = ilo_error.IloError('error')
ilo_mock_object.set_secure_boot_mode.side_effect = exc
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.IloOperationError,
ilo_common.set_secure_boot_mode,
task, False)
ilo_mock_object.set_secure_boot_mode.assert_called_once_with(False)
@mock.patch.object(ilo_common, 'get_ilo_object', spec_set=True,
autospec=True)
[docs] def test_set_secure_boot_mode_not_supported(self,
ilo_object_mock):
ilo_mock_object = ilo_object_mock.return_value
exc = ilo_error.IloCommandNotSupportedError('error')
ilo_mock_object.set_secure_boot_mode.side_effect = exc
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
self.assertRaises(exception.IloOperationNotSupported,
ilo_common.set_secure_boot_mode,
task, False)
ilo_mock_object.set_secure_boot_mode.assert_called_once_with(False)
@mock.patch.object(os, 'chmod', spec_set=True,
autospec=True)
@mock.patch.object(shutil, 'copyfile', spec_set=True,
autospec=True)
[docs] def test_copy_image_to_web_server(self, copy_mock,
chmod_mock):
CONF.deploy.http_url = "http://x.y.z.a/webserver/"
CONF.deploy.http_root = "/webserver"
expected_url = "http://x.y.z.a/webserver/image-UUID"
source = 'tmp_image_file'
destination = "image-UUID"
image_path = "/webserver/image-UUID"
actual_url = ilo_common.copy_image_to_web_server(source, destination)
self.assertEqual(expected_url, actual_url)
copy_mock.assert_called_once_with(source, image_path)
chmod_mock.assert_called_once_with(image_path, 0o644)
@mock.patch.object(os, 'chmod', spec_set=True,
autospec=True)
@mock.patch.object(shutil, 'copyfile', spec_set=True,
autospec=True)
[docs] def test_copy_image_to_web_server_fails(self, copy_mock,
chmod_mock):
CONF.deploy.http_url = "http://x.y.z.a/webserver/"
CONF.deploy.http_root = "/webserver"
source = 'tmp_image_file'
destination = "image-UUID"
image_path = "/webserver/image-UUID"
exc = exception.ImageUploadFailed('reason')
copy_mock.side_effect = exc
self.assertRaises(exception.ImageUploadFailed,
ilo_common.copy_image_to_web_server,
source, destination)
copy_mock.assert_called_once_with(source, image_path)
self.assertFalse(chmod_mock.called)
@mock.patch.object(ilo_common, 'ironic_utils', autospec=True)
[docs] def test_remove_image_from_web_server(self, utils_mock):
# | GIVEN |
CONF.deploy.http_url = "http://x.y.z.a/webserver/"
CONF.deploy.http_root = "/webserver"
object_name = 'tmp_image_file'
# | WHEN |
ilo_common.remove_image_from_web_server(object_name)
# | THEN |
(utils_mock.unlink_without_raise.
assert_called_once_with("/webserver/tmp_image_file"))
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
@mock.patch.object(ilo_common, 'LOG')
[docs] def test_copy_image_to_swift(self, LOG_mock, swift_api_mock):
# | GIVEN |
self.config(swift_ilo_container='ilo_container', group='ilo')
self.config(swift_object_expiry_timeout=1, group='ilo')
container = CONF.ilo.swift_ilo_container
timeout = CONF.ilo.swift_object_expiry_timeout
swift_obj_mock = swift_api_mock.return_value
destination_object_name = 'destination_object_name'
source_file_path = 'tmp_image_file'
object_headers = {'X-Delete-After': timeout}
# | WHEN |
ilo_common.copy_image_to_swift(source_file_path,
destination_object_name)
# | THEN |
swift_obj_mock.create_object.assert_called_once_with(
container, destination_object_name, source_file_path,
object_headers=object_headers)
swift_obj_mock.get_temp_url.assert_called_once_with(
container, destination_object_name, timeout)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
[docs] def test_copy_image_to_swift_throws_error_if_swift_operation_fails(
self, swift_api_mock):
# | GIVEN |
self.config(swift_ilo_container='ilo_container', group='ilo')
self.config(swift_object_expiry_timeout=1, group='ilo')
swift_obj_mock = swift_api_mock.return_value
destination_object_name = 'destination_object_name'
source_file_path = 'tmp_image_file'
swift_obj_mock.create_object.side_effect = (
exception.SwiftOperationError(operation='create_object',
error='failed'))
# | WHEN | & | THEN |
self.assertRaises(exception.SwiftOperationError,
ilo_common.copy_image_to_swift,
source_file_path, destination_object_name)
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
[docs] def test_remove_image_from_swift(self, swift_api_mock):
# | GIVEN |
self.config(swift_ilo_container='ilo_container', group='ilo')
container = CONF.ilo.swift_ilo_container
swift_obj_mock = swift_api_mock.return_value
object_name = 'object_name'
# | WHEN |
ilo_common.remove_image_from_swift(object_name)
# | THEN |
swift_obj_mock.delete_object.assert_called_once_with(
container, object_name)
@mock.patch.object(ilo_common, 'LOG')
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
[docs] def test_remove_image_from_swift_suppresses_notfound_exc(
self, swift_api_mock, LOG_mock):
# | GIVEN |
self.config(swift_ilo_container='ilo_container', group='ilo')
container = CONF.ilo.swift_ilo_container
swift_obj_mock = swift_api_mock.return_value
object_name = 'object_name'
raised_exc = exception.SwiftObjectNotFoundError(
operation='delete_object', object=object_name, container=container)
swift_obj_mock.delete_object.side_effect = raised_exc
# | WHEN |
ilo_common.remove_image_from_swift(object_name)
# | THEN |
LOG_mock.warning.assert_called_once_with(
mock.ANY, {'associated_with_msg': "", 'err': raised_exc})
@mock.patch.object(ilo_common, 'LOG')
@mock.patch.object(swift, 'SwiftAPI', spec_set=True, autospec=True)
[docs] def test_remove_image_from_swift_suppresses_operror_exc(
self, swift_api_mock, LOG_mock):
# | GIVEN |
self.config(swift_ilo_container='ilo_container', group='ilo')
container = CONF.ilo.swift_ilo_container
swift_obj_mock = swift_api_mock.return_value
object_name = 'object_name'
raised_exc = exception.SwiftOperationError(operation='delete_object',
error='failed')
swift_obj_mock.delete_object.side_effect = raised_exc
# | WHEN |
ilo_common.remove_image_from_swift(object_name, 'alice_in_wonderland')
# | THEN |
LOG_mock.exception.assert_called_once_with(
mock.ANY, {'object_name': object_name, 'container': container,
'associated_with_msg': ("associated with "
"alice_in_wonderland"),
'err': raised_exc})
@mock.patch.object(ironic_utils, 'unlink_without_raise', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, '_get_floppy_image_name', spec_set=True,
autospec=True)
[docs] def test_destroy_floppy_image_from_web_server(self, get_floppy_name_mock,
utils_mock):
get_floppy_name_mock.return_value = 'image-uuid'
CONF.deploy.http_root = "/webserver/"
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
ilo_common.destroy_floppy_image_from_web_server(task.node)
get_floppy_name_mock.assert_called_once_with(task.node)
utils_mock.assert_called_once_with('/webserver/image-uuid')
@mock.patch.object(manager_utils, 'node_set_boot_device', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'setup_vmedia_for_boot', spec_set=True,
autospec=True)
@mock.patch.object(deploy_utils, 'is_secure_boot_requested', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'set_secure_boot_mode', spec_set=True,
autospec=True)
[docs] def test_update_secure_boot_mode_passed_true(self,
func_set_secure_boot_mode,
func_is_secure_boot_req):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
func_is_secure_boot_req.return_value = True
ilo_common.update_secure_boot_mode(task, True)
func_set_secure_boot_mode.assert_called_once_with(task, True)
@mock.patch.object(deploy_utils, 'is_secure_boot_requested', spec_set=True,
autospec=True)
@mock.patch.object(ilo_common, 'set_secure_boot_mode', spec_set=True,
autospec=True)
[docs] def test_update_secure_boot_mode_passed_false(self,
func_set_secure_boot_mode,
func_is_secure_boot_req):
with task_manager.acquire(self.context, self.node.uuid,
shared=False) as task:
func_is_secure_boot_req.return_value = False
ilo_common.update_secure_boot_mode(task, False)
self.assertFalse(func_set_secure_boot_mode.called)
@mock.patch.object(ironic_utils, 'unlink_without_raise', spec_set=True,
autospec=True)
[docs] def test_remove_single_or_list_of_files_with_file_list(self, unlink_mock):
# | GIVEN |
file_list = ['/any_path1/any_file1',
'/any_path2/any_file2',
'/any_path3/any_file3']
# | WHEN |
ilo_common.remove_single_or_list_of_files(file_list)
# | THEN |
calls = [mock.call('/any_path1/any_file1'),
mock.call('/any_path2/any_file2'),
mock.call('/any_path3/any_file3')]
unlink_mock.assert_has_calls(calls)
@mock.patch.object(ironic_utils, 'unlink_without_raise', spec_set=True,
autospec=True)
[docs] def test_remove_single_or_list_of_files_with_file_str(self, unlink_mock):
# | GIVEN |
file_path = '/any_path1/any_file'
# | WHEN |
ilo_common.remove_single_or_list_of_files(file_path)
# | THEN |
unlink_mock.assert_called_once_with('/any_path1/any_file')
@mock.patch.object(__builtin__, 'open', autospec=True)
[docs] def test_verify_image_checksum(self, open_mock):
# | GIVEN |
data = b'Yankee Doodle went to town riding on a pony;'
file_like_object = six.BytesIO(data)
open_mock().__enter__.return_value = file_like_object
actual_hash = hashlib.md5(data).hexdigest()
# | WHEN |
ilo_common.verify_image_checksum(file_like_object, actual_hash)
# | THEN |
# no any exception thrown
[docs] def test_verify_image_checksum_throws_for_nonexistent_file(self):
# | GIVEN |
invalid_file_path = '/some/invalid/file/path'
# | WHEN | & | THEN |
self.assertRaises(exception.ImageRefValidationFailed,
ilo_common.verify_image_checksum,
invalid_file_path, 'hash_xxx')
@mock.patch.object(__builtin__, 'open', autospec=True)
[docs] def test_verify_image_checksum_throws_for_failed_validation(self,
open_mock):
# | GIVEN |
data = b'Yankee Doodle went to town riding on a pony;'
file_like_object = six.BytesIO(data)
open_mock().__enter__.return_value = file_like_object
invalid_hash = 'invalid_hash_value'
# | WHEN | & | THEN |
self.assertRaises(exception.ImageRefValidationFailed,
ilo_common.verify_image_checksum,
file_like_object,
invalid_hash)