565 lines
25 KiB
Python
565 lines
25 KiB
Python
# Copyright 2016 Hewlett Packard Enterprise 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 Firmware controller."""
|
|
import distutils.spawn
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
|
|
import ddt
|
|
import mock
|
|
from six.moves import builtins as __builtin__
|
|
|
|
from proliantutils import exception
|
|
from proliantutils.ilo import common
|
|
from proliantutils.ilo import firmware_controller
|
|
|
|
|
|
@ddt.ddt
|
|
class FirmwareControllerModuleTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
# | BEFORE_EACH |
|
|
super(FirmwareControllerModuleTestCase, self).setUp()
|
|
self.any_scexe_file = 'any_file.scexe'
|
|
self.any_rpm_file = 'any_file.rpm'
|
|
self.any_raw_fw_file = 'any_fw_file.bin'
|
|
|
|
@ddt.data('absolute_path_to_exe', None)
|
|
@mock.patch.object(firmware_controller, 'shutil', spec=['which'])
|
|
@mock.patch.object(distutils.spawn, 'find_executable', autospec=True)
|
|
def test_find_executable_returns(self, expected_ret_val,
|
|
find_executable_mock, shutil_mock):
|
|
# | GIVEN |
|
|
find_executable_mock.return_value = expected_ret_val
|
|
shutil_mock.which.return_value = expected_ret_val
|
|
# | WHEN |
|
|
actual_ret_val = firmware_controller.find_executable('executable')
|
|
# | THEN |
|
|
self.assertEqual(expected_ret_val, actual_ret_val)
|
|
|
|
@ddt.data('ilo', 'cpld', 'power_pic', 'bios', 'chassis')
|
|
def test_check_firmware_update_component_passes_for_valid_component(
|
|
self, component_type):
|
|
# | GIVEN |
|
|
ris_or_ribcl_obj_mock = mock.MagicMock()
|
|
update_firmware_mock = mock.MagicMock()
|
|
# Note(deray): Need to set __name__ attribute explicitly to keep
|
|
# ``six.wraps`` happy. Passing this to the `name` argument at the time
|
|
# creation of Mock doesn't help.
|
|
update_firmware_mock.__name__ = 'update_firmware_mock'
|
|
wrapped_func = (firmware_controller.
|
|
check_firmware_update_component(update_firmware_mock))
|
|
# | WHEN |
|
|
wrapped_func(
|
|
ris_or_ribcl_obj_mock, self.any_raw_fw_file, component_type)
|
|
# | THEN |
|
|
update_firmware_mock.assert_called_once_with(
|
|
ris_or_ribcl_obj_mock, self.any_raw_fw_file, component_type)
|
|
|
|
def test_check_firmware_update_component_throws_for_invalid_component(
|
|
self):
|
|
# | GIVEN |
|
|
def func(ris_or_ribcl_obj, filename, component_type):
|
|
pass
|
|
|
|
wrapped_func = (firmware_controller.
|
|
check_firmware_update_component(func))
|
|
ris_or_ribcl_obj_mock = mock.MagicMock()
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(exception.InvalidInputError,
|
|
wrapped_func,
|
|
ris_or_ribcl_obj_mock,
|
|
self.any_raw_fw_file,
|
|
'invalid_component')
|
|
|
|
def test_get_fw_extractor_will_set_the_fw_file_attribute(self):
|
|
# | WHEN |
|
|
fw_img_extractor = (firmware_controller.
|
|
get_fw_extractor(self.any_scexe_file))
|
|
# | THEN |
|
|
self.assertEqual(self.any_scexe_file, fw_img_extractor.fw_file)
|
|
|
|
@mock.patch.object(firmware_controller, '_extract_scexe_file',
|
|
autospec=True)
|
|
def test__extract_scexe_file_gets_invoked_for_scexe_firmware_file(
|
|
self, _extract_scexe_file_mock):
|
|
# _extract_scexe_file gets invoked when fw_img_extractor is initialized
|
|
# with scexe firmware file
|
|
# | WHEN |
|
|
fw_img_extractor = (firmware_controller.
|
|
get_fw_extractor(self.any_scexe_file))
|
|
fw_img_extractor._do_extract('some_target_file', 'some_extract_path')
|
|
# | THEN |
|
|
_extract_scexe_file_mock.assert_called_once_with(
|
|
fw_img_extractor, 'some_target_file', 'some_extract_path')
|
|
|
|
@mock.patch.object(firmware_controller, '_extract_rpm_file', autospec=True)
|
|
def test__extract_rpm_file_gets_invoked_for_rpm_firmware_file(
|
|
self, _extract_rpm_file_mock):
|
|
# _extract_rpm_file gets invoked when fw_img_extractor is initialized
|
|
# with rpm firmware file
|
|
# | WHEN |
|
|
fw_img_extractor = (firmware_controller.
|
|
get_fw_extractor(self.any_rpm_file))
|
|
fw_img_extractor._do_extract('some_target_file', 'some_extract_path')
|
|
# | THEN |
|
|
_extract_rpm_file_mock.assert_called_once_with(
|
|
fw_img_extractor, 'some_target_file', 'some_extract_path')
|
|
|
|
@ddt.data('any_file.hex', 'any_file.bin', 'any_file.vme', 'any_file.flash')
|
|
def test_no_op_extract_gets_invoked_for_raw_firmware_file(self,
|
|
any_raw_file):
|
|
# no_op extract when fw_img_extractor is initialized
|
|
# with raw firmware file
|
|
# | WHEN |
|
|
fw_img_extractor = firmware_controller.get_fw_extractor(any_raw_file)
|
|
return_result, is_extracted = fw_img_extractor.extract()
|
|
# | THEN |
|
|
self.assertEqual(any_raw_file, return_result)
|
|
self.assertFalse(is_extracted)
|
|
|
|
def test_get_fw_extractor_raises_exception_with_unknown_firmware_file(
|
|
self):
|
|
# | GIVEN |
|
|
any_invalid_format_firmware_file = 'any_file.abc'
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(exception.InvalidInputError,
|
|
firmware_controller.get_fw_extractor,
|
|
any_invalid_format_firmware_file)
|
|
|
|
@mock.patch.object(firmware_controller.utils, 'trycmd', autospec=True)
|
|
def test__extract_scexe_file_issues_command_as(self, utils_trycmd_mock):
|
|
# | GIVEN |
|
|
any_scexe_firmware_file = 'any_file.scexe'
|
|
any_extract_path = 'any_extract_path'
|
|
utils_trycmd_mock.return_value = ('out', 'err')
|
|
# | WHEN |
|
|
firmware_controller._extract_scexe_file(
|
|
None, any_scexe_firmware_file, any_extract_path)
|
|
# | THEN |
|
|
utils_trycmd_mock.assert_called_once_with(
|
|
any_scexe_firmware_file, '--unpack=' + any_extract_path)
|
|
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'subprocess', autospec=True)
|
|
@mock.patch.object(
|
|
firmware_controller, 'find_executable', lambda x: mock.ANY)
|
|
def test__extract_rpm_file_creates_dir_if_extract_path_doesnt_exist(
|
|
self, subprocess_mock, os_mock):
|
|
# | GIVEN |
|
|
any_rpm_firmware_file = 'any_file.rpm'
|
|
any_extract_path = 'any_extract_path'
|
|
|
|
os_mock.path.exists.return_value = False
|
|
mock_cpio = mock.MagicMock()
|
|
mock_cpio.communicate.return_value = ('out', 'err')
|
|
subsequent_popen_call_returns = [mock.MagicMock(), mock_cpio]
|
|
subprocess_mock.Popen = mock.MagicMock(
|
|
side_effect=subsequent_popen_call_returns)
|
|
# | WHEN |
|
|
firmware_controller._extract_rpm_file(
|
|
None, any_rpm_firmware_file, any_extract_path)
|
|
# | THEN |
|
|
os_mock.makedirs.assert_called_once_with(any_extract_path)
|
|
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'subprocess', autospec=True)
|
|
@mock.patch.object(
|
|
firmware_controller, 'find_executable', lambda x: mock.ANY)
|
|
def test__extract_rpm_file_doesnt_create_dir_if_extract_path_present(
|
|
self, subprocess_mock, os_mock):
|
|
# extract_rpm_file doesn't create dir if extract path
|
|
# is already present
|
|
# | GIVEN |
|
|
any_rpm_firmware_file = 'any_file.rpm'
|
|
any_extract_path = 'any_extract_path'
|
|
|
|
os_mock.path.exists.return_value = True
|
|
mock_cpio = mock.MagicMock()
|
|
mock_cpio.communicate.return_value = ('out', 'err')
|
|
subsequent_popen_call_returns = [mock.MagicMock(), mock_cpio]
|
|
subprocess_mock.Popen = mock.MagicMock(
|
|
side_effect=subsequent_popen_call_returns)
|
|
# | WHEN |
|
|
firmware_controller._extract_rpm_file(
|
|
None, any_rpm_firmware_file, any_extract_path)
|
|
# | THEN |
|
|
self.assertFalse(os_mock.makedirs.called)
|
|
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'subprocess', autospec=True)
|
|
def test__extract_rpm_file_raises_exception_if_rpm_or_cpio_exec_not_found(
|
|
self, subprocess_mock, os_mock):
|
|
# | GIVEN |
|
|
any_rpm_firmware_file = 'any_file.rpm'
|
|
any_extract_path = 'any_extract_path'
|
|
return_values_for_find_executable_stub = (
|
|
[None, 'absolute_path_to_cpio'], # `rpm2cpio` not present
|
|
['absolute_path_to_rpm2cpio', None] # `cpio` not present
|
|
)
|
|
|
|
# | WHEN | & | THEN |
|
|
for input in return_values_for_find_executable_stub:
|
|
with mock.patch.object(
|
|
firmware_controller, 'find_executable') as find_mock:
|
|
find_mock.side_effect = input
|
|
self.assertRaises(exception.ImageExtractionFailed,
|
|
firmware_controller._extract_rpm_file, None,
|
|
any_rpm_firmware_file, any_extract_path)
|
|
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'subprocess', autospec=True)
|
|
@mock.patch.object(
|
|
firmware_controller, 'find_executable', lambda x: mock.ANY)
|
|
def test__extract_rpm_file_issues_commands_as(self,
|
|
subprocess_mock,
|
|
os_mock):
|
|
# | GIVEN |
|
|
any_rpm_firmware_file = 'any_file.rpm'
|
|
any_extract_path = 'any_extract_path'
|
|
|
|
os_mock.path.exists.return_value = True
|
|
rpm2cpio_mock = mock.MagicMock()
|
|
cpio_mock = mock.MagicMock()
|
|
cpio_mock.communicate.return_value = ('out', 'err')
|
|
subsequent_popen_call_returns = [rpm2cpio_mock, cpio_mock]
|
|
subprocess_mock.Popen = mock.MagicMock(
|
|
side_effect=subsequent_popen_call_returns)
|
|
# | WHEN |
|
|
firmware_controller._extract_rpm_file(
|
|
None, any_rpm_firmware_file, any_extract_path)
|
|
# | THEN |
|
|
popen_calls_to_assert = [
|
|
mock.call.Popen('rpm2cpio ' + any_rpm_firmware_file,
|
|
shell=True,
|
|
stdout=subprocess_mock.PIPE),
|
|
mock.call.Popen('cpio -idm',
|
|
shell=True,
|
|
stdin=rpm2cpio_mock.stdout),
|
|
]
|
|
subprocess_mock.assert_has_calls(popen_calls_to_assert)
|
|
cpio_mock.communicate.assert_called_once_with()
|
|
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'subprocess', autospec=True)
|
|
@mock.patch.object(
|
|
firmware_controller, 'find_executable', lambda x: mock.ANY)
|
|
def test__extract_rpm_file_raises_exception_if_it_fails(self,
|
|
subprocess_mock,
|
|
os_mock):
|
|
# | GIVEN |
|
|
any_rpm_firmware_file = 'any_file.rpm'
|
|
any_extract_path = 'any_extract_path'
|
|
|
|
rpm2cpio_mock = mock.MagicMock()
|
|
cpio_mock = mock.MagicMock()
|
|
cpio_mock.communicate.side_effect = ValueError
|
|
subsequent_popen_call_returns = [rpm2cpio_mock, cpio_mock]
|
|
subprocess_mock.Popen = mock.MagicMock(
|
|
side_effect=subsequent_popen_call_returns)
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(exception.ImageExtractionFailed,
|
|
firmware_controller._extract_rpm_file,
|
|
None, any_rpm_firmware_file, any_extract_path)
|
|
|
|
def test__get_firmware_file(self):
|
|
# | GIVEN |
|
|
temp_dir_setup = setup_fixture_create_fw_file_extracts_for('scexe')
|
|
# | WHEN |
|
|
return_result = firmware_controller._get_firmware_file(temp_dir_setup)
|
|
# | THEN |
|
|
self.assertTrue(return_result.endswith('.bin'))
|
|
teardown_fixture_create_fw_file_extracts_for(temp_dir_setup)
|
|
|
|
@mock.patch.object(
|
|
firmware_controller, '_get_firmware_file', autospec=True)
|
|
@mock.patch.object(common, 'get_filename_and_extension_of', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'tempfile', autospec=True)
|
|
@mock.patch.object(firmware_controller.uuid, 'uuid4', autospec=True)
|
|
@mock.patch.object(firmware_controller.os, 'link', autospec=True)
|
|
def test__get_firmware_file_in_new_path(
|
|
self, os_link_mock, uuid4_mock, tempfile_mock,
|
|
get_filename_and_extension_of_mock, _get_firmware_file_mock):
|
|
# | GIVEN |
|
|
_get_firmware_file_mock.return_value = 'some_raw_fw_file.bin'
|
|
get_filename_and_extension_of_mock.return_value = ('some_raw_fw_file',
|
|
'.bin')
|
|
tempfile_mock.gettempdir.return_value = '/tmp'
|
|
uuid4_mock.return_value = 12345
|
|
# | WHEN |
|
|
new_fw_file_path = (firmware_controller.
|
|
_get_firmware_file_in_new_path('any_path'))
|
|
# | THEN |
|
|
_get_firmware_file_mock.assert_called_once_with('any_path')
|
|
# tests the hard linking of the raw fw file so found
|
|
os_link_mock.assert_called_once_with(
|
|
'some_raw_fw_file.bin', '/tmp/12345_some_raw_fw_file.bin')
|
|
self.assertEqual(new_fw_file_path, '/tmp/12345_some_raw_fw_file.bin')
|
|
|
|
@mock.patch.object(
|
|
firmware_controller, '_get_firmware_file', autospec=True)
|
|
def test__get_firmware_file_in_new_path_returns_none_for_file_not_found(
|
|
self, _get_firmware_file_mock):
|
|
# | GIVEN |
|
|
_get_firmware_file_mock.return_value = None
|
|
# | WHEN |
|
|
actual_result = (firmware_controller.
|
|
_get_firmware_file_in_new_path('any_path'))
|
|
# | THEN |
|
|
self.assertIsNone(actual_result)
|
|
|
|
|
|
def teardown_fixture_create_fw_file_extracts_for(temp_dir):
|
|
# os.removedirs(temp_dir)
|
|
shutil.rmtree(temp_dir)
|
|
|
|
|
|
def setup_fixture_create_fw_file_extracts_for(format):
|
|
temp_dir = tempfile.mkdtemp()
|
|
fw_file_exts = [
|
|
'_ilo', '.bin', '.xml', '.TXT', '.hpsetup', '.cpq_package.inc'
|
|
]
|
|
|
|
if format == 'scexe':
|
|
fw_files_dir = temp_dir
|
|
elif format == 'rpm':
|
|
fw_files_dir = os.path.join(
|
|
temp_dir +
|
|
'/please_remove_rpm_file_extracts/usr/lib/i386-linux-gnu/' +
|
|
'hp-firmware-iloX-xxxx'
|
|
)
|
|
else:
|
|
fw_files_dir = temp_dir
|
|
|
|
if not os.path.exists(fw_files_dir):
|
|
os.makedirs(fw_files_dir)
|
|
|
|
for fw_file_ext in fw_file_exts:
|
|
tempfile.NamedTemporaryFile(suffix=fw_file_ext,
|
|
dir=fw_files_dir,
|
|
delete=False)
|
|
|
|
return temp_dir
|
|
|
|
|
|
class FirmwareImageExtractorTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
# | BEFORE_EACH |
|
|
self.any_scexe_file = 'any_file.scexe'
|
|
self.any_rpm_file = 'any_file.rpm'
|
|
|
|
@mock.patch.object(common, 'add_exec_permission_to', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'tempfile', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, '_get_firmware_file_in_new_path',
|
|
autospec=True)
|
|
@mock.patch.object(firmware_controller, 'shutil', autospec=True)
|
|
def test_extract_method_calls__do_extract_in_turn(
|
|
self, shutil_mock, _get_firmware_file_in_new_path_mock,
|
|
os_mock, tempfile_mock, add_exec_permission_to_mock):
|
|
# | GIVEN |
|
|
os_mock.path.splitext.return_value = ('any_file', '.scexe')
|
|
fw_img_extractor = (firmware_controller.
|
|
get_fw_extractor(self.any_scexe_file))
|
|
# Now mock the _do_extract method of fw_img_extractor instance
|
|
_do_extract_mock = mock.MagicMock()
|
|
fw_img_extractor._do_extract = _do_extract_mock
|
|
|
|
expected_return_result = 'extracted_firmware_file'
|
|
_get_firmware_file_in_new_path_mock.return_value = (
|
|
expected_return_result)
|
|
# | WHEN |
|
|
actual_return_result, is_extracted = fw_img_extractor.extract()
|
|
# | THEN |
|
|
_do_extract_mock.assert_called_once_with(self.any_scexe_file, mock.ANY)
|
|
self.assertEqual(expected_return_result, actual_return_result)
|
|
|
|
@mock.patch.object(common, 'add_exec_permission_to', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'tempfile', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, '_get_firmware_file_in_new_path',
|
|
autospec=True)
|
|
@mock.patch.object(firmware_controller, 'shutil', autospec=True)
|
|
def test_extract_deletes_temp_extracted_folder_before_raising_exception(
|
|
self, shutil_mock, _get_firmware_file_in_new_path_mock,
|
|
os_mock, tempfile_mock, add_exec_permission_to_mock):
|
|
# | GIVEN |
|
|
os_mock.path.splitext.return_value = ('any_file', '.rpm')
|
|
|
|
fw_img_extractor = (firmware_controller.
|
|
get_fw_extractor(self.any_rpm_file))
|
|
# Now mock the _do_extract method of fw_img_extractor instance
|
|
exc = exception.ImageExtractionFailed(
|
|
image_ref=self.any_rpm_file, reason='God only knows!')
|
|
_do_extract_mock = mock.MagicMock(side_effect=exc)
|
|
fw_img_extractor._do_extract = _do_extract_mock
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(exception.ImageExtractionFailed,
|
|
fw_img_extractor.extract)
|
|
shutil_mock.rmtree.assert_called_once_with(
|
|
tempfile_mock.mkdtemp.return_value, ignore_errors=True)
|
|
|
|
@mock.patch.object(common, 'add_exec_permission_to', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'tempfile', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'os', autospec=True)
|
|
@mock.patch.object(firmware_controller, '_extract_scexe_file',
|
|
autospec=True)
|
|
@mock.patch.object(firmware_controller, '_get_firmware_file_in_new_path',
|
|
autospec=True)
|
|
@mock.patch.object(firmware_controller, 'shutil', autospec=True)
|
|
def test_extract_method_raises_exception_if_raw_fw_file_not_found(
|
|
self, shutil_mock, _get_firmware_file_in_new_path_mock,
|
|
_extract_scexe_mock, os_mock, tempfile_mock,
|
|
add_exec_permission_to_mock):
|
|
# | GIVEN |
|
|
os_mock.path.splitext.return_value = ('any_file', '.scexe')
|
|
_get_firmware_file_in_new_path_mock.return_value = None
|
|
fw_img_extractor = (firmware_controller.
|
|
get_fw_extractor(self.any_scexe_file))
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(exception.InvalidInputError,
|
|
fw_img_extractor.extract)
|
|
|
|
@mock.patch.object(common, 'add_exec_permission_to', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'tempfile', autospec=True)
|
|
@mock.patch.object(firmware_controller, '_extract_scexe_file',
|
|
autospec=True)
|
|
@mock.patch.object(firmware_controller, '_extract_rpm_file', autospec=True)
|
|
# don't use autospec=True here(below one), setting side_effect
|
|
# causes issue. refer https://bugs.python.org/issue17826
|
|
@mock.patch.object(firmware_controller, '_get_firmware_file_in_new_path')
|
|
@mock.patch.object(firmware_controller, 'shutil', autospec=True)
|
|
def test_successive_calls_to_extract_method(
|
|
self, shutil_mock, _get_firmware_file_in_new_path_mock,
|
|
_extract_rpm_mock, _extract_scexe_mock, tempfile_mock,
|
|
add_exec_permission_to_mock):
|
|
"""This is more of an integration test of the extract method
|
|
|
|
"""
|
|
# | GIVEN |
|
|
firmware_files = [
|
|
self.any_scexe_file,
|
|
'any_file.bin',
|
|
self.any_rpm_file,
|
|
]
|
|
actual_raw_fw_files = []
|
|
expected_raw_fw_files = [
|
|
('extracted_file_from_scexe', True),
|
|
('any_file.bin', False),
|
|
('extracted_file_from_rpm', True)
|
|
]
|
|
_get_firmware_file_in_new_path_mock.side_effect = [
|
|
'extracted_file_from_scexe',
|
|
'extracted_file_from_rpm',
|
|
]
|
|
# | WHEN |
|
|
for fw_file in firmware_files:
|
|
fw_img_extractor = firmware_controller.get_fw_extractor(fw_file)
|
|
raw_fw_file, is_extracted = fw_img_extractor.extract()
|
|
actual_raw_fw_files.append((raw_fw_file, is_extracted))
|
|
# | THEN |
|
|
self.assertSequenceEqual(actual_raw_fw_files, expected_raw_fw_files)
|
|
|
|
|
|
@ddt.ddt
|
|
class FirmwareImageUploaderTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
# | BEFORE_EACH |
|
|
self.any_scexe_file = 'any_file.scexe'
|
|
self.any_rpm_file = 'any_file.rpm'
|
|
|
|
@mock.patch.object(firmware_controller.FirmwareImageUploader,
|
|
'_get_socket', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'socket')
|
|
@mock.patch.object(__builtin__, 'open', autospec=True)
|
|
def test_upload_file_to_returns_cookie_after_successful_upload(
|
|
self, open_mock, socket_mock, _get_socket_mock):
|
|
# | GIVEN |
|
|
sock_mock = _get_socket_mock.return_value
|
|
sock_mock.read.side_effect = [b'data returned from socket with ',
|
|
b'Set-Cookie: blah_blah_cookie',
|
|
b'']
|
|
fw_img_uploader = (firmware_controller.
|
|
FirmwareImageUploader('any_raw_file'))
|
|
# | WHEN |
|
|
cookie = fw_img_uploader.upload_file_to(('host', 'port'), 60)
|
|
# | THEN |
|
|
self.assertEqual('blah_blah_cookie', cookie)
|
|
|
|
@mock.patch.object(firmware_controller.FirmwareImageUploader,
|
|
'_get_socket', autospec=True)
|
|
@mock.patch.object(firmware_controller, 'socket')
|
|
@mock.patch.object(__builtin__, 'open', autospec=True)
|
|
def test_upload_file_to_throws_exception_when_cookie_not_returned(
|
|
self, open_mock, socket_mock, _get_socket_mock):
|
|
# | GIVEN |
|
|
sock_mock = _get_socket_mock.return_value
|
|
sock_mock.read.side_effect = [b'data returned from socket with ',
|
|
b'No-Cookie',
|
|
b'']
|
|
fw_img_uploader = (firmware_controller.
|
|
FirmwareImageUploader('any_raw_file'))
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(exception.IloError, fw_img_uploader.upload_file_to,
|
|
('host', 'port'), 60)
|
|
|
|
@mock.patch.object(firmware_controller, 'socket')
|
|
@mock.patch.object(firmware_controller, 'ssl')
|
|
def test__get_socket_returns_ssl_wrapped_socket_if_all_goes_well(
|
|
self, ssl_mock, socket_mock):
|
|
# | GIVEN |
|
|
socket_mock.getaddrinfo().__iter__.return_value = [
|
|
# (family, socktype, proto, canonname, sockaddr),
|
|
(10, 1, 6, '', ('2606:2800:220:1:248:1893:25c8:1946', 80, 0, 0)),
|
|
(2, 1, 6, '', ('0.0.0.0-some-address', 80)),
|
|
]
|
|
fw_img_uploader = (firmware_controller.
|
|
FirmwareImageUploader('any_raw_file'))
|
|
fw_img_uploader.hostname = 'host'
|
|
fw_img_uploader.port = 443
|
|
fw_img_uploader.timeout = 'timeout'
|
|
# | WHEN |
|
|
returned_sock = fw_img_uploader._get_socket()
|
|
# | THEN |
|
|
socket_mock.socket.assert_has_calls([
|
|
mock.call(10, 1, 6),
|
|
mock.call().settimeout('timeout'),
|
|
mock.call().connect(
|
|
('2606:2800:220:1:248:1893:25c8:1946', 80, 0, 0)),
|
|
|
|
mock.call(2, 1, 6),
|
|
mock.call().settimeout('timeout'),
|
|
mock.call().connect(('0.0.0.0-some-address', 80)),
|
|
])
|
|
self.assertTrue(ssl_mock.wrap_socket.called)
|
|
self.assertEqual(returned_sock, ssl_mock.wrap_socket())
|
|
|
|
@ddt.data(('foo.bar.blah.blah', exception.IloConnectionError),)
|
|
@ddt.unpack
|
|
def test__get_socket_throws_exception_in_case_of_failed_connection(
|
|
self, input_hostname, expected_exception_type):
|
|
# | GIVEN |
|
|
fw_img_uploader = (firmware_controller.
|
|
FirmwareImageUploader('any_raw_file'))
|
|
fw_img_uploader.hostname = input_hostname
|
|
fw_img_uploader.port = 443
|
|
fw_img_uploader.timeout = 1
|
|
# | WHEN | & | THEN |
|
|
self.assertRaises(expected_exception_type, fw_img_uploader._get_socket)
|