Michael Hudson-Doyle has proposed merging
~mwhudson/curtin:partition-verify-refactor into curtin:master.
Commit message:
block_meta: refactor partition verification slightly
I want to call these functions from the new partitioning code and this
tweaking makes this easier.
Requested reviews:
curtin developers (curtin-dev)
For more details, see:
https://code.launchpad.net/~mwhudson/curtin/+git/curtin/+merge/412495
--
Your team curtin developers is requested to review the proposed merge of
~mwhudson/curtin:partition-verify-refactor into curtin:master.
diff --git a/curtin/commands/block_meta.py b/curtin/commands/block_meta.py
index 5c3226d..f5f20f6 100644
--- a/curtin/commands/block_meta.py
+++ b/curtin/commands/block_meta.py
@@ -765,12 +765,7 @@ def verify_exists(devpath):
raise RuntimeError("Device %s does not exist" % devpath)
-def verify_size(devpath, expected_size_bytes, sfdisk_info=None):
- if not sfdisk_info:
- sfdisk_info = block.sfdisk_info(devpath)
-
- part_info = block.get_partition_sfdisk_info(devpath,
- sfdisk_info=sfdisk_info)
+def verify_size(devpath, expected_size_bytes, part_info):
(found_type, _code) = ptable_uuid_to_flag_entry(part_info.get('type'))
if found_type == 'extended':
found_size_bytes = int(part_info['size']) * 512
@@ -784,30 +779,25 @@ def verify_size(devpath, expected_size_bytes, sfdisk_info=None):
raise RuntimeError(msg)
-def verify_ptable_flag(devpath, expected_flag, sfdisk_info=None):
+def verify_ptable_flag(devpath, expected_flag, label, part_info):
if (expected_flag not in SGDISK_FLAGS.keys()) and (expected_flag not in
MSDOS_FLAGS.keys()):
raise RuntimeError(
'Cannot verify unknown partition flag: %s' % expected_flag)
- if not sfdisk_info:
- sfdisk_info = block.sfdisk_info(devpath)
-
- entry = block.get_partition_sfdisk_info(devpath, sfdisk_info=sfdisk_info)
- LOG.debug("Device %s ptable entry: %s", devpath, util.json_dumps(entry))
found_flag = None
- if (sfdisk_info['label'] in ('dos', 'msdos')):
+ if (label in ('dos', 'msdos')):
if expected_flag == 'boot':
- found_flag = 'boot' if entry.get('bootable') is True else None
+ found_flag = 'boot' if part_info.get('bootable') is True else None
elif expected_flag == 'extended':
- (found_flag, _code) = ptable_uuid_to_flag_entry(entry['type'])
+ (found_flag, _code) = ptable_uuid_to_flag_entry(part_info['type'])
elif expected_flag == 'logical':
(_parent, partnumber) = block.get_blockdev_for_partition(devpath)
found_flag = 'logical' if int(partnumber) > 4 else None
# gpt and msdos primary partitions look up flag by entry['type']
if found_flag is None:
- (found_flag, _code) = ptable_uuid_to_flag_entry(entry['type'])
+ (found_flag, _code) = ptable_uuid_to_flag_entry(part_info['type'])
msg = (
'Verifying %s partition flag, expecting %s, found %s' % (
devpath, expected_flag, found_flag))
@@ -816,16 +806,13 @@ def verify_ptable_flag(devpath, expected_flag, sfdisk_info=None):
raise RuntimeError(msg)
-def partition_verify_sfdisk(devpath, info):
- verify_exists(devpath)
- sfdisk_info = block.sfdisk_info(devpath)
- if not sfdisk_info:
- raise RuntimeError('Failed to extract sfdisk info from %s' % devpath)
- verify_size(devpath, int(util.human2bytes(info['size'])),
- sfdisk_info=sfdisk_info)
- expected_flag = info.get('flag')
+def partition_verify_sfdisk(part_action, label, sfdisk_part_info):
+ devpath = sfdisk_part_info['node']
+ verify_size(
+ devpath, int(util.human2bytes(part_action['size'])), sfdisk_part_info)
+ expected_flag = part_action.get('flag')
if expected_flag:
- verify_ptable_flag(devpath, info['flag'], sfdisk_info=sfdisk_info)
+ verify_ptable_flag(devpath, expected_flag, label, sfdisk_part_info)
def partition_verify_fdasd(disk_path, partnumber, info):
@@ -937,7 +924,9 @@ def partition_handler(info, storage_config):
if disk_ptable == 'vtoc':
partition_verify_fdasd(disk, partnumber, info)
else:
- partition_verify_sfdisk(part_path, info)
+ sfdisk_info = block.sfdisk_info(disk)
+ part_info = block.get_partition_sfdisk_info(part_path, sfdisk_info)
+ partition_verify_sfdisk(info, sfdisk_info['label'], part_info)
LOG.debug(
'%s partition %s already present, skipping create',
disk, partnumber)
diff --git a/tests/unittests/test_commands_block_meta.py b/tests/unittests/test_commands_block_meta.py
index c3f8d14..4dad39d 100644
--- a/tests/unittests/test_commands_block_meta.py
+++ b/tests/unittests/test_commands_block_meta.py
@@ -2570,8 +2570,6 @@ class TestPartitionVerifySfdisk(CiTestCase):
def setUp(self):
super(TestPartitionVerifySfdisk, self).setUp()
base = 'curtin.commands.block_meta.'
- self.add_patch(base + 'verify_exists', 'm_verify_exists')
- self.add_patch(base + 'block.sfdisk_info', 'm_block_sfdisk_info')
self.add_patch(base + 'verify_size', 'm_verify_size')
self.add_patch(base + 'verify_ptable_flag', 'm_verify_ptable_flag')
self.info = {
@@ -2586,34 +2584,29 @@ class TestPartitionVerifySfdisk(CiTestCase):
self.devpath = self.random_string()
def test_partition_verify_sfdisk(self):
- block_meta.partition_verify_sfdisk(self.devpath, self.info)
- self.assertEqual(
- [call(self.devpath)],
- self.m_verify_exists.call_args_list)
- self.assertEqual(
- [call(self.devpath)],
- self.m_block_sfdisk_info.call_args_list)
+ devpath = self.random_string()
+ sfdisk_part_info = {
+ 'node': devpath,
+ }
+ label = self.random_string()
+ block_meta.partition_verify_sfdisk(self.info, label, sfdisk_part_info)
self.assertEqual(
- [call(self.devpath, self.part_size,
- sfdisk_info=self.m_block_sfdisk_info.return_value)],
+ [call(devpath, self.part_size, sfdisk_part_info)],
self.m_verify_size.call_args_list)
self.assertEqual(
- [call(self.devpath, self.info['flag'],
- sfdisk_info=self.m_block_sfdisk_info.return_value)],
+ [call(devpath, self.info['flag'], label, sfdisk_part_info)],
self.m_verify_ptable_flag.call_args_list)
def test_partition_verify_skips_ptable_no_flag(self):
del self.info['flag']
- block_meta.partition_verify_sfdisk(self.devpath, self.info)
- self.assertEqual(
- [call(self.devpath)],
- self.m_verify_exists.call_args_list)
- self.assertEqual(
- [call(self.devpath)],
- self.m_block_sfdisk_info.call_args_list)
+ devpath = self.random_string()
+ sfdisk_part_info = {
+ 'node': devpath,
+ }
+ label = self.random_string()
+ block_meta.partition_verify_sfdisk(self.info, label, sfdisk_part_info)
self.assertEqual(
- [call(self.devpath, self.part_size,
- sfdisk_info=self.m_block_sfdisk_info.return_value)],
+ [call(devpath, self.part_size, sfdisk_part_info)],
self.m_verify_size.call_args_list)
self.assertEqual([], self.m_verify_ptable_flag.call_args_list)
@@ -2741,15 +2734,17 @@ class TestVerifyPtableFlag(CiTestCase):
def test_verify_ptable_flag_finds_boot_on_gpt(self):
devpath = '/dev/vda15'
expected_flag = 'boot'
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_gpt)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'gpt',
+ self.sfdisk_info_gpt['partitions'][2])
def test_verify_ptable_flag_raises_exception_missing_flag(self):
devpath = '/dev/vda1'
expected_flag = 'boot'
with self.assertRaises(RuntimeError):
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_gpt)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'gpt',
+ self.sfdisk_info_gpt['partitions'][0])
def test_verify_ptable_flag_raises_exception_invalid_flag(self):
devpath = '/dev/vda1'
@@ -2757,52 +2752,49 @@ class TestVerifyPtableFlag(CiTestCase):
self.assertNotIn(expected_flag, block_meta.SGDISK_FLAGS.keys())
self.assertNotIn(expected_flag, block_meta.MSDOS_FLAGS.keys())
with self.assertRaises(RuntimeError):
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_gpt)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'gpt',
+ self.sfdisk_info_gpt['partitions'][0])
def test_verify_ptable_flag_checks_bootable_not_table_type(self):
devpath = '/dev/vdb1'
expected_flag = 'boot'
- del self.sfdisk_info_dos['partitions'][0]['bootable']
+ partinfo = self.sfdisk_info_dos['partitions'][0]
+ del partinfo['bootable']
self.sfdisk_info_dos['partitions'][0]['type'] = '0x80'
with self.assertRaises(RuntimeError):
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_dos)
-
- def test_verify_ptable_flag_calls_block_sfdisk_if_info_none(self):
- devpath = '/dev/vda15'
- expected_flag = 'boot'
- self.m_block_sfdisk_info.return_value = self.sfdisk_info_gpt
- block_meta.verify_ptable_flag(devpath, expected_flag, sfdisk_info=None)
- self.assertEqual(
- [call(devpath)],
- self.m_block_sfdisk_info.call_args_list)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'dos', partinfo)
def test_verify_ptable_flag_finds_boot_on_msdos(self):
devpath = '/dev/vdb1'
expected_flag = 'boot'
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_dos)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'dos',
+ self.sfdisk_info_dos['partitions'][0])
def test_verify_ptable_flag_finds_linux_on_dos_primary_partition(self):
devpath = '/dev/vdb2'
expected_flag = 'linux'
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_dos)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'dos',
+ self.sfdisk_info_dos['partitions'][1])
def test_verify_ptable_flag_finds_dos_extended_partition(self):
devpath = '/dev/vdb3'
expected_flag = 'extended'
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_dos)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'dos',
+ self.sfdisk_info_dos['partitions'][2])
def test_verify_ptable_flag_finds_dos_logical_partition(self):
devpath = '/dev/vdb5'
expected_flag = 'logical'
self.m_block_get_blockdev_for_partition.return_value = (
('/dev/vdb', '5'))
- block_meta.verify_ptable_flag(devpath, expected_flag,
- sfdisk_info=self.sfdisk_info_dos)
+ block_meta.verify_ptable_flag(
+ devpath, expected_flag, 'dos',
+ self.sfdisk_info_dos['partitions'][4])
class TestGetDevicePathsFromStorageConfig(CiTestCase):
--
Mailing list: https://launchpad.net/~curtin-dev
Post to : curtin-dev@lists.launchpad.net
Unsubscribe : https://launchpad.net/~curtin-dev
More help : https://help.launchpad.net/ListHelp
