e303a369dc
When IPA gets a non-raw image, it performs an on-the-fly conversion using qemu-img convert, as well as running qemu-img frequently to get basic information about the image before validating it. Now, we ensure that before any qemu-img calls are made, that we have inspected the image for safety and pass through the detected format. If given a disk_format=raw image and image streaming is enabled (default), we retain the existing behavior of not inspecting it in any way and streaming it bit-perfect to the device. In this case, we never use qemu-based tools on the image at all. If given a disk_format=raw image and image streaming is disabled, this change fixes a bug where the image may have been converted if it was not actually raw in the first place. We now stream these bit-perfect to the device. Adds two config options: - [DEFAULT]/disable_deep_image_inspection, which can be set to "True" in order to disable all security features. Do not do this. - [DEFAULT]/permitted_image_formats, default raw,qcow2, for image types IPA should accept. Both of these configuration options are wired up to be set by the lookup data returned by Ironic at lookup time. This uses a image format inspection module imported from Nova; this inspector will eventually live in oslo.utils, at which point we'll migrate our usage of the inspector to it. Closes-Bug: #2071740 Change-Id: I5254b80717cb5a7f9084e3eff32a00b968f987b7
1534 lines
76 KiB
Python
1534 lines
76 KiB
Python
# 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 gzip
|
|
import shutil
|
|
import tempfile
|
|
from unittest import mock
|
|
|
|
from ironic_lib import exception
|
|
from ironic_lib import utils
|
|
from oslo_concurrency import processutils
|
|
from oslo_config import cfg
|
|
import requests
|
|
|
|
from ironic_python_agent import disk_partitioner
|
|
from ironic_python_agent import disk_utils
|
|
from ironic_python_agent import errors
|
|
from ironic_python_agent import hardware
|
|
from ironic_python_agent import partition_utils
|
|
from ironic_python_agent import qemu_img
|
|
from ironic_python_agent.tests.unit import base
|
|
|
|
|
|
CONF = cfg.CONF
|
|
|
|
|
|
@mock.patch.object(shutil, 'copyfileobj', autospec=True)
|
|
@mock.patch.object(requests, 'get', autospec=True)
|
|
class GetConfigdriveTestCase(base.IronicAgentTest):
|
|
|
|
@mock.patch.object(gzip, 'GzipFile', autospec=True)
|
|
def test_get_configdrive(self, mock_gzip, mock_requests, mock_copy):
|
|
mock_requests.return_value = mock.MagicMock(content='Zm9vYmFy',
|
|
status_code=200)
|
|
tempdir = tempfile.mkdtemp()
|
|
(size, path) = partition_utils.get_configdrive('http://1.2.3.4/cd',
|
|
'fake-node-uuid',
|
|
tempdir=tempdir)
|
|
self.assertTrue(path.startswith(tempdir))
|
|
mock_requests.assert_called_once_with('http://1.2.3.4/cd',
|
|
verify=True, cert=None,
|
|
timeout=60)
|
|
mock_gzip.assert_called_once_with('configdrive', 'rb',
|
|
fileobj=mock.ANY)
|
|
mock_copy.assert_called_once_with(mock.ANY, mock.ANY)
|
|
|
|
@mock.patch.object(gzip, 'GzipFile', autospec=True)
|
|
def test_get_configdrive_insecure(self, mock_gzip, mock_requests,
|
|
mock_copy):
|
|
self.config(insecure=True)
|
|
mock_requests.return_value = mock.MagicMock(content='Zm9vYmFy',
|
|
status_code=200)
|
|
tempdir = tempfile.mkdtemp()
|
|
(size, path) = partition_utils.get_configdrive('http://1.2.3.4/cd',
|
|
'fake-node-uuid',
|
|
tempdir=tempdir)
|
|
self.assertTrue(path.startswith(tempdir))
|
|
mock_requests.assert_called_once_with('http://1.2.3.4/cd',
|
|
verify=False, cert=None,
|
|
timeout=60)
|
|
mock_gzip.assert_called_once_with('configdrive', 'rb',
|
|
fileobj=mock.ANY)
|
|
mock_copy.assert_called_once_with(mock.ANY, mock.ANY)
|
|
|
|
@mock.patch.object(gzip, 'GzipFile', autospec=True)
|
|
def test_get_configdrive_ssl(self, mock_gzip, mock_requests, mock_copy):
|
|
self.config(cafile='cafile', keyfile='keyfile', certfile='certfile')
|
|
mock_requests.return_value = mock.MagicMock(content='Zm9vYmFy',
|
|
status_code=200)
|
|
tempdir = tempfile.mkdtemp()
|
|
(size, path) = partition_utils.get_configdrive('http://1.2.3.4/cd',
|
|
'fake-node-uuid',
|
|
tempdir=tempdir)
|
|
self.assertTrue(path.startswith(tempdir))
|
|
mock_requests.assert_called_once_with('http://1.2.3.4/cd',
|
|
verify='cafile',
|
|
cert=('certfile', 'keyfile'),
|
|
timeout=60)
|
|
mock_gzip.assert_called_once_with('configdrive', 'rb',
|
|
fileobj=mock.ANY)
|
|
mock_copy.assert_called_once_with(mock.ANY, mock.ANY)
|
|
|
|
def test_get_configdrive_binary(self, mock_requests, mock_copy):
|
|
mock_requests.return_value = mock.MagicMock(content=b'content',
|
|
status_code=200)
|
|
tempdir = tempfile.mkdtemp()
|
|
(size, path) = partition_utils.get_configdrive('http://1.2.3.4/cd',
|
|
'fake-node-uuid',
|
|
tempdir=tempdir)
|
|
self.assertTrue(path.startswith(tempdir))
|
|
self.assertEqual(b'content', open(path, 'rb').read())
|
|
mock_requests.assert_called_once_with('http://1.2.3.4/cd',
|
|
verify=True, cert=None,
|
|
timeout=60)
|
|
self.assertFalse(mock_copy.called)
|
|
|
|
@mock.patch.object(gzip, 'GzipFile', autospec=True)
|
|
def test_get_configdrive_base64_string(self, mock_gzip, mock_requests,
|
|
mock_copy):
|
|
partition_utils.get_configdrive('Zm9vYmFy', 'fake-node-uuid')
|
|
self.assertFalse(mock_requests.called)
|
|
mock_gzip.assert_called_once_with('configdrive', 'rb',
|
|
fileobj=mock.ANY)
|
|
mock_copy.assert_called_once_with(mock.ANY, mock.ANY)
|
|
|
|
def test_get_configdrive_bad_url(self, mock_requests, mock_copy):
|
|
mock_requests.side_effect = requests.exceptions.RequestException
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.get_configdrive,
|
|
'http://1.2.3.4/cd', 'fake-node-uuid')
|
|
self.assertFalse(mock_copy.called)
|
|
|
|
def test_get_configdrive_bad_status_code(self, mock_requests, mock_copy):
|
|
mock_requests.return_value = mock.MagicMock(text='Not found',
|
|
status_code=404)
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.get_configdrive,
|
|
'http://1.2.3.4/cd', 'fake-node-uuid')
|
|
self.assertFalse(mock_copy.called)
|
|
|
|
def test_get_configdrive_base64_error(self, mock_requests, mock_copy):
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.get_configdrive,
|
|
'malformed', 'fake-node-uuid')
|
|
self.assertFalse(mock_copy.called)
|
|
|
|
@mock.patch.object(gzip, 'GzipFile', autospec=True)
|
|
def test_get_configdrive_gzip_error(self, mock_gzip, mock_requests,
|
|
mock_copy):
|
|
mock_requests.return_value = mock.MagicMock(content='Zm9vYmFy',
|
|
status_code=200)
|
|
mock_copy.side_effect = IOError
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.get_configdrive,
|
|
'http://1.2.3.4/cd', 'fake-node-uuid')
|
|
mock_requests.assert_called_once_with('http://1.2.3.4/cd',
|
|
verify=True, cert=None,
|
|
timeout=60)
|
|
mock_gzip.assert_called_once_with('configdrive', 'rb',
|
|
fileobj=mock.ANY)
|
|
mock_copy.assert_called_once_with(mock.ANY, mock.ANY)
|
|
|
|
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
class GetLabelledPartitionTestCases(base.IronicAgentTest):
|
|
|
|
def setUp(self):
|
|
super(GetLabelledPartitionTestCases, self).setUp()
|
|
self.dev = "/dev/fake"
|
|
self.config_part_label = "config-2"
|
|
self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
|
|
|
|
def test_get_partition_present(self, mock_execute):
|
|
lsblk_output = 'NAME="fake12" LABEL="config-2"\n'
|
|
part_result = '/dev/fake12'
|
|
mock_execute.side_effect = [(None, ''), (lsblk_output, '')]
|
|
result = partition_utils.get_labelled_partition(self.dev,
|
|
self.config_part_label,
|
|
self.node_uuid)
|
|
self.assertEqual(part_result, result)
|
|
execute_calls = [
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('lsblk', '-Po', 'name,label', self.dev,
|
|
check_exit_code=[0, 1],
|
|
use_standard_locale=True)
|
|
]
|
|
mock_execute.assert_has_calls(execute_calls)
|
|
|
|
def test_get_partition_present_uppercase(self, mock_execute):
|
|
lsblk_output = 'NAME="fake12" LABEL="CONFIG-2"\n'
|
|
part_result = '/dev/fake12'
|
|
mock_execute.side_effect = [(None, ''), (lsblk_output, '')]
|
|
result = partition_utils.get_labelled_partition(self.dev,
|
|
self.config_part_label,
|
|
self.node_uuid)
|
|
self.assertEqual(part_result, result)
|
|
execute_calls = [
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('lsblk', '-Po', 'name,label', self.dev,
|
|
check_exit_code=[0, 1],
|
|
use_standard_locale=True)
|
|
]
|
|
mock_execute.assert_has_calls(execute_calls)
|
|
|
|
def test_get_partition_absent(self, mock_execute):
|
|
mock_execute.side_effect = [(None, ''),
|
|
(None, '')]
|
|
result = partition_utils.get_labelled_partition(self.dev,
|
|
self.config_part_label,
|
|
self.node_uuid)
|
|
self.assertIsNone(result)
|
|
execute_calls = [
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('lsblk', '-Po', 'name,label', self.dev,
|
|
check_exit_code=[0, 1],
|
|
use_standard_locale=True)
|
|
]
|
|
mock_execute.assert_has_calls(execute_calls)
|
|
|
|
def test_get_partition_DeployFail_exc(self, mock_execute):
|
|
label = 'config-2'
|
|
lsblk_output = ('NAME="fake12" LABEL="%s"\n'
|
|
'NAME="fake13" LABEL="%s"\n' %
|
|
(label, label))
|
|
mock_execute.side_effect = [(None, ''), (lsblk_output, '')]
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'fake .*fake12 .*fake13',
|
|
partition_utils.get_labelled_partition,
|
|
self.dev, self.config_part_label,
|
|
self.node_uuid)
|
|
execute_calls = [
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('lsblk', '-Po', 'name,label', self.dev,
|
|
check_exit_code=[0, 1],
|
|
use_standard_locale=True)
|
|
]
|
|
mock_execute.assert_has_calls(execute_calls)
|
|
|
|
@mock.patch.object(partition_utils.LOG, 'error', autospec=True)
|
|
def test_get_partition_exc(self, mock_log, mock_execute):
|
|
mock_execute.side_effect = processutils.ProcessExecutionError
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Failed to retrieve partition labels',
|
|
partition_utils.get_labelled_partition,
|
|
self.dev, self.config_part_label,
|
|
self.node_uuid)
|
|
execute_calls = [
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('lsblk', '-Po', 'name,label', self.dev,
|
|
check_exit_code=[0, 1],
|
|
use_standard_locale=True)
|
|
]
|
|
mock_execute.assert_has_calls(execute_calls)
|
|
self.assertEqual(1, mock_log.call_count)
|
|
|
|
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
class IsDiskLargerThanMaxSizeTestCases(base.IronicAgentTest):
|
|
|
|
dev = "/dev/fake"
|
|
node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
|
|
|
|
def _test_is_disk_larger_than_max_size(self, mock_execute, blk_out):
|
|
mock_execute.return_value = ('%s\n' % blk_out, '')
|
|
result = partition_utils._is_disk_larger_than_max_size(self.dev,
|
|
self.node_uuid)
|
|
mock_execute.assert_called_once_with('blockdev', '--getsize64',
|
|
'/dev/fake',
|
|
use_standard_locale=True)
|
|
return result
|
|
|
|
def test_is_disk_larger_than_max_size_false(self, mock_execute):
|
|
blkid_out = "53687091200"
|
|
ret = self._test_is_disk_larger_than_max_size(mock_execute,
|
|
blk_out=blkid_out)
|
|
self.assertFalse(ret)
|
|
|
|
def test_is_disk_larger_than_max_size_true(self, mock_execute):
|
|
blkid_out = "4398046511104"
|
|
ret = self._test_is_disk_larger_than_max_size(mock_execute,
|
|
blk_out=blkid_out)
|
|
self.assertTrue(ret)
|
|
|
|
@mock.patch.object(partition_utils.LOG, 'error', autospec=True)
|
|
def test_is_disk_larger_than_max_size_exc(self, mock_log, mock_execute):
|
|
mock_execute.side_effect = processutils.ProcessExecutionError
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Failed to get size of disk',
|
|
partition_utils._is_disk_larger_than_max_size,
|
|
self.dev, self.node_uuid)
|
|
mock_execute.assert_called_once_with('blockdev', '--getsize64',
|
|
'/dev/fake',
|
|
use_standard_locale=True)
|
|
self.assertEqual(1, mock_log.call_count)
|
|
|
|
|
|
@mock.patch.object(disk_partitioner.DiskPartitioner, 'commit', lambda _: None)
|
|
class WorkOnDiskTestCase(base.IronicAgentTest):
|
|
|
|
def setUp(self):
|
|
super(WorkOnDiskTestCase, self).setUp()
|
|
self.image_path = '/tmp/xyz/image'
|
|
self.root_mb = 128
|
|
self.swap_mb = 64
|
|
self.ephemeral_mb = 0
|
|
self.ephemeral_format = None
|
|
self.configdrive_mb = 0
|
|
self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
|
|
self.dev = '/dev/fake'
|
|
self.swap_part = '/dev/fake-part1'
|
|
self.root_part = '/dev/fake-part2'
|
|
|
|
self.mock_ibd_obj = mock.patch.object(
|
|
disk_utils, 'is_block_device', autospec=True)
|
|
self.mock_ibd = self.mock_ibd_obj.start()
|
|
self.addCleanup(self.mock_ibd_obj.stop)
|
|
self.mock_mp_obj = mock.patch.object(
|
|
disk_utils, 'make_partitions', autospec=True)
|
|
self.mock_mp = self.mock_mp_obj.start()
|
|
self.addCleanup(self.mock_mp_obj.stop)
|
|
self.mock_remlbl_obj = mock.patch.object(
|
|
disk_utils, 'destroy_disk_metadata', autospec=True)
|
|
self.mock_remlbl = self.mock_remlbl_obj.start()
|
|
self.addCleanup(self.mock_remlbl_obj.stop)
|
|
self.mock_mp.return_value = {'swap': self.swap_part,
|
|
'root': self.root_part}
|
|
|
|
def test_no_root_partition(self):
|
|
self.mock_ibd.return_value = False
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.work_on_disk, self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.ephemeral_format, self.image_path,
|
|
self.node_uuid)
|
|
self.mock_ibd.assert_called_once_with(self.root_part)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
|
|
def test_no_swap_partition(self):
|
|
self.mock_ibd.side_effect = iter([True, False])
|
|
calls = [mock.call(self.root_part),
|
|
mock.call(self.swap_part)]
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.work_on_disk, self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.ephemeral_format, self.image_path,
|
|
self.node_uuid)
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
|
|
def test_no_ephemeral_partition(self):
|
|
ephemeral_part = '/dev/fake-part1'
|
|
swap_part = '/dev/fake-part2'
|
|
root_part = '/dev/fake-part3'
|
|
ephemeral_mb = 256
|
|
ephemeral_format = 'exttest'
|
|
|
|
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
|
|
'swap': swap_part,
|
|
'root': root_part}
|
|
self.mock_ibd.side_effect = iter([True, True, False])
|
|
calls = [mock.call(root_part),
|
|
mock.call(swap_part),
|
|
mock.call(ephemeral_part)]
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.work_on_disk, self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb, ephemeral_format,
|
|
self.image_path, self.node_uuid)
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
|
|
@mock.patch.object(utils, 'unlink_without_raise', autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive', autospec=True)
|
|
def test_no_configdrive_partition(self, mock_configdrive, mock_unlink):
|
|
mock_configdrive.return_value = (10, 'fake-path')
|
|
swap_part = '/dev/fake-part1'
|
|
configdrive_part = '/dev/fake-part2'
|
|
root_part = '/dev/fake-part3'
|
|
configdrive_url = 'http://1.2.3.4/cd'
|
|
configdrive_mb = 10
|
|
|
|
self.mock_mp.return_value = {'swap': swap_part,
|
|
'configdrive': configdrive_part,
|
|
'root': root_part}
|
|
self.mock_ibd.side_effect = iter([True, True, False])
|
|
calls = [mock.call(root_part),
|
|
mock.call(swap_part),
|
|
mock.call(configdrive_part)]
|
|
self.assertRaises(exception.InstanceDeployFailure,
|
|
partition_utils.work_on_disk, self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.ephemeral_format, self.image_path,
|
|
self.node_uuid, preserve_ephemeral=False,
|
|
configdrive=configdrive_url)
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
configdrive_mb, self.node_uuid,
|
|
commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
mock_unlink.assert_called_once_with('fake-path')
|
|
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan',
|
|
lambda d: None)
|
|
@mock.patch.object(utils, 'mkfs', lambda fs, path, label=None: None)
|
|
@mock.patch.object(disk_utils, 'block_uuid', lambda p: 'uuid')
|
|
@mock.patch.object(disk_utils, 'populate_image', autospec=True)
|
|
def test_without_image(self, mock_populate):
|
|
ephemeral_part = '/dev/fake-part1'
|
|
swap_part = '/dev/fake-part2'
|
|
root_part = '/dev/fake-part3'
|
|
ephemeral_mb = 256
|
|
ephemeral_format = 'exttest'
|
|
|
|
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
|
|
'swap': swap_part,
|
|
'root': root_part}
|
|
self.mock_ibd.return_value = True
|
|
calls = [mock.call(root_part),
|
|
mock.call(swap_part),
|
|
mock.call(ephemeral_part)]
|
|
res = partition_utils.work_on_disk(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
ephemeral_format,
|
|
None, self.node_uuid)
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
self.assertEqual(root_part, res['partitions']['root'])
|
|
self.assertEqual('uuid', res['root uuid'])
|
|
self.assertFalse(mock_populate.called)
|
|
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan',
|
|
lambda d: None)
|
|
@mock.patch.object(utils, 'mkfs', lambda fs, path, label=None: None)
|
|
@mock.patch.object(disk_utils, 'block_uuid', lambda p: 'uuid')
|
|
@mock.patch.object(disk_utils, 'populate_image', lambda image_path,
|
|
root_path, conv_flags=None, source_format=None,
|
|
is_raw=False: None)
|
|
def test_gpt_disk_label(self):
|
|
ephemeral_part = '/dev/fake-part1'
|
|
swap_part = '/dev/fake-part2'
|
|
root_part = '/dev/fake-part3'
|
|
ephemeral_mb = 256
|
|
ephemeral_format = 'exttest'
|
|
source_format = 'raw'
|
|
|
|
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
|
|
'swap': swap_part,
|
|
'root': root_part}
|
|
self.mock_ibd.return_value = True
|
|
calls = [mock.call(root_part),
|
|
mock.call(swap_part),
|
|
mock.call(ephemeral_part)]
|
|
partition_utils.work_on_disk(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
ephemeral_format,
|
|
self.image_path, self.node_uuid,
|
|
disk_label='gpt', conv_flags=None,
|
|
source_format=source_format, is_raw=True)
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label='gpt',
|
|
cpu_arch="")
|
|
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan', autospec=True)
|
|
@mock.patch.object(disk_utils, 'block_uuid', autospec=True)
|
|
@mock.patch.object(disk_utils, 'populate_image', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test_uefi(self, mock_mkfs, mock_populate_image,
|
|
mock_block_uuid, mock_trigger_device_rescan):
|
|
"""Test that we create a fat filesystem with UEFI localboot."""
|
|
root_part = '/dev/fake-part1'
|
|
efi_part = '/dev/fake-part2'
|
|
source_format = 'format'
|
|
|
|
self.mock_mp.return_value = {'root': root_part,
|
|
'efi system partition': efi_part}
|
|
self.mock_ibd.return_value = True
|
|
mock_ibd_calls = [mock.call(root_part),
|
|
mock.call(efi_part)]
|
|
|
|
partition_utils.work_on_disk(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.ephemeral_format,
|
|
self.image_path, self.node_uuid,
|
|
boot_mode="uefi",
|
|
source_format=source_format, is_raw=False)
|
|
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="uefi",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
self.assertEqual(self.mock_ibd.call_args_list, mock_ibd_calls)
|
|
mock_mkfs.assert_called_once_with(fs='vfat', path=efi_part,
|
|
label='efi-part')
|
|
mock_populate_image.assert_called_once_with(
|
|
self.image_path, root_part, conv_flags=None,
|
|
source_format=source_format, is_raw=False)
|
|
mock_block_uuid.assert_any_call(root_part)
|
|
mock_block_uuid.assert_any_call(efi_part)
|
|
mock_trigger_device_rescan.assert_called_once_with(self.dev)
|
|
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan', autospec=True)
|
|
@mock.patch.object(disk_utils, 'block_uuid', autospec=True)
|
|
@mock.patch.object(disk_utils, 'populate_image', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test_preserve_ephemeral(self, mock_mkfs, mock_populate_image,
|
|
mock_block_uuid, mock_trigger_device_rescan):
|
|
"""Test that ephemeral partition doesn't get overwritten."""
|
|
ephemeral_part = '/dev/fake-part1'
|
|
root_part = '/dev/fake-part2'
|
|
ephemeral_mb = 256
|
|
ephemeral_format = 'exttest'
|
|
|
|
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
|
|
'root': root_part}
|
|
self.mock_ibd.return_value = True
|
|
calls = [mock.call(root_part),
|
|
mock.call(ephemeral_part)]
|
|
partition_utils.work_on_disk(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
ephemeral_format,
|
|
self.image_path, self.node_uuid,
|
|
preserve_ephemeral=True)
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=False,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="")
|
|
self.assertFalse(mock_mkfs.called)
|
|
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan', autospec=True)
|
|
@mock.patch.object(disk_utils, 'block_uuid', autospec=True)
|
|
@mock.patch.object(disk_utils, 'populate_image', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test_ppc64le_prep_part(self, mock_mkfs, mock_populate_image,
|
|
mock_block_uuid, mock_trigger_device_rescan):
|
|
"""Test that PReP partition uuid is returned."""
|
|
prep_part = '/dev/fake-part1'
|
|
root_part = '/dev/fake-part2'
|
|
|
|
self.mock_mp.return_value = {'PReP Boot partition': prep_part,
|
|
'root': root_part}
|
|
self.mock_ibd.return_value = True
|
|
calls = [mock.call(root_part),
|
|
mock.call(prep_part)]
|
|
partition_utils.work_on_disk(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.ephemeral_format, self.image_path,
|
|
self.node_uuid, cpu_arch='ppc64le')
|
|
self.assertEqual(self.mock_ibd.call_args_list, calls)
|
|
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
|
|
self.swap_mb, self.ephemeral_mb,
|
|
self.configdrive_mb,
|
|
self.node_uuid, commit=True,
|
|
boot_option="local",
|
|
boot_mode="bios",
|
|
disk_label=None,
|
|
cpu_arch="ppc64le")
|
|
self.assertFalse(mock_mkfs.called)
|
|
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan', autospec=True)
|
|
@mock.patch.object(disk_utils, 'block_uuid', autospec=True)
|
|
@mock.patch.object(disk_utils, 'populate_image', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test_convert_to_sparse(self, mock_mkfs, mock_populate_image,
|
|
mock_block_uuid, mock_trigger_device_rescan):
|
|
ephemeral_part = '/dev/fake-part1'
|
|
swap_part = '/dev/fake-part2'
|
|
root_part = '/dev/fake-part3'
|
|
ephemeral_mb = 256
|
|
ephemeral_format = 'exttest'
|
|
fmt = 'format'
|
|
|
|
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
|
|
'swap': swap_part,
|
|
'root': root_part}
|
|
self.mock_ibd.return_value = True
|
|
partition_utils.work_on_disk(self.dev, self.root_mb,
|
|
self.swap_mb, ephemeral_mb,
|
|
ephemeral_format,
|
|
self.image_path, self.node_uuid,
|
|
disk_label='gpt', conv_flags='sparse',
|
|
source_format=fmt,
|
|
is_raw=False)
|
|
|
|
mock_populate_image.assert_called_once_with(self.image_path,
|
|
root_part,
|
|
conv_flags='sparse',
|
|
source_format=fmt,
|
|
is_raw=False)
|
|
|
|
|
|
class CreateConfigDriveTestCases(base.IronicAgentTest):
|
|
|
|
def setUp(self):
|
|
super(CreateConfigDriveTestCases, self).setUp()
|
|
self.dev = "/dev/fake"
|
|
self.config_part_label = "config-2"
|
|
self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
|
|
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'list_partitions',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_exists(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_list_partitions, mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_dd, mock_unlink, mock_execute):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_part = '/dev/fake-part1'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
mock_get_labelled_partition.return_value = configdrive_part
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
partition_utils.create_config_drive_partition(self.node_uuid, self.dev,
|
|
config_url)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
mock_get_labelled_partition.assert_called_with(self.dev,
|
|
self.config_part_label,
|
|
self.node_uuid)
|
|
self.assertFalse(mock_list_partitions.called)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('mount', '-o', 'ro', '-t', 'auto',
|
|
'/dev/fake-part1', mock.ANY),
|
|
mock.call('umount', mock.ANY)])
|
|
self.assertFalse(mock_table_type.called)
|
|
mock_dd.assert_called_with(configdrive_file, configdrive_part)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
|
|
@mock.patch('oslo_utils.uuidutils.generate_uuid', lambda: 'fake-uuid')
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_gpt(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_get_partition_by_uuid,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_dd, mock_unlink, mock_execute):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
|
|
mock_table_type.return_value = 'gpt'
|
|
expected_part = '/dev/fake4'
|
|
mock_get_partition_by_uuid.return_value = expected_part
|
|
partition_utils.create_config_drive_partition(self.node_uuid, self.dev,
|
|
config_url)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('sgdisk', '-n', '0:-64MB:0', '-u', '0:fake-uuid',
|
|
self.dev),
|
|
mock.call('sync'),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('sgdisk', '-v', self.dev),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('test', '-e', expected_part, attempts=15,
|
|
delay_on_retry=True)
|
|
])
|
|
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_dd.assert_called_with(configdrive_file, expected_part)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
|
|
@mock.patch('oslo_utils.uuidutils.generate_uuid', lambda: 'fake-uuid')
|
|
@mock.patch.object(partition_utils, '_try_build_fat32_config_drive',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, '_does_config_drive_work',
|
|
autospec=True)
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_gpt_with_fallback(
|
|
self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_get_partition_by_uuid,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_dd, mock_unlink, mock_execute,
|
|
mock_config_drive_work,
|
|
mock_rebuild_config_drive):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
|
|
mock_table_type.return_value = 'gpt'
|
|
expected_part = '/dev/fake4'
|
|
mock_get_partition_by_uuid.return_value = expected_part
|
|
mock_config_drive_work.return_value = False
|
|
|
|
partition_utils.create_config_drive_partition(self.node_uuid, self.dev,
|
|
config_url)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('sgdisk', '-n', '0:-64MB:0', '-u', '0:fake-uuid',
|
|
self.dev),
|
|
mock.call('sync'),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('sgdisk', '-v', self.dev),
|
|
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('test', '-e', expected_part, attempts=15,
|
|
delay_on_retry=True)
|
|
])
|
|
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_dd.assert_called_with(configdrive_file, expected_part)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_config_drive_work.assert_called_once_with(expected_part)
|
|
mock_rebuild_config_drive.assert_called_once_with(expected_part,
|
|
configdrive_file)
|
|
|
|
@mock.patch('oslo_utils.uuidutils.generate_uuid', lambda: 'fake-uuid')
|
|
@mock.patch.object(partition_utils, '_try_build_fat32_config_drive',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, '_does_config_drive_work',
|
|
autospec=True)
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_gpt_use_vfat(
|
|
self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_get_partition_by_uuid,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_dd, mock_unlink, mock_execute,
|
|
mock_config_drive_work,
|
|
mock_rebuild_config_drive):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
CONF.set_override('config_drive_rebuild', True)
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
|
|
mock_table_type.return_value = 'gpt'
|
|
expected_part = '/dev/fake4'
|
|
mock_get_partition_by_uuid.return_value = expected_part
|
|
mock_config_drive_work.return_value = True
|
|
|
|
partition_utils.create_config_drive_partition(self.node_uuid, self.dev,
|
|
config_url)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('sgdisk', '-n', '0:-64MB:0', '-u', '0:fake-uuid',
|
|
self.dev),
|
|
mock.call('sync'),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('sgdisk', '-v', self.dev),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('test', '-e', expected_part, attempts=15,
|
|
delay_on_retry=True)
|
|
])
|
|
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_dd.assert_not_called()
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_config_drive_work.assert_not_called()
|
|
mock_rebuild_config_drive.assert_called_once_with(expected_part,
|
|
configdrive_file)
|
|
|
|
@mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(partition_utils.LOG, 'warning', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, '_is_disk_larger_than_max_size',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'list_partitions',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def _test_create_partition_mbr(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_list_partitions,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_disk_exceeds, mock_dd,
|
|
mock_unlink, mock_log, mock_execute,
|
|
mock_count, disk_size_exceeds_max=False,
|
|
is_iscsi_device=False,
|
|
is_nvme_device=False):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
mock_disk_exceeds.return_value = disk_size_exceeds_max
|
|
|
|
initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
|
|
'flags': 'boot', 'filesystem': 'ext4',
|
|
'size': 49151},
|
|
{'end': 51099, 'number': 3, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 5, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046}]
|
|
updated_partitions = [{'end': 49152, 'number': 1, 'start': 1,
|
|
'flags': 'boot', 'filesystem': 'ext4',
|
|
'size': 49151},
|
|
{'end': 51099, 'number': 3, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 4, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 5, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046}]
|
|
mock_list_partitions.side_effect = [initial_partitions,
|
|
updated_partitions]
|
|
# 2 primary partitions, 0 logical partitions
|
|
mock_count.return_value = (2, 0)
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
|
|
self.node_uuid = "12345678-1234-1234-1234-1234567890abcxyz"
|
|
if is_iscsi_device:
|
|
self.dev = ('/dev/iqn.2008-10.org.openstack:%s.fake' %
|
|
self.node_uuid)
|
|
expected_part = '%s-part4' % self.dev
|
|
elif is_nvme_device:
|
|
self.dev = '/dev/nvmefake0'
|
|
expected_part = '%sp4' % self.dev
|
|
else:
|
|
expected_part = '/dev/fake4'
|
|
|
|
partition_utils.create_config_drive_partition(self.node_uuid, self.dev,
|
|
config_url)
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
if disk_size_exceeds_max:
|
|
self.assertEqual(1, mock_log.call_count)
|
|
parted_call = mock.call('parted', '-a', 'optimal', '-s',
|
|
'--', self.dev, 'mkpart',
|
|
'primary', 'fat32', 2097087,
|
|
2097151)
|
|
else:
|
|
self.assertEqual(0, mock_log.call_count)
|
|
parted_call = mock.call('parted', '-a', 'optimal', '-s',
|
|
'--', self.dev, 'mkpart',
|
|
'primary', 'fat32', '-64MiB',
|
|
'-0')
|
|
mock_execute.assert_has_calls([
|
|
parted_call,
|
|
mock.call('sync'),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('sgdisk', '-v', self.dev),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('test', '-e', expected_part, attempts=15,
|
|
delay_on_retry=True)
|
|
])
|
|
self.assertEqual(2, mock_list_partitions.call_count)
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_disk_exceeds.assert_called_with(self.dev, self.node_uuid)
|
|
mock_dd.assert_called_with(configdrive_file, expected_part)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_count.assert_called_with(self.dev)
|
|
|
|
def test__create_partition_mbr_disk_under_2TB(self):
|
|
self._test_create_partition_mbr(disk_size_exceeds_max=False,
|
|
is_iscsi_device=True,
|
|
is_nvme_device=False)
|
|
|
|
def test__create_partition_mbr_disk_under_2TB_nvme(self):
|
|
self._test_create_partition_mbr(disk_size_exceeds_max=False,
|
|
is_iscsi_device=False,
|
|
is_nvme_device=True)
|
|
|
|
def test__create_partition_mbr_disk_exceeds_2TB(self):
|
|
self._test_create_partition_mbr(disk_size_exceeds_max=True,
|
|
is_iscsi_device=False,
|
|
is_nvme_device=False)
|
|
|
|
def test__create_partition_mbr_disk_exceeds_2TB_nvme(self):
|
|
self._test_create_partition_mbr(disk_size_exceeds_max=True,
|
|
is_iscsi_device=False,
|
|
is_nvme_device=True)
|
|
|
|
@mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, '_is_disk_larger_than_max_size',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'list_partitions',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_part_create_fail(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_list_partitions,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_disk_exceeds, mock_dd,
|
|
mock_unlink, mock_execute,
|
|
mock_count):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
|
|
'flags': 'boot', 'filesystem': 'ext4',
|
|
'size': 49151},
|
|
{'end': 51099, 'number': 3, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 5, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046}]
|
|
updated_partitions = [{'end': 49152, 'number': 1, 'start': 1,
|
|
'flags': 'boot', 'filesystem': 'ext4',
|
|
'size': 49151},
|
|
{'end': 51099, 'number': 3, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 5, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046}]
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
mock_disk_exceeds.return_value = False
|
|
mock_list_partitions.side_effect = [initial_partitions,
|
|
initial_partitions,
|
|
updated_partitions]
|
|
# 2 primary partitions, 0 logical partitions
|
|
mock_count.return_value = (2, 0)
|
|
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Disk partitioning failed on device',
|
|
partition_utils.create_config_drive_partition,
|
|
self.node_uuid, self.dev, config_url)
|
|
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('parted', '-a', 'optimal', '-s', '--',
|
|
self.dev, 'mkpart', 'primary',
|
|
'fat32', '-64MiB', '-0'),
|
|
mock.call('sync'),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('partprobe', self.dev, attempts=10),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('sgdisk', '-v', self.dev),
|
|
])
|
|
|
|
self.assertEqual(2, mock_list_partitions.call_count)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_disk_exceeds.assert_called_with(self.dev, self.node_uuid)
|
|
self.assertFalse(mock_dd.called)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_count.assert_called_once_with(self.dev)
|
|
|
|
@mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, '_is_disk_larger_than_max_size',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'list_partitions',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_part_create_exc(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_list_partitions,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_disk_exceeds, mock_dd,
|
|
mock_unlink, mock_execute,
|
|
mock_count):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
initial_partitions = [{'end': 49152, 'number': 1, 'start': 1,
|
|
'flags': 'boot', 'filesystem': 'ext4',
|
|
'size': 49151},
|
|
{'end': 51099, 'number': 3, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 5, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046}]
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
mock_disk_exceeds.return_value = False
|
|
mock_list_partitions.side_effect = [initial_partitions,
|
|
initial_partitions]
|
|
# 2 primary partitions, 0 logical partitions
|
|
mock_count.return_value = (2, 0)
|
|
|
|
mock_execute.side_effect = processutils.ProcessExecutionError
|
|
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Failed to create config drive on disk',
|
|
partition_utils.create_config_drive_partition,
|
|
self.node_uuid, self.dev, config_url)
|
|
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
mock_execute.assert_called_with('parted', '-a', 'optimal', '-s', '--',
|
|
self.dev, 'mkpart', 'primary',
|
|
'fat32', '-64MiB', '-0')
|
|
self.assertEqual(1, mock_list_partitions.call_count)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_disk_exceeds.assert_called_with(self.dev, self.node_uuid)
|
|
self.assertFalse(mock_dd.called)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_count.assert_called_once_with(self.dev)
|
|
|
|
@mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'list_partitions',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_num_parts_exceed(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_list_partitions,
|
|
mock_table_type,
|
|
mock_fix_gpt_partition,
|
|
mock_dd, mock_unlink,
|
|
mock_count):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
partitions = [{'end': 49152, 'number': 1, 'start': 1,
|
|
'flags': 'boot', 'filesystem': 'ext4',
|
|
'size': 49151},
|
|
{'end': 51099, 'number': 2, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 3, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046},
|
|
{'end': 51099, 'number': 4, 'start': 49153,
|
|
'flags': '', 'filesystem': '', 'size': 2046}]
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
mock_list_partitions.side_effect = [partitions, partitions]
|
|
# 4 primary partitions, 0 logical partitions
|
|
mock_count.return_value = (4, 0)
|
|
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Config drive cannot be created for node',
|
|
partition_utils.create_config_drive_partition,
|
|
self.node_uuid, self.dev, config_url)
|
|
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
self.assertEqual(1, mock_list_partitions.call_count)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_table_type.assert_called_with(self.dev)
|
|
self.assertFalse(mock_dd.called)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_count.assert_called_once_with(self.dev)
|
|
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_conf_drive_sz_exceed(self, mock_get_configdrive,
|
|
mock_get_labelled_partition,
|
|
mock_unlink, mock_execute):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 65
|
|
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Config drive size exceeds maximum limit',
|
|
partition_utils.create_config_drive_partition,
|
|
self.node_uuid, self.dev, config_url)
|
|
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
|
|
@mock.patch.object(disk_utils, 'count_mbr_partitions', autospec=True)
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(utils, 'unlink_without_raise',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'fix_gpt_partition',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'get_partition_table_type',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_labelled_partition',
|
|
autospec=True)
|
|
@mock.patch.object(partition_utils, 'get_configdrive',
|
|
autospec=True)
|
|
def test_create_partition_conf_drive_error_counting(
|
|
self, mock_get_configdrive, mock_get_labelled_partition,
|
|
mock_table_type, mock_fix_gpt_partition,
|
|
mock_unlink, mock_execute, mock_count):
|
|
config_url = 'http://1.2.3.4/cd'
|
|
configdrive_file = '/tmp/xyz'
|
|
configdrive_mb = 10
|
|
|
|
mock_get_configdrive.return_value = (configdrive_mb, configdrive_file)
|
|
mock_get_labelled_partition.return_value = None
|
|
mock_count.side_effect = ValueError('Booooom')
|
|
|
|
self.assertRaisesRegex(exception.InstanceDeployFailure,
|
|
'Failed to check the number of primary ',
|
|
partition_utils.create_config_drive_partition,
|
|
self.node_uuid, self.dev, config_url)
|
|
|
|
mock_get_configdrive.assert_called_with(config_url, self.node_uuid)
|
|
mock_unlink.assert_called_with(configdrive_file)
|
|
mock_fix_gpt_partition.assert_called_with(self.dev, self.node_uuid)
|
|
mock_table_type.assert_called_with(self.dev)
|
|
mock_count.assert_called_once_with(self.dev)
|
|
|
|
|
|
# NOTE(TheJulia): trigger_device_rescan is systemwide thus pointless
|
|
# to execute in the file test case. Also, CI unit test jobs lack sgdisk.
|
|
@mock.patch.object(disk_utils, 'trigger_device_rescan', autospec=True)
|
|
@mock.patch.object(disk_utils, 'wait_for_disk_to_become_available',
|
|
autospec=True)
|
|
@mock.patch.object(disk_utils, 'is_block_device', autospec=True)
|
|
@mock.patch.object(disk_utils, 'block_uuid', autospec=True)
|
|
@mock.patch.object(disk_utils, 'dd', autospec=True)
|
|
@mock.patch.object(qemu_img, 'convert_image', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
@mock.patch.object(disk_utils, 'populate_image', autospec=True)
|
|
# NOTE(dtantsur): destroy_disk_metadata resets file size, disabling it
|
|
@mock.patch.object(disk_utils, 'destroy_disk_metadata', autospec=True)
|
|
class RealFilePartitioningTestCase(base.IronicAgentTest):
|
|
"""This test applies some real-world partitioning scenario to a file.
|
|
|
|
This test covers the whole partitioning, mocking everything not possible
|
|
on a file. That helps us assure, that we do all partitioning math properly
|
|
and also conducts integration testing of DiskPartitioner.
|
|
"""
|
|
|
|
# Allow calls to utils.execute() and related functions
|
|
block_execute = False
|
|
|
|
def setUp(self):
|
|
super(RealFilePartitioningTestCase, self).setUp()
|
|
try:
|
|
utils.execute('parted', '--version')
|
|
except OSError as exc:
|
|
self.skipTest('parted utility was not found: %s' % exc)
|
|
self.file = tempfile.NamedTemporaryFile(delete=False)
|
|
# NOTE(ifarkas): the file needs to be closed, so fuser won't report
|
|
# any usage
|
|
self.file.close()
|
|
# NOTE(dtantsur): 20 MiB file with zeros
|
|
utils.execute('dd', 'if=/dev/zero', 'of=%s' % self.file.name,
|
|
'bs=1', 'count=0', 'seek=20MiB')
|
|
|
|
@staticmethod
|
|
def _run_without_root(func, *args, **kwargs):
|
|
"""Make sure root is not required when using utils.execute."""
|
|
real_execute = utils.execute
|
|
|
|
def fake_execute(*cmd, **kwargs):
|
|
kwargs.pop('run_as_root', None)
|
|
return real_execute(*cmd, **kwargs)
|
|
|
|
with mock.patch.object(utils, 'execute', fake_execute):
|
|
return func(*args, **kwargs)
|
|
|
|
def test_different_sizes(self, mock_destroy, mock_populate, mock_mkfs,
|
|
mock_convert, mock_dd, mock_block_uuid,
|
|
mock_is_block, mock_wait, mock_trigger_rescan):
|
|
# NOTE(dtantsur): Keep this list in order with expected partitioning
|
|
fields = ['ephemeral_mb', 'swap_mb', 'root_mb']
|
|
variants = ((0, 0, 12), (4, 2, 8), (0, 4, 10), (5, 0, 10))
|
|
for variant in variants:
|
|
kwargs = dict(zip(fields, variant))
|
|
self._run_without_root(partition_utils.work_on_disk,
|
|
self.file.name, ephemeral_format='ext4',
|
|
node_uuid='', image_path='path', **kwargs)
|
|
part_table = self._run_without_root(
|
|
disk_utils.list_partitions, self.file.name)
|
|
for part, expected_size in zip(part_table, filter(None, variant)):
|
|
self.assertEqual(expected_size, part['size'],
|
|
"comparison failed for %s" % list(variant))
|
|
|
|
def test_whole_disk(self, mock_destroy, mock_populate, mock_mkfs,
|
|
mock_convert, mock_dd, mock_block_uuid,
|
|
mock_is_block, mock_wait, mock_trigger_rescan):
|
|
# 6 MiB ephemeral + 3 MiB swap + 9 MiB root + 1 MiB for MBR
|
|
# + 1 MiB MAGIC == 20 MiB whole disk
|
|
# TODO(dtantsur): figure out why we need 'magic' 1 more MiB
|
|
# and why the is different on Ubuntu and Fedora (see below)
|
|
self._run_without_root(partition_utils.work_on_disk, self.file.name,
|
|
root_mb=9, ephemeral_mb=6, swap_mb=3,
|
|
ephemeral_format='ext4', node_uuid='',
|
|
image_path='path')
|
|
part_table = self._run_without_root(
|
|
disk_utils.list_partitions, self.file.name)
|
|
sizes = [part['size'] for part in part_table]
|
|
# NOTE(dtantsur): parted in Ubuntu 12.04 will occupy the last MiB,
|
|
# parted in Fedora 20 won't - thus two possible variants for last part
|
|
self.assertEqual([6, 3], sizes[:2],
|
|
"unexpected partitioning %s" % part_table)
|
|
self.assertIn(sizes[2], (9, 10))
|
|
|
|
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
@mock.patch.object(hardware, 'is_md_device', autospec=True)
|
|
class TestGetPartition(base.IronicAgentTest):
|
|
|
|
fake_dev = '/dev/fake'
|
|
fake_root_uuid = '11111111-2222-3333-4444-555555555555'
|
|
|
|
def test(self, mock_is_md_device, mock_execute):
|
|
mock_is_md_device.side_effect = [False]
|
|
mock_is_md_device.side_effect = [False, False]
|
|
lsblk_output = ('''KNAME="test" UUID="" TYPE="disk"
|
|
KNAME="test1" UUID="256a39e3-ca3c-4fb8-9cc2-b32eec441f47" TYPE="part"
|
|
KNAME="test2" UUID="%s" TYPE="part"''' % self.fake_root_uuid)
|
|
mock_execute.side_effect = ((None, ''), (None, ''), (lsblk_output, ''))
|
|
|
|
root_part = partition_utils.get_partition(
|
|
self.fake_dev, self.fake_root_uuid)
|
|
self.assertEqual('/dev/test2', root_part)
|
|
expected = [mock.call('partx', '-av', self.fake_dev, attempts=3,
|
|
delay_on_retry=True),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('lsblk', '-PbioKNAME,UUID,PARTUUID,TYPE,LABEL',
|
|
self.fake_dev)]
|
|
mock_execute.assert_has_calls(expected)
|
|
|
|
def test_no_device_found(self, mock_is_md_device, mock_execute):
|
|
mock_is_md_device.side_effect = [False, False]
|
|
lsblk_output = ('''KNAME="test" UUID="" TYPE="disk"
|
|
KNAME="test1" UUID="256a39e3-ca3c-4fb8-9cc2-b32eec441f47" TYPE="part"
|
|
KNAME="test2" UUID="" TYPE="part"''')
|
|
mock_execute.side_effect = (
|
|
(None, ''), (None, ''), (lsblk_output, ''),
|
|
processutils.ProcessExecutionError('boom'),
|
|
processutils.ProcessExecutionError('kaboom'))
|
|
|
|
self.assertRaises(errors.DeviceNotFound,
|
|
partition_utils.get_partition, self.fake_dev,
|
|
self.fake_root_uuid)
|
|
expected = [mock.call('partx', '-av', self.fake_dev, attempts=3,
|
|
delay_on_retry=True),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('lsblk', '-PbioKNAME,UUID,PARTUUID,TYPE,LABEL',
|
|
self.fake_dev)]
|
|
mock_execute.assert_has_calls(expected)
|
|
|
|
def test_fallback_partuuid(self, mock_is_md_device, mock_execute):
|
|
mock_is_md_device.side_effect = [False]
|
|
lsblk_output = ('''KNAME="test" UUID="" TYPE="disk"
|
|
KNAME="test1" UUID="256a39e3-ca3c-4fb8-9cc2-b32eec441f47" TYPE="part"
|
|
KNAME="test2" UUID="" TYPE="part"''')
|
|
findfs_output = ('/dev/loop0\n', None)
|
|
mock_execute.side_effect = (
|
|
(None, ''), (None, ''), (lsblk_output, ''),
|
|
processutils.ProcessExecutionError('boom'),
|
|
findfs_output)
|
|
|
|
result = partition_utils.get_partition(
|
|
self.fake_dev, self.fake_root_uuid)
|
|
self.assertEqual('/dev/loop0', result)
|
|
expected = [mock.call('partx', '-av', self.fake_dev, attempts=3,
|
|
delay_on_retry=True),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('lsblk', '-PbioKNAME,UUID,PARTUUID,TYPE,LABEL',
|
|
self.fake_dev),
|
|
mock.call('findfs', 'UUID=%s' % self.fake_root_uuid),
|
|
mock.call('findfs', 'PARTUUID=%s' % self.fake_root_uuid)]
|
|
mock_execute.assert_has_calls(expected)
|
|
|
|
def test_command_fail(self, mock_is_md_device, mock_execute):
|
|
mock_is_md_device.side_effect = [False, False]
|
|
mock_execute.side_effect = (None, None,
|
|
processutils.ProcessExecutionError('boom'))
|
|
self.assertRaises(errors.CommandExecutionError,
|
|
partition_utils.get_partition, self.fake_dev,
|
|
self.fake_root_uuid)
|
|
|
|
expected = [mock.call('partx', '-av', self.fake_dev, attempts=3,
|
|
delay_on_retry=True),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('lsblk', '-PbioKNAME,UUID,PARTUUID,TYPE,LABEL',
|
|
self.fake_dev)]
|
|
mock_execute.assert_has_calls(expected)
|
|
|
|
def test_partuuid(self, mock_is_md_device, mock_execute):
|
|
mock_is_md_device.side_effect = [False, False]
|
|
lsblk_output = ('''KNAME="test" UUID="" TYPE="disk"
|
|
KNAME="test1" UUID="256a39e3-ca3c-4fb8-9cc2-b32eec441f47" TYPE="part"
|
|
KNAME="test2" UUID="903e7bf9-8a13-4f7f-811b-25dc16faf6f7" TYPE="part"\
|
|
LABEL="%s"''' % self.fake_root_uuid)
|
|
mock_execute.side_effect = ((None, ''), (None, ''), (lsblk_output, ''))
|
|
|
|
root_part = partition_utils.get_partition(
|
|
self.fake_dev, self.fake_root_uuid)
|
|
self.assertEqual('/dev/test2', root_part)
|
|
expected = [mock.call('partx', '-av', self.fake_dev, attempts=3,
|
|
delay_on_retry=True),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('lsblk', '-PbioKNAME,UUID,PARTUUID,TYPE,LABEL',
|
|
self.fake_dev)]
|
|
mock_execute.assert_has_calls(expected)
|
|
|
|
def test_label(self, mock_is_md_device, mock_execute):
|
|
mock_is_md_device.side_effect = [False, False]
|
|
lsblk_output = ('''KNAME="test" UUID="" TYPE="disk"
|
|
KNAME="test1" UUID="256a39e3-ca3c-4fb8-9cc2-b32eec441f47" TYPE="part"
|
|
KNAME="test2" PARTUUID="%s" TYPE="part"''' % self.fake_root_uuid)
|
|
mock_execute.side_effect = ((None, ''), (None, ''), (lsblk_output, ''))
|
|
|
|
root_part = partition_utils.get_partition(
|
|
self.fake_dev, self.fake_root_uuid)
|
|
self.assertEqual('/dev/test2', root_part)
|
|
expected = [mock.call('partx', '-av', self.fake_dev, attempts=3,
|
|
delay_on_retry=True),
|
|
mock.call('udevadm', 'settle'),
|
|
mock.call('lsblk', '-PbioKNAME,UUID,PARTUUID,TYPE,LABEL',
|
|
self.fake_dev)]
|
|
mock_execute.assert_has_calls(expected)
|
|
|
|
|
|
@mock.patch.object(utils, 'execute', autospec=True)
|
|
class TestConfigDriveTestRecovery(base.IronicAgentTest):
|
|
|
|
fake_dev = '/dev/fake'
|
|
configdrive_file = '/tmp/config-drive'
|
|
|
|
def test__does_config_drive_work(self, mock_execute):
|
|
self.assertTrue(partition_utils._does_config_drive_work(self.fake_dev))
|
|
mock_execute.assert_has_calls([
|
|
mock.call('mount', '-o', 'ro', '-t', 'auto', self.fake_dev,
|
|
mock.ANY),
|
|
mock.call('umount', mock.ANY)])
|
|
|
|
def test__does_config_drive_failed(self, mock_execute):
|
|
mock_execute.side_effect = processutils.ProcessExecutionError('boom')
|
|
self.assertFalse(
|
|
partition_utils._does_config_drive_work(self.fake_dev)
|
|
)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('mount', '-o', 'ro', '-t', 'auto', self.fake_dev,
|
|
mock.ANY)])
|
|
|
|
@mock.patch.object(shutil, 'copytree', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test__try_build_fat32_config_drive(self,
|
|
mock_mkfs,
|
|
mock_copy,
|
|
mock_execute):
|
|
partition_utils._try_build_fat32_config_drive(self.fake_dev,
|
|
self.configdrive_file)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('mount', '-o', 'loop,ro', '-t', 'auto',
|
|
self.configdrive_file, mock.ANY),
|
|
mock.call('mount', '-t', 'auto', self.fake_dev, mock.ANY),
|
|
mock.call('umount', mock.ANY),
|
|
mock.call('umount', mock.ANY),
|
|
])
|
|
mock_mkfs.assert_called_once_with(fs='vfat', path=self.fake_dev,
|
|
label='CONFIG-2')
|
|
# Validate we called copy as we expect, both source and destination
|
|
# are temporary folders.
|
|
mock_copy.assert_called_once_with(mock.ANY, mock.ANY,
|
|
dirs_exist_ok=True)
|
|
|
|
@mock.patch.object(shutil, 'copytree', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test__try_build_fat32_config_drive_graceful_fail(
|
|
self,
|
|
mock_mkfs,
|
|
mock_copy,
|
|
mock_execute):
|
|
mock_execute.side_effect = processutils.ProcessExecutionError('boom')
|
|
self.assertIsNone(
|
|
partition_utils._try_build_fat32_config_drive(
|
|
self.fake_dev,
|
|
self.configdrive_file)
|
|
)
|
|
mock_execute.assert_called_once_with(
|
|
'mount', '-o', 'loop,ro', '-t', 'auto',
|
|
self.configdrive_file, mock.ANY)
|
|
mock_mkfs.assert_not_called()
|
|
# Validate we called copy as we expect, both source and destination
|
|
# are temporary folders.
|
|
mock_copy.assert_not_called()
|
|
|
|
@mock.patch.object(shutil, 'copytree', autospec=True)
|
|
@mock.patch.object(utils, 'mkfs', autospec=True)
|
|
def test__try_build_fat32_config_drive_fails_once_invalid(
|
|
self,
|
|
mock_mkfs,
|
|
mock_copy,
|
|
mock_execute):
|
|
mock_mkfs.side_effect = processutils.ProcessExecutionError('boom')
|
|
self.assertRaisesRegex(
|
|
exception.InstanceDeployFailure,
|
|
'A failure occurred while attempting to format.*',
|
|
partition_utils._try_build_fat32_config_drive,
|
|
self.fake_dev,
|
|
self.configdrive_file)
|
|
mock_execute.assert_has_calls([
|
|
mock.call('mount', '-o', 'loop,ro', '-t', 'auto',
|
|
self.configdrive_file, mock.ANY),
|
|
mock.call('umount', mock.ANY),
|
|
mock.call('umount', mock.ANY),
|
|
])
|
|
|
|
mock_mkfs.assert_called_once_with(fs='vfat', path=self.fake_dev,
|
|
label='CONFIG-2')
|
|
mock_copy.assert_not_called()
|