openstack/ironic-python-agent
Code Issues Proposed changes
Files
d298e06b4996f9efe41cee541e34d21a789d014b
ironic-python-agent/ironic_python_agent/tests/unit/extensions/test_standby.py
Jay Faulkner 36e5993a04 [codespell] Fix spelling issues in IPA 
This fixes several spelling issues identified by codepsell. In some
cases, I may have manually modified a line to make the output more clear
or to correct grammatical issues which were obvious in the codespell
output.

Later changes in this chain will provide the codespell config used to
generate this, as well as adding this commit's SHA, once landed, to a
.git-blame-ignore-revs file to ensure it will not pollute git historys
for modern clients.

Related-Bug: 2047654
Change-Id: I240cf8484865c9b748ceb51f3c7b9fd973cb5ada
2023-12-28 10:54:46 -08:00

2139 lines
97 KiB
Python
Raw Blame History

# Copyright 2013 Rackspace, Inc.
#
# 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
import time
from unittest import mock
from ironic_lib import exception
from oslo_concurrency import processutils
from oslo_config import cfg
import requests
from ironic_python_agent import errors
from ironic_python_agent.extensions import standby
from ironic_python_agent import hardware
from ironic_python_agent import partition_utils
from ironic_python_agent.tests.unit import base
from ironic_python_agent import utils
CONF = cfg.CONF
def _build_fake_image_info(url='http://example.org'):
return {
'id': 'fake_id',
'node_uuid': '1be26c0b-03f2-4d2e-ae87-c02d7f33c123',
'urls': [url],
'image_type': 'whole-disk-image',
'os_hash_algo': 'sha256',
'os_hash_value': 'fake-checksum',
}
def _build_fake_partition_image_info():
return {
'id': 'fake_id',
'urls': [
'http://example.org',
],
'node_uuid': 'node_uuid',
'root_mb': '10',
'swap_mb': '10',
'ephemeral_mb': '10',
'ephemeral_format': 'abc',
'preserve_ephemeral': 'False',
'image_type': 'partition',
'disk_label': 'msdos',
'deploy_boot_mode': 'bios',
'os_hash_algo': 'sha256',
'os_hash_value': 'fake-checksum'}
class TestStandbyExtension(base.IronicAgentTest):
def setUp(self):
super(TestStandbyExtension, self).setUp()
self.agent_extension = standby.StandbyExtension()
self.fake_cpu = hardware.CPU(model_name='fuzzypickles',
frequency=1024,
count=1,
architecture='generic',
flags='')
with mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True):
hardware.cache_node(
{'uuid': '1-2-3-4',
'instance_info': {}})
def test_validate_image_info_success(self):
standby._validate_image_info(None, _build_fake_image_info())
def test_validate_image_info_success_with_new_hash_fields(self):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'md5'
image_info['os_hash_value'] = 'fake-checksum'
standby._validate_image_info(None, image_info)
def test_validate_image_info_success_without_md5(self):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'fake-checksum'
standby._validate_image_info(None, image_info)
def test_validate_image_info_success_ignore_none_md5(self):
image_info = _build_fake_image_info()
image_info['checksum'] = None
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'fake-checksum'
standby._validate_image_info(None, image_info)
def test_validate_image_info_legacy_md5_checksum_enabled(self):
image_info = _build_fake_image_info()
CONF.set_override('md5_enabled', True)
image_info['checksum'] = 'fake-checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
standby._validate_image_info(None, image_info)
def test_validate_image_info_url(self):
image_info = _build_fake_image_info()
image_info['checksum'] = 'https://fake.url'
del image_info['os_hash_algo']
del image_info['os_hash_value']
standby._validate_image_info(None, image_info)
def test_validate_image_info_sha256(self):
image_info = _build_fake_image_info()
image_info['checksum'] = 'a' * 64
del image_info['os_hash_algo']
del image_info['os_hash_value']
standby._validate_image_info(None, image_info)
def test_validate_image_info_legacy_md5_checksum(self):
CONF.set_override('md5_enabled', False)
image_info = _build_fake_image_info()
del image_info['os_hash_algo']
del image_info['os_hash_value']
image_info['checksum'] = 'fake-checksum'
self.assertRaisesRegex(errors.InvalidCommandParamsError,
'Image checksum is not',
standby._validate_image_info,
None,
image_info)
def test_validate_image_info_missing_field(self):
for field in ['id', 'urls', 'os_hash_value']:
invalid_info = _build_fake_image_info()
del invalid_info[field]
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_validate_image_info_invalid_urls(self):
invalid_info = _build_fake_image_info()
invalid_info['urls'] = 'this_is_not_a_list'
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_validate_image_info_empty_urls(self):
invalid_info = _build_fake_image_info()
invalid_info['urls'] = []
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_validate_image_info_invalid_checksum(self):
invalid_info = _build_fake_image_info()
invalid_info['checksum'] = {'not': 'a string'}
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_validate_image_info_empty_checksum(self):
invalid_info = _build_fake_image_info()
invalid_info['checksum'] = ''
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_validate_image_info_no_hash_value(self):
invalid_info = _build_fake_image_info()
invalid_info['os_hash_algo'] = 'sha512'
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_validate_image_info_no_hash_algo(self):
invalid_info = _build_fake_image_info()
invalid_info['os_hash_value'] = 'fake-checksum'
self.assertRaises(errors.InvalidCommandParamsError,
standby._validate_image_info,
invalid_info)
def test_image_location(self):
image_info = _build_fake_image_info()
location = standby._image_location(image_info)
# Can't hardcode /tmp here, each test is running in an isolated
# tempdir
expected_loc = os.path.join(tempfile.gettempdir(), 'fake_id')
self.assertEqual(expected_loc, location)
def test_verify_basic_auth_creds(self):
image_info = _build_fake_image_info()
self.assertIsNone(standby._verify_basic_auth_creds("SpongeBob",
"SquarePants",
image_info['id']))
def test_gen_auth_from_image_info_user_pass_success(self):
image_info = _build_fake_image_info()
image_info['image_server_auth_strategy'] = 'http_basic'
image_info['image_server_user'] = 'SpongeBob'
image_info['image_server_password'] = 'SquarePants'
exp_auth = requests.auth.HTTPBasicAuth('SpongeBob', 'SquarePants')
return_auth = \
standby._gen_auth_from_image_info_user_pass(image_info,
image_info['id'])
self.assertEqual(exp_auth, return_auth)
def test_gen_auth_from_image_info_user_pass_none(self):
image_info = _build_fake_image_info()
image_info['image_server_auth_strategy'] = ''
image_info['image_server_user'] = 'SpongeBob'
image_info['image_server_password'] = 'SquarePants'
return_auth = \
standby._gen_auth_from_image_info_user_pass(image_info,
image_info['id'])
self.assertIsNone(return_auth)
def test_gen_auth_from_oslo_conf_user_pass_success(self):
image_info = _build_fake_image_info()
CONF.set_override('image_server_auth_strategy', 'http_basic')
CONF.set_override('image_server_password', 'SpongeBob')
CONF.set_override('image_server_user', 'SquarePants')
correct_auth = \
requests.auth.HTTPBasicAuth(CONF['image_server_user'],
CONF['image_server_password'])
return_auth = \
standby._gen_auth_from_oslo_conf_user_pass(image_info['id'])
self.assertEqual(correct_auth, return_auth)
def test_gen_auth_from_oslo_conf_user_pass_none(self):
image_info = _build_fake_image_info()
CONF.set_override('image_server_auth_strategy', 'noauth')
CONF.set_override('image_server_password', 'SpongeBob')
CONF.set_override('image_server_user', 'SquarePants')
return_auth = \
standby._gen_auth_from_oslo_conf_user_pass(image_info['id'])
self.assertIsNone(return_auth)
def test_verify_basic_auth_creds_empty_user(self):
image_info = _build_fake_image_info()
self.assertRaises(errors.ImageDownloadError,
standby._verify_basic_auth_creds,
"",
"SquarePants",
image_info['id'])
def test_verify_basic_auth_creds_empty_password(self):
image_info = _build_fake_image_info()
self.assertRaises(errors.ImageDownloadError,
standby._verify_basic_auth_creds,
"SpongeBob",
"",
image_info['id'])
def test_verify_basic_auth_creds_none_user(self):
image_info = _build_fake_image_info()
self.assertRaises(errors.ImageDownloadError,
standby._verify_basic_auth_creds,
None,
"SquarePants",
image_info['id'])
def test_verify_basic_auth_creds_none_password(self):
image_info = _build_fake_image_info()
self.assertRaises(errors.ImageDownloadError,
standby._verify_basic_auth_creds,
"SpongeBob",
None,
image_info['id'])
@mock.patch('ironic_lib.disk_utils.fix_gpt_partition', autospec=True)
@mock.patch('ironic_lib.disk_utils.trigger_device_rescan', autospec=True)
@mock.patch('ironic_lib.disk_utils.convert_image', autospec=True)
@mock.patch('ironic_lib.disk_utils.udev_settle', autospec=True)
@mock.patch('ironic_lib.disk_utils.destroy_disk_metadata', autospec=True)
def test_write_image(self, wipe_mock, udev_mock, convert_mock,
rescan_mock, fix_gpt_mock):
image_info = _build_fake_image_info()
device = '/dev/sda'
location = standby._image_location(image_info)
standby._write_image(image_info, device)
convert_mock.assert_called_once_with(location, device,
out_format='host_device',
cache='directsync',
out_of_order=True,
sparse_size='0')
wipe_mock.assert_called_once_with(device, '')
udev_mock.assert_called_once_with()
rescan_mock.assert_called_once_with(device)
fix_gpt_mock.assert_called_once_with(device, node_uuid=None)
@mock.patch('ironic_lib.disk_utils.fix_gpt_partition', autospec=True)
@mock.patch('ironic_lib.disk_utils.trigger_device_rescan', autospec=True)
@mock.patch('ironic_lib.disk_utils.convert_image', autospec=True)
@mock.patch('ironic_lib.disk_utils.udev_settle', autospec=True)
@mock.patch('ironic_lib.disk_utils.destroy_disk_metadata', autospec=True)
def test_write_image_gpt_fails(self, wipe_mock, udev_mock, convert_mock,
rescan_mock, fix_gpt_mock):
image_info = _build_fake_image_info()
device = '/dev/sda'
fix_gpt_mock.side_effect = exception.InstanceDeployFailure
standby._write_image(image_info, device)
@mock.patch('ironic_lib.disk_utils.convert_image', autospec=True)
@mock.patch('ironic_lib.disk_utils.udev_settle', autospec=True)
@mock.patch('ironic_lib.disk_utils.destroy_disk_metadata', autospec=True)
def test_write_image_fails(self, wipe_mock, udev_mock, convert_mock):
image_info = _build_fake_image_info()
device = '/dev/sda'
convert_mock.side_effect = processutils.ProcessExecutionError
self.assertRaises(errors.ImageWriteError,
standby._write_image,
image_info,
device)
@mock.patch.object(utils, 'get_node_boot_mode', lambda self: 'bios')
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_lib.disk_utils.get_image_mb', autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_exception(self, work_on_disk_mock,
image_mb_mock,
execute_mock, open_mock,
dispatch_mock):
image_info = _build_fake_partition_image_info()
device = '/dev/sda'
root_mb = image_info['root_mb']
swap_mb = image_info['swap_mb']
ephemeral_mb = image_info['ephemeral_mb']
ephemeral_format = image_info['ephemeral_format']
node_uuid = image_info['node_uuid']
pr_ep = image_info['preserve_ephemeral']
boot_mode = image_info['deploy_boot_mode']
disk_label = image_info['disk_label']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 1
dispatch_mock.return_value = self.fake_cpu
exc = errors.ImageWriteError
Exception_returned = processutils.ProcessExecutionError
work_on_disk_mock.side_effect = Exception_returned
self.assertRaises(exc, standby._write_image, image_info,
device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
image_path,
node_uuid,
configdrive='configdrive',
preserve_ephemeral=pr_ep,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
@mock.patch.object(utils, 'get_node_boot_mode', lambda self: 'bios')
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_lib.disk_utils.get_image_mb', autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_no_node_uuid(self, work_on_disk_mock,
image_mb_mock,
execute_mock, open_mock,
dispatch_mock):
image_info = _build_fake_partition_image_info()
image_info['node_uuid'] = None
device = '/dev/sda'
root_mb = image_info['root_mb']
swap_mb = image_info['swap_mb']
ephemeral_mb = image_info['ephemeral_mb']
ephemeral_format = image_info['ephemeral_format']
node_uuid = image_info['node_uuid']
pr_ep = image_info['preserve_ephemeral']
boot_mode = image_info['deploy_boot_mode']
disk_label = image_info['disk_label']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 1
dispatch_mock.return_value = self.fake_cpu
uuids = {'root uuid': 'root_uuid'}
expected_uuid = {'root uuid': 'root_uuid'}
image_mb_mock.return_value = 1
work_on_disk_mock.return_value = uuids
standby._write_image(image_info, device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
image_path,
node_uuid,
configdrive='configdrive',
preserve_ephemeral=pr_ep,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
self.assertEqual(expected_uuid, work_on_disk_mock.return_value)
self.assertIsNone(node_uuid)
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_lib.disk_utils.get_image_mb', autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_exception_image_mb(self,
work_on_disk_mock,
image_mb_mock,
execute_mock,
open_mock,
dispatch_mock):
dispatch_mock.return_value = self.fake_cpu
image_info = _build_fake_partition_image_info()
device = '/dev/sda'
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 20
exc = errors.InvalidCommandParamsError
self.assertRaises(exc, standby._write_image, image_info,
device)
image_mb_mock.assert_called_once_with(image_path)
self.assertFalse(work_on_disk_mock.called)
@mock.patch.object(utils, 'get_node_boot_mode', lambda self: 'bios')
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
@mock.patch('ironic_lib.disk_utils.get_image_mb', autospec=True)
def test_write_partition_image(self, image_mb_mock, work_on_disk_mock,
execute_mock, open_mock, dispatch_mock):
image_info = _build_fake_partition_image_info()
device = '/dev/sda'
root_mb = image_info['root_mb']
swap_mb = image_info['swap_mb']
ephemeral_mb = image_info['ephemeral_mb']
ephemeral_format = image_info['ephemeral_format']
node_uuid = image_info['node_uuid']
pr_ep = image_info['preserve_ephemeral']
boot_mode = image_info['deploy_boot_mode']
disk_label = image_info['disk_label']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
uuids = {'root uuid': 'root_uuid'}
expected_uuid = {'root uuid': 'root_uuid'}
image_mb_mock.return_value = 1
dispatch_mock.return_value = self.fake_cpu
work_on_disk_mock.return_value = uuids
standby._write_image(image_info, device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
image_path,
node_uuid,
configdrive='configdrive',
preserve_ephemeral=pr_ep,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
self.assertEqual(expected_uuid, work_on_disk_mock.return_value)
@mock.patch('ironic_python_agent.extensions.standby.LOG', autospec=True)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_download_image(self, requests_mock, open_mock, hash_mock,
log_mock):
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.status_code = 200
response.headers = {}
response.iter_content.return_value = ['some', 'content']
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.read.return_value = None
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
standby._download_image(image_info)
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60)
write = file_mock.write
write.assert_any_call('some')
write.assert_any_call('content')
self.assertEqual(2, write.call_count)
log_mock_calls = [
mock.call.info('Attempting to download image from %s',
'http://example.org'),
mock.call.debug('Verifying image at %(image_location)s against '
'%(algo_name)s checksum %(checksum)s',
{'image_location': mock.ANY,
'algo_name': mock.ANY,
'checksum': 'fake-checksum'}),
mock.call.info('Image downloaded from %(image_location)s in '
'%(totaltime)s seconds. Transferred %(size)s '
'bytes. Server originally reported: %(reported)s.',
{'image_location': mock.ANY,
'totaltime': mock.ANY,
'size': 11,
'reported': None}),
]
log_mock.assert_has_calls(log_mock_calls)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
@mock.patch.dict(os.environ, {})
def test_download_image_proxy(
self, requests_mock, open_mock, hash_mock):
image_info = _build_fake_image_info()
proxies = {'http': 'http://a.b.com',
'https': 'https://secure.a.b.com'}
no_proxy = '.example.org,.b.com'
image_info['proxies'] = proxies
image_info['no_proxy'] = no_proxy
image_info['os_hash_value'] = 'fake-checksum'
response = requests_mock.return_value
response.status_code = 200
response.iter_content.return_value = ['some', 'content']
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.read.return_value = None
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = 'fake-checksum'
standby._download_image(image_info)
self.assertEqual(no_proxy, os.environ['no_proxy'])
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies=proxies,
timeout=60)
write = file_mock.write
write.assert_any_call('some')
write.assert_any_call('content')
self.assertEqual(2, write.call_count)
hash_mock.assert_has_calls([
mock.call('sha256'),
mock.call().update(b'some'),
mock.call().update(b'content'),
mock.call().hexdigest()])
@mock.patch('requests.get', autospec=True)
def test_download_image_bad_status(self, requests_mock):
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.status_code = 404
self.assertRaises(errors.ImageDownloadError,
standby._download_image,
image_info)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_download_image_basic_auth_conf_success(self, requests_mock,
open_mock, hash_mock):
image_info = _build_fake_image_info()
CONF.set_override('image_server_auth_strategy', 'http_basic')
CONF.set_override('image_server_password', 'SpongeBob')
CONF.set_override('image_server_user', 'SquarePants')
user = CONF.image_server_user
password = CONF.image_server_password
correct_auth = requests.auth.HTTPBasicAuth(user, password)
response = requests_mock.return_value
response.status_code = 200
response.iter_content.return_value = ['some', 'content']
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.read.return_value = None
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
standby._download_image(image_info)
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60, auth=correct_auth)
write = file_mock.write
write.assert_any_call('some')
write.assert_any_call('content')
self.assertEqual(2, write.call_count)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_download_image_basic_auth_image_info_success(self,
requests_mock,
open_mock,
hash_mock):
image_info = _build_fake_image_info()
image_info['image_server_auth_strategy'] = 'http_basic'
image_info['image_server_password'] = 'SpongeBob'
image_info['image_server_user'] = 'SquarePants'
user = image_info['image_server_user']
password = image_info['image_server_password']
correct_auth = requests.auth.HTTPBasicAuth(user, password)
response = requests_mock.return_value
response.status_code = 200
response.iter_content.return_value = ['some', 'content']
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.read.return_value = None
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
standby._download_image(image_info)
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60, auth=correct_auth)
write = file_mock.write
write.assert_any_call('some')
write.assert_any_call('content')
self.assertEqual(2, write.call_count)
def test_download_image_bad_basic_auth_conf_credential(self):
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
CONF.set_override('image_server_auth_strategy', 'http_basic')
self.assertRaises(errors.ImageDownloadError,
standby._download_image,
image_info)
def test_download_image_bad_basic_auth_image_info_credential(self):
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
image_info['image_server_auth_strategy'] = 'http_basic'
self.assertRaises(errors.ImageDownloadError,
standby._download_image,
image_info)
def test_download_image_bad_basic_auth_mixed_credential(self):
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
image_info['image_server_auth_strategy'] = 'http_basic'
CONF.set_override('image_server_password', 'SpongeBob')
CONF.set_override('image_server_user', 'SquarePants')
self.assertRaises(errors.ImageDownloadError,
standby._download_image,
image_info)
def test_download_image_bad_basic_auth_mixed_credential_second(self):
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
CONF.set_override('image_server_auth_strategy', 'http_basic')
image_info['image_server_password'] = 'SpongeBob'
image_info['image_server_user'] = 'SquarePants'
self.assertRaises(errors.ImageDownloadError,
standby._download_image,
image_info)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_download_image_verify_fails(self, requests_mock, open_mock,
hash_mock):
# Set the config to 0 retries, so we don't retry in this case
# and cause the test download to loop multiple times.
self.config(image_download_connection_retries=0)
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.status_code = 200
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = 'invalid-checksum'
self.assertRaises(errors.ImageChecksumError,
standby._download_image,
image_info)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_success(self, requests_mock, open_mock, hash_mock):
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.status_code = 200
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
image_location = '/foo/bar'
image_download = standby.ImageDownload(image_info)
image_download.verify_image(image_location)
self.assertEqual(0, image_download.bytes_transferred)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_success_with_new_hash_fields(self, requests_mock,
open_mock,
hashlib_mock):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'fake-sha512-value'
response = requests_mock.return_value
response.status_code = 200
hexdigest_mock = hashlib_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
image_location = '/foo/bar'
image_download = standby.ImageDownload(image_info)
image_download.verify_image(image_location)
hashlib_mock.assert_called_with('sha512')
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_success_without_md5(self, requests_mock,
open_mock, hashlib_mock):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'fake-sha512-value'
response = requests_mock.return_value
response.status_code = 200
hexdigest_mock = hashlib_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
image_location = '/foo/bar'
image_download = standby.ImageDownload(image_info)
image_download.verify_image(image_location)
hashlib_mock.assert_called_with('sha512')
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_success_with_md5_fallback(self, requests_mock,
open_mock, hash_mock):
CONF.set_override('md5_enabled', True)
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'algo-beyond-milky-way'
image_info['os_hash_value'] = 'mysterious-alien-codes'
image_info['checksum'] = 'd41d8cd98f00b204e9800998ecf8427e'
response = requests_mock.return_value
response.status_code = 200
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['checksum']
image_location = '/foo/bar'
image_download = standby.ImageDownload(image_info)
image_download.verify_image(image_location)
# NOTE(TheJulia): This is the one test which falls all the
# way back to md5 as the default, legacy logic because it
# got bad input to start with.
hash_mock.assert_has_calls([
mock.call('md5'),
mock.call().__bool__(),
mock.call().hexdigest()])
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_fails_if_unknown_is_used(self, requests_mock,
open_mock, hash_mock):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'algo-beyond-milky-way'
image_info['os_hash_value'] = 'mysterious-alien-codes'
self.assertRaisesRegex(
errors.RESTError,
'An error occurred: Unable to verify image fake_id with '
'available checksums.',
standby.ImageDownload,
image_info)
hash_mock.assert_not_called()
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_failure_with_new_hash_fields(self, requests_mock,
open_mock,
hashlib_mock):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'fake-sha512-value'
response = requests_mock.return_value
response.status_code = 200
image_download = standby.ImageDownload(image_info)
image_location = '/foo/bar'
hexdigest_mock = hashlib_mock.return_value.hexdigest
hexdigest_mock.return_value = 'invalid-checksum'
self.assertRaises(errors.ImageChecksumError,
image_download.verify_image,
image_location)
hashlib_mock.assert_called_with('sha512')
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_failure(self, requests_mock, open_mock, hash_mock):
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.status_code = 200
image_download = standby.ImageDownload(image_info)
image_location = '/foo/bar'
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = 'invalid-checksum'
self.assertRaises(errors.ImageChecksumError,
image_download.verify_image,
image_location)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_verify_image_failure_without_fallback(self, requests_mock,
open_mock, hashlib_mock):
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'unsupported-algorithm'
image_info['os_hash_value'] = 'fake-value'
response = requests_mock.return_value
response.status_code = 200
self.assertRaisesRegex(errors.RESTError,
'Unable to verify image.*'
'unsupported-algorithm',
standby.ImageDownload,
image_info)
@mock.patch('ironic_lib.disk_utils.get_disk_identifier',
lambda dev: 'ROOT')
@mock.patch('ironic_python_agent.utils.execute',
autospec=True)
@mock.patch('ironic_lib.disk_utils.list_partitions',
autospec=True)
@mock.patch.object(partition_utils, 'create_config_drive_partition',
autospec=True)
@mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_prepare_image(self,
download_mock,
write_mock,
dispatch_mock,
configdrive_copy_mock,
list_part_mock,
execute_mock):
image_info = _build_fake_image_info()
download_mock.return_value = None
write_mock.return_value = None
dispatch_mock.return_value = 'manager'
configdrive_copy_mock.return_value = None
list_part_mock.return_value = [mock.MagicMock()]
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive='configdrive_data'
)
async_result.join()
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, 'manager',
'configdrive_data')
dispatch_mock.assert_called_once_with('get_os_install_device',
permit_refresh=True)
configdrive_copy_mock.assert_called_once_with(image_info['node_uuid'],
'manager',
'configdrive_data')
self.assertEqual('SUCCEEDED', async_result.command_status)
self.assertIn('result', async_result.command_result)
cmd_result = ('prepare_image: image ({}) written to device {} '
'root_uuid=ROOT').format(image_info['id'], 'manager')
self.assertEqual(cmd_result, async_result.command_result['result'])
list_part_mock.assert_called_with('manager')
execute_mock.assert_called_with('partprobe', 'manager',
run_as_root=True,
attempts=mock.ANY)
self.assertEqual({'root uuid': 'ROOT'},
self.agent_extension.partition_uuids)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_lib.disk_utils.list_partitions',
autospec=True)
@mock.patch.object(partition_utils, 'create_config_drive_partition',
autospec=True)
@mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_prepare_partition_image(self,
download_mock,
write_mock,
dispatch_mock,
configdrive_copy_mock,
list_part_mock,
execute_mock):
image_info = _build_fake_partition_image_info()
download_mock.return_value = None
write_mock.return_value = {'root uuid': 'root_uuid'}
dispatch_mock.return_value = 'manager'
configdrive_copy_mock.return_value = None
list_part_mock.return_value = [mock.MagicMock()]
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive='configdrive_data'
)
async_result.join()
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, 'manager',
'configdrive_data')
dispatch_mock.assert_called_once_with('get_os_install_device',
permit_refresh=True)
self.assertFalse(configdrive_copy_mock.called)
self.assertEqual('SUCCEEDED', async_result.command_status)
self.assertIn('result', async_result.command_result)
cmd_result = ('prepare_image: image ({}) written to device {} '
'root_uuid={}').format(
image_info['id'], 'manager', 'root_uuid')
self.assertEqual(cmd_result, async_result.command_result['result'])
download_mock.reset_mock()
write_mock.reset_mock()
configdrive_copy_mock.reset_mock()
# image is now cached, make sure download/write doesn't happen
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive='configdrive_data'
)
async_result.join()
self.assertEqual(0, download_mock.call_count)
self.assertEqual(0, write_mock.call_count)
self.assertFalse(configdrive_copy_mock.called)
self.assertEqual('SUCCEEDED', async_result.command_status)
self.assertIn('result', async_result.command_result)
cmd_result = ('prepare_image: image ({}) written to device {} '
'root_uuid={}').format(
image_info['id'], 'manager', 'root_uuid')
self.assertEqual(cmd_result, async_result.command_result['result'])
list_part_mock.assert_called_with('manager')
execute_mock.assert_called_with('partprobe', 'manager',
run_as_root=True,
attempts=mock.ANY)
self.assertEqual({'root uuid': 'root_uuid'},
self.agent_extension.partition_uuids)
@mock.patch('ironic_lib.disk_utils.get_disk_identifier',
lambda dev: 'ROOT')
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch.object(partition_utils, 'create_config_drive_partition',
autospec=True)
@mock.patch('ironic_lib.disk_utils.list_partitions',
autospec=True)
@mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_prepare_image_no_configdrive(self,
download_mock,
write_mock,
dispatch_mock,
list_part_mock,
configdrive_copy_mock,
execute_mock):
image_info = _build_fake_image_info()
download_mock.return_value = None
write_mock.return_value = None
dispatch_mock.return_value = 'manager'
configdrive_copy_mock.return_value = None
list_part_mock.return_value = [mock.MagicMock()]
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive=None
)
async_result.join()
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, 'manager', None)
dispatch_mock.assert_called_once_with('get_os_install_device',
permit_refresh=True)
self.assertEqual(0, configdrive_copy_mock.call_count)
self.assertEqual('SUCCEEDED', async_result.command_status)
self.assertIn('result', async_result.command_result)
cmd_result = ('prepare_image: image ({}) written to device {} '
'root_uuid=ROOT').format(image_info['id'], 'manager')
self.assertEqual(cmd_result, async_result.command_result['result'])
@mock.patch('ironic_lib.disk_utils.get_disk_identifier',
lambda dev: 'ROOT')
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
@mock.patch.object(partition_utils, 'create_config_drive_partition',
autospec=True)
@mock.patch('ironic_lib.disk_utils.list_partitions',
autospec=True)
@mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_prepare_image_bad_partition(self,
download_mock,
write_mock,
dispatch_mock,
list_part_mock,
configdrive_copy_mock,
work_on_disk_mock):
list_part_mock.side_effect = processutils.ProcessExecutionError
image_info = _build_fake_image_info()
download_mock.return_value = None
write_mock.return_value = None
dispatch_mock.return_value = 'manager'
configdrive_copy_mock.return_value = None
work_on_disk_mock.return_value = {
'root uuid': 'a318821b-2a60-40e5-a011-7ac07fce342b',
'partitions': {
'root': '/dev/foo-part1',
}
}
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive=None
)
async_result.join()
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, 'manager', None)
dispatch_mock.assert_called_once_with('get_os_install_device',
permit_refresh=True)
self.assertFalse(configdrive_copy_mock.called)
self.assertEqual('FAILED', async_result.command_status)
@mock.patch('ironic_lib.disk_utils.get_disk_identifier',
side_effect=OSError, autospec=True)
@mock.patch('ironic_python_agent.utils.execute',
autospec=True)
@mock.patch('ironic_lib.disk_utils.list_partitions',
autospec=True)
@mock.patch.object(partition_utils, 'create_config_drive_partition',
autospec=True)
@mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_prepare_image_no_hexdump(self,
download_mock,
write_mock,
dispatch_mock,
configdrive_copy_mock,
list_part_mock,
execute_mock,
disk_id_mock):
image_info = _build_fake_image_info()
download_mock.return_value = None
write_mock.return_value = None
dispatch_mock.return_value = 'manager'
configdrive_copy_mock.return_value = None
list_part_mock.return_value = [mock.MagicMock()]
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive='configdrive_data'
)
async_result.join()
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, 'manager',
'configdrive_data')
dispatch_mock.assert_called_once_with('get_os_install_device',
permit_refresh=True)
configdrive_copy_mock.assert_called_once_with(image_info['node_uuid'],
'manager',
'configdrive_data')
self.assertEqual('SUCCEEDED', async_result.command_status)
self.assertIn('result', async_result.command_result)
cmd_result = ('prepare_image: image ({}) written to device {} '
'root_uuid=None').format(image_info['id'], 'manager')
self.assertEqual(cmd_result, async_result.command_result['result'])
list_part_mock.assert_called_with('manager')
execute_mock.assert_called_with('partprobe', 'manager',
run_as_root=True,
attempts=mock.ANY)
self.assertEqual({}, self.agent_extension.partition_uuids)
@mock.patch('ironic_python_agent.utils.execute', mock.Mock())
@mock.patch('ironic_lib.disk_utils.list_partitions',
lambda _dev: [mock.Mock()])
@mock.patch('ironic_lib.disk_utils.get_disk_identifier',
lambda dev: 'ROOT')
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
@mock.patch.object(partition_utils, 'create_config_drive_partition',
autospec=True)
@mock.patch('ironic_python_agent.hardware.dispatch_to_managers',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby.StandbyExtension'
'._cache_and_write_image', autospec=True)
@mock.patch('ironic_python_agent.extensions.standby.StandbyExtension'
'._stream_raw_image_onto_device', autospec=True)
def _test_prepare_image_raw(self, image_info, stream_mock,
cache_write_mock, dispatch_mock,
configdrive_copy_mock, work_on_disk_mock,
partition=False):
# Calls get_cpus().architecture with partition images
dispatch_mock.side_effect = ['/dev/foo', self.fake_cpu]
configdrive_copy_mock.return_value = None
work_on_disk_mock.return_value = {
'root uuid': 'a318821b-2a60-40e5-a011-7ac07fce342b',
'partitions': {
'root': '/dev/foo-part1',
}
}
if partition:
expected_device = '/dev/foo-part1'
else:
expected_device = '/dev/foo'
async_result = self.agent_extension.prepare_image(
image_info=image_info,
configdrive=None
)
async_result.join()
dispatch_mock.assert_any_call('get_os_install_device',
permit_refresh=True)
self.assertFalse(configdrive_copy_mock.called)
# Assert we've streamed the image or not
if image_info['stream_raw_images']:
stream_mock.assert_called_once_with(mock.ANY, image_info,
expected_device)
self.assertFalse(cache_write_mock.called)
self.assertIs(partition, work_on_disk_mock.called)
else:
cache_write_mock.assert_called_once_with(mock.ANY, image_info,
'/dev/foo', None)
self.assertFalse(stream_mock.called)
def test_prepare_image_raw_stream_true(self):
image_info = _build_fake_image_info()
image_info['disk_format'] = 'raw'
image_info['stream_raw_images'] = True
self._test_prepare_image_raw(image_info)
def test_prepare_image_raw_and_stream_false(self):
image_info = _build_fake_image_info()
image_info['disk_format'] = 'raw'
image_info['stream_raw_images'] = False
self._test_prepare_image_raw(image_info)
def test_prepare_partition_image_raw_stream_true(self):
image_info = _build_fake_partition_image_info()
image_info['disk_format'] = 'raw'
image_info['stream_raw_images'] = True
self._test_prepare_image_raw(image_info, partition=True)
self.assertEqual({'root uuid': 'a318821b-2a60-40e5-a011-7ac07fce342b',
'partitions': {'root': '/dev/foo-part1'}},
self.agent_extension.partition_uuids)
def test_prepare_partition_image_raw_and_stream_false(self):
image_info = _build_fake_partition_image_info()
image_info['disk_format'] = 'raw'
image_info['stream_raw_images'] = False
self._test_prepare_image_raw(image_info, partition=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_run_shutdown_command_invalid(self, execute_mock):
self.assertRaises(errors.InvalidCommandParamsError,
self.agent_extension._run_shutdown_command, 'boot')
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_run_shutdown_command_fails(self, execute_mock):
execute_mock.side_effect = processutils.ProcessExecutionError
self.assertRaises(errors.SystemRebootError,
self.agent_extension._run_shutdown_command, 'reboot')
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_run_shutdown_command_valid(self, execute_mock):
execute_mock.return_value = ('', '')
self.agent_extension._run_shutdown_command('poweroff')
calls = [mock.call('sync'),
mock.call('poweroff', use_standard_locale=True)]
execute_mock.assert_has_calls(calls)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_run_shutdown_command_valid_poweroff_sysrq(self, execute_mock):
execute_mock.side_effect = [
('', ''), ('', ''),
processutils.ProcessExecutionError(''),
('', '')]
self.agent_extension._run_shutdown_command('poweroff')
calls = [mock.call('hwclock', '-v', '--systohc'),
mock.call('sync'),
mock.call('poweroff', use_standard_locale=True),
mock.call("echo o > /proc/sysrq-trigger", shell=True)]
execute_mock.assert_has_calls(calls)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_run_shutdown_command_valid_reboot_sysrq(self, execute_mock):
execute_mock.side_effect = [('', ''), ('', ''), ('',
'Running in chroot, ignoring request.'),
('', '')]
self.agent_extension._run_shutdown_command('reboot')
calls = [mock.call('sync'),
mock.call('reboot', use_standard_locale=True),
mock.call("echo b > /proc/sysrq-trigger", shell=True)]
execute_mock.assert_has_calls(calls)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_run_image(self, execute_mock):
execute_mock.return_value = ('', '')
success_result = self.agent_extension.run_image()
success_result.join()
calls = [mock.call('sync'),
mock.call('reboot', use_standard_locale=True)]
execute_mock.assert_has_calls(calls)
self.assertEqual('SUCCEEDED', success_result.command_status)
execute_mock.reset_mock()
execute_mock.return_value = ('', '')
execute_mock.side_effect = processutils.ProcessExecutionError
failed_result = self.agent_extension.run_image()
failed_result.join()
execute_mock.assert_any_call('sync')
self.assertEqual('FAILED', failed_result.command_status)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_power_off(self, execute_mock):
execute_mock.return_value = ('', '')
success_result = self.agent_extension.power_off()
success_result.join()
calls = [mock.call('sync'),
mock.call('poweroff', use_standard_locale=True)]
execute_mock.assert_has_calls(calls)
self.assertEqual('SUCCEEDED', success_result.command_status)
execute_mock.reset_mock()
execute_mock.return_value = ('', '')
execute_mock.side_effect = processutils.ProcessExecutionError
failed_result = self.agent_extension.power_off()
failed_result.join()
execute_mock.assert_any_call('sync')
self.assertEqual('FAILED', failed_result.command_status)
@mock.patch('ironic_python_agent.utils.determine_time_method',
autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_power_off_with_ntp_server(self, execute_mock, mock_timemethod):
self.config(fail_if_clock_not_set=False)
self.config(ntp_server='192.168.1.1')
execute_mock.return_value = ('', '')
mock_timemethod.return_value = 'ntpdate'
success_result = self.agent_extension.power_off()
success_result.join()
calls = [mock.call('ntpdate', '192.168.1.1'),
mock.call('hwclock', '-v', '--systohc'),
mock.call('sync'),
mock.call('poweroff', use_standard_locale=True)]
execute_mock.assert_has_calls(calls)
self.assertEqual('SUCCEEDED', success_result.command_status)
self.config(fail_if_clock_not_set=True)
execute_mock.reset_mock()
execute_mock.return_value = ('', '')
execute_mock.side_effect = processutils.ProcessExecutionError
failed_result = self.agent_extension.power_off()
failed_result.join()
execute_mock.assert_any_call('ntpdate', '192.168.1.1')
self.assertEqual('FAILED', failed_result.command_status)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_sync(self, execute_mock):
result = self.agent_extension.sync()
execute_mock.assert_called_once_with('sync')
self.assertEqual('SUCCEEDED', result.command_status)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_sync_error(self, execute_mock):
execute_mock.side_effect = processutils.ProcessExecutionError
self.assertRaises(
errors.CommandExecutionError, self.agent_extension.sync)
execute_mock.assert_called_once_with('sync')
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_cache_and_write_image(self, download_mock, write_mock):
image_info = _build_fake_image_info()
device = '/dev/foo'
self.agent_extension._cache_and_write_image(image_info, device)
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, device, None)
@mock.patch('ironic_python_agent.extensions.standby._write_image',
autospec=True)
@mock.patch('ironic_python_agent.extensions.standby._download_image',
autospec=True)
def test_cache_and_write_image_configdirve(self, download_mock,
write_mock):
image_info = _build_fake_image_info()
device = '/dev/foo'
self.agent_extension._cache_and_write_image(image_info, device,
'configdrive_data')
download_mock.assert_called_once_with(image_info)
write_mock.assert_called_once_with(image_info, device,
'configdrive_data')
@mock.patch('ironic_python_agent.extensions.standby.LOG', autospec=True)
@mock.patch('ironic_lib.disk_utils.block_uuid', autospec=True)
@mock.patch('ironic_lib.disk_utils.fix_gpt_partition', autospec=True)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_stream_raw_image_onto_device(self, requests_mock, open_mock,
hash_mock, fix_gpt_mock,
block_uuid_mock,
mock_log):
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.headers = {'Content-Length': 11}
response.status_code = 200
response.iter_content.return_value = ['some', 'content']
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.read.return_value = None
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
self.agent_extension.partition_uuids = {}
block_uuid_mock.return_value = 'aaaabbbb'
self.agent_extension._stream_raw_image_onto_device(image_info,
'/dev/foo')
hash_mock.assert_has_calls([
mock.call('sha256'),
mock.call().update(b'some'),
mock.call().update(b'content'),
mock.call().hexdigest()])
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60)
expected_calls = [mock.call('some'), mock.call('content')]
file_mock.write.assert_has_calls(expected_calls)
fix_gpt_mock.assert_called_once_with('/dev/foo', node_uuid=None)
block_uuid_mock.assert_called_once_with('/dev/foo')
self.assertEqual(
'aaaabbbb',
self.agent_extension.partition_uuids['root uuid']
)
mock_log_calls = [
mock.call.info('Attempting to download image from %s',
'http://example.org'),
mock.call.debug('Verifying image at %(image_location)s '
'against %(algo_name)s checksum %(checksum)s',
{'image_location': '/dev/foo',
'algo_name': mock.ANY,
'checksum': 'fake-checksum'}),
mock.call.info('Image streamed onto device %(device)s in '
'%(totaltime)s seconds for %(size)s bytes. '
'Server originally reported %(reported)s.',
{'device': '/dev/foo',
'totaltime': mock.ANY,
'size': 11,
'reported': 11}),
mock.call.info('%(device)s UUID is now %(root_uuid)s',
{'device': '/dev/foo', 'root_uuid': 'aaaabbbb'})
]
mock_log.assert_has_calls(mock_log_calls)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_stream_raw_image_onto_device_write_error(self, requests_mock,
open_mock, hash_mock):
self.config(image_download_connection_timeout=1)
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
response = requests_mock.return_value
response.status_code = 200
response.headers = {}
response.iter_content.return_value = [b'some', b'content']
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.write.side_effect = Exception('Surprise!!!1!')
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
self.assertRaises(errors.ImageDownloadError,
self.agent_extension._stream_raw_image_onto_device,
image_info, '/dev/foo')
calls = [mock.call('http://example.org', cert=None, proxies={},
stream=True, timeout=1, verify=True),
mock.call().iter_content(mock.ANY),
mock.call('http://example.org', cert=None, proxies={},
stream=True, timeout=1, verify=True),
mock.call().iter_content(mock.ANY),
mock.call('http://example.org', cert=None, proxies={},
stream=True, timeout=1, verify=True),
mock.call().iter_content(mock.ANY)]
requests_mock.assert_has_calls(calls)
write_calls = [mock.call(b'some'),
mock.call(b'some'),
mock.call(b'some')]
file_mock.write.assert_has_calls(write_calls)
@mock.patch('ironic_lib.disk_utils.fix_gpt_partition', autospec=True)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('requests.get', autospec=True)
def test_stream_raw_image_onto_device_socket_read_timeout(
self, requests_mock, open_mock, hash_mock, fix_gpt_mock):
class create_timeout(object):
status_code = 200
def __init__(self, url, stream, proxies, verify, cert, timeout):
self.count = 0
def __iter__(self):
return self
def __next__(self):
if self.count:
time.sleep(0.1)
return None
self.count += 1
return b"meow"
def iter_content(self, chunk_size):
return self
@property
def headers(self):
return {}
self.config(image_download_connection_timeout=1)
self.config(image_download_connection_retries=2)
self.config(image_download_connection_retry_interval=0)
image_info = _build_fake_image_info()
file_mock = mock.Mock()
open_mock.return_value.__enter__.return_value = file_mock
file_mock.read.return_value = None
hexdigest_mock = hash_mock.return_value.hexdigest
hexdigest_mock.return_value = image_info['os_hash_value']
requests_mock.side_effect = create_timeout
self.assertRaisesRegex(
errors.ImageDownloadError,
'Timed out reading next chunk',
self.agent_extension._stream_raw_image_onto_device,
image_info,
'/dev/foo')
calls = [mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=1),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=1),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=1)]
requests_mock.assert_has_calls(calls)
write_calls = [mock.call(b'meow'),
mock.call(b'meow'),
mock.call(b'meow')]
file_mock.write.assert_has_calls(write_calls)
fix_gpt_mock.assert_not_called()
def test__message_format_partition_bios(self):
image_info = _build_fake_partition_image_info()
msg = ('image ({}) already present on device {} ')
device = '/dev/fake'
partition_uuids = {'root uuid': 'root_uuid',
'efi system partition uuid': None}
result_msg = standby._message_format(msg, image_info,
device, partition_uuids)
expected_msg = ('image (fake_id) already present on device '
'/dev/fake root_uuid=root_uuid')
self.assertEqual(expected_msg, result_msg)
def test__message_format_partition_uefi(self):
image_info = _build_fake_partition_image_info()
image_info['deploy_boot_mode'] = 'uefi'
msg = ('image ({}) already present on device {} ')
device = '/dev/fake'
partition_uuids = {'root uuid': 'root_uuid',
'efi system partition uuid': 'efi_id'}
result_msg = standby._message_format(msg, image_info,
device, partition_uuids)
expected_msg = ('image (fake_id) already present on device '
'/dev/fake root_uuid=root_uuid '
'efi_system_partition_uuid=efi_id')
self.assertEqual(expected_msg, result_msg)
@mock.patch('ironic_python_agent.utils.determine_time_method',
autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test__sync_clock(self, execute_mock, mock_timemethod):
self.config(ntp_server='192.168.1.1')
self.config(fail_if_clock_not_set=True)
execute_mock.return_value = ('', '')
mock_timemethod.return_value = 'chronyd'
self.agent_extension._sync_clock()
calls = [mock.call('chronyc', 'shutdown', check_exit_code=[0, 1]),
mock.call("chronyd -q 'server 192.168.1.1 iburst'",
shell=True),
mock.call('hwclock', '-v', '--systohc')]
execute_mock.assert_has_calls(calls)
execute_mock.reset_mock()
execute_mock.side_effect = [
('', ''), ('', ''),
processutils.ProcessExecutionError('boop')
]
self.assertRaises(errors.ClockSyncError,
self.agent_extension._sync_clock)
execute_mock.assert_any_call('hwclock', '-v', '--systohc')
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
def test_get_partition_uuids(self, execute_mock):
self.agent_extension.partition_uuids = {'1': '2'}
result = self.agent_extension.get_partition_uuids()
self.assertEqual({'1': '2'}, result.serialize()['command_result'])
@mock.patch.object(utils, 'get_node_boot_mode', lambda self: 'uefi')
@mock.patch.object(utils, 'get_partition_table_type_from_specs',
lambda self: 'gpt')
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_lib.disk_utils.get_image_mb', autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_no_node_uuid_uefi(
self, work_on_disk_mock,
image_mb_mock,
execute_mock, open_mock,
dispatch_mock):
image_info = _build_fake_partition_image_info()
image_info['node_uuid'] = None
device = '/dev/sda'
root_mb = image_info['root_mb']
swap_mb = image_info['swap_mb']
ephemeral_mb = image_info['ephemeral_mb']
ephemeral_format = image_info['ephemeral_format']
node_uuid = image_info['node_uuid']
pr_ep = image_info['preserve_ephemeral']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 1
dispatch_mock.return_value = self.fake_cpu
uuids = {'root uuid': 'root_uuid'}
expected_uuid = {'root uuid': 'root_uuid'}
image_mb_mock.return_value = 1
work_on_disk_mock.return_value = uuids
standby._write_image(image_info, device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
image_path,
node_uuid,
configdrive='configdrive',
preserve_ephemeral=pr_ep,
boot_mode='uefi',
disk_label='gpt',
cpu_arch=cpu_arch)
self.assertEqual(expected_uuid, work_on_disk_mock.return_value)
self.assertIsNone(node_uuid)
@mock.patch('hashlib.new', autospec=True)
@mock.patch('requests.get', autospec=True)
class TestImageDownload(base.IronicAgentTest):
def test_download_image(self, requests_mock, hash_mock):
content = ['SpongeBob', 'SquarePants']
response = requests_mock.return_value
response.status_code = 200
response.iter_content.return_value = content
image_info = _build_fake_image_info()
hash_mock.return_value.hexdigest.return_value = image_info[
'os_hash_value']
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60)
self.assertEqual(image_info['os_hash_value'],
image_download._hash_algo.hexdigest())
@mock.patch('time.sleep', autospec=True)
def test_download_image_fail(self, sleep_mock, requests_mock, time_mock):
response = requests_mock.return_value
response.status_code = 401
response.text = 'Unauthorized'
time_mock.return_value = 0.0
image_info = _build_fake_image_info()
msg = ('Error downloading image: Download of image fake_id failed: '
'URL: http://example.org; time: .* seconds. Error: '
'Received status code 401 from http://example.org, expected '
'200. Response body: Unauthorized')
self.assertRaisesRegex(errors.ImageDownloadError, msg,
standby.ImageDownload, image_info)
requests_mock.assert_called_once_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60)
self.assertFalse(sleep_mock.called)
@mock.patch('time.sleep', autospec=True)
def test_download_image_retries(self, sleep_mock, requests_mock,
time_mock):
self.config(image_download_connection_retries=2)
response = requests_mock.return_value
response.status_code = 500
response.text = 'Oops'
time_mock.return_value = 0.0
image_info = _build_fake_image_info()
msg = ('Error downloading image: Download of image fake_id failed: '
'URL: http://example.org; time: .* seconds. Error: '
'Received status code 500 from http://example.org, expected '
'200. Response body: Oops')
self.assertRaisesRegex(errors.ImageDownloadError, msg,
standby.ImageDownload, image_info)
requests_mock.assert_called_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60)
self.assertEqual(3, requests_mock.call_count)
sleep_mock.assert_called_with(10)
self.assertEqual(2, sleep_mock.call_count)
@mock.patch('time.sleep', autospec=True)
def test_download_image_retries_success(self, sleep_mock, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fail_response = mock.Mock()
fail_response.status_code = 500
fail_response.text = " "
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [requests.Timeout, fail_response, response]
image_info = _build_fake_image_info()
hash_mock.return_value.hexdigest.return_value = image_info[
'os_hash_value']
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_called_with(image_info['urls'][0],
cert=None, verify=True,
stream=True, proxies={},
timeout=60)
self.assertEqual(3, requests_mock.call_count)
sleep_mock.assert_called_with(10)
self.assertEqual(2, sleep_mock.call_count)
def test_download_image_and_checksum(self, requests_mock, hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = "019fe036425da1c562f2e9f5299820bf"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = fake_cs + '\n'
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info()
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'http://example.com/checksum'
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None, verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
def test_download_image_and_checksum_md5(self, requests_mock, hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = "019fe036425da1c562f2e9f5299820bf"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = fake_cs + '\n'
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info()
del image_info['os_hash_value']
del image_info['os_hash_algo']
CONF.set_override('md5_enabled', True)
image_info['checksum'] = 'http://example.com/checksum'
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None, verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
hash_mock.assert_has_calls([
mock.call('md5')])
def test_download_image_and_checksum_multiple_md5(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = "019fe036425da1c562f2e9f5299820bf"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
foobar irrelevant file.img
%s image.img
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
CONF.set_override('md5_enabled', True)
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None, verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
def test_download_image_and_centos_checksum_md5(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = "019fe036425da1c562f2e9f5299820bf"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
# centos-image.img: 1005593088 bytes
MD5 (centos-image.img) = %s
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/centos-image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
CONF.set_override('md5_enabled', True)
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None,
verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
def test_download_image_and_centos_checksum_sha256(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = ('3b678e4fb651d450f4970e1647abc9b0a38bff3febd3d558753'
'623c66369a633')
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
# centos-image.img: 1005593088 bytes
SHA256 (centos-image.img) = %s
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = iter(content)
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/centos-image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None,
verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
hash_mock.assert_has_calls([
mock.call('sha256')])
def test_download_image_and_centos_checksum_sha512(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = ('3b678e4fb651d450f4970e1647abc9b0a38bff3febd3d558753'
'623c66369a6333b678e4fb651d450f4970e1647abc9b0a38b'
'ff3febd3d558753623c66369a633')
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
# centos-image.img: 1005593088 bytes
SHA512 (centos-image.img) = %s
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = iter(content)
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/centos-image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None,
verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
hash_mock.assert_has_calls([
mock.call('sha512')])
def test_download_image_and_checksum_multiple_sha256(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = ('3b678e4fb651d450f4970e1647abc9b0a38bff3febd3d558753'
'623c66369a633')
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
foobar irrelevant file.img
%s image.img
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = iter(content)
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None, verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
hash_mock.assert_has_calls([
mock.call('sha256')])
def test_download_image_and_checksum_multiple_sha512(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = ('3b678e4fb651d450f4970e1647abc9b0a38bff3febd3d558753'
'623c66369a6333b678e4fb651d450f4970e1647abc9b0a38b'
'ff3febd3d558753623c66369a633')
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
foobar irrelevant file.img
%s image.img
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = iter(content)
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
hash_mock.return_value.hexdigest.return_value = fake_cs
image_download = standby.ImageDownload(image_info)
self.assertEqual(content, list(image_download))
requests_mock.assert_has_calls([
mock.call('http://example.com/checksum', cert=None, verify=True,
stream=True, proxies={}, timeout=60),
mock.call(image_info['urls'][0], cert=None, verify=True,
stream=True, proxies={}, timeout=60),
])
self.assertEqual(fake_cs, image_download._hash_algo.hexdigest())
hash_mock.assert_has_calls([
mock.call('sha512')])
def test_download_image_and_checksum_unknown_file(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = "019fe036425da1c562f2e9f5299820bf"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
foobar irrelevant file.img
%s not-my-image.img
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'http://example.com/checksum'
hash_mock.return_value.hexdigest.return_value = fake_cs
self.assertRaisesRegex(errors.ImageDownloadError,
'Checksum file does not contain name image.img',
standby.ImageDownload, image_info)
def test_download_image_and_checksum_unknown_file_md5(self,
requests_mock,
hash_mock):
CONF.set_override('md5_enabled', True)
content = ['SpongeBob', 'SquarePants']
fake_cs = "019fe036425da1c562f2e9f5299820bf"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = """
foobar irrelevant file.img
%s not-my-image.img
""" % fake_cs
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
hash_mock.return_value.hexdigest.return_value = fake_cs
self.assertRaisesRegex(errors.ImageDownloadError,
'Checksum file does not contain name image.img',
standby.ImageDownload, image_info)
def test_download_image_and_checksum_empty_file_md5(self, requests_mock,
hash_mock):
CONF.set_override('md5_enabled', True)
content = ['SpongeBob', 'SquarePants']
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = " "
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_algo']
del image_info['os_hash_value']
self.assertRaisesRegex(errors.ImageDownloadError,
'Empty checksum file',
standby.ImageDownload, image_info)
def test_download_image_and_checksum_empty_file(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = " "
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'http://example.com/checksum'
self.assertRaisesRegex(errors.ImageDownloadError,
'Empty checksum file',
standby.ImageDownload, image_info)
def test_download_image_and_checksum_failed(self, requests_mock,
hash_mock):
self.config(image_download_connection_retry_interval=0)
content = ['SpongeBob', 'SquarePants']
cs_response = mock.Mock()
cs_response.status_code = 400
cs_response.text = " "
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
# 3 retries on status code
requests_mock.side_effect = [cs_response, cs_response, cs_response,
response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['os_hash_value'] = 'http://example.com/checksum'
image_info['os_hash_algo'] = 'sha512'
self.assertRaisesRegex(errors.ImageDownloadError,
'Received status code 400 from '
'http://example.com/checksum',
standby.ImageDownload, image_info)
def test_download_image_and_checksum_failed_md5(self,
requests_mock,
hash_mock):
CONF.set_override('md5_enabled', True)
self.config(image_download_connection_retry_interval=0)
content = ['SpongeBob', 'SquarePants']
cs_response = mock.Mock()
cs_response.status_code = 400
cs_response.text = " "
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
# 3 retries on status code
requests_mock.side_effect = [cs_response, cs_response, cs_response,
response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['checksum'] = 'http://example.com/checksum'
del image_info['os_hash_value']
del image_info['os_hash_algo']
self.assertRaisesRegex(errors.ImageDownloadError,
'Received status code 400 from '
'http://example.com/checksum',
standby.ImageDownload, image_info)
def test_download_image_and_invalid_checksum(self, requests_mock,
hash_mock):
content = ['SpongeBob', 'SquarePants']
fake_cs = "invalid"
cs_response = mock.Mock()
cs_response.status_code = 200
cs_response.text = fake_cs + '\n'
response = mock.Mock()
response.status_code = 200
response.iter_content.return_value = content
requests_mock.side_effect = [cs_response, response]
image_info = _build_fake_image_info(
'http://example.com/path/image.img')
image_info['os_hash_algo'] = 'sha512'
image_info['os_hash_value'] = 'http://example.com/checksum'
self.assertRaisesRegex(
errors.ImageDownloadError,
r"Invalid checksum file \(No valid checksum found\) \['invalid'\]",
standby.ImageDownload, image_info)
View Git Blame Copy Permalink