Merge ~mwhudson/curtin:partition-verify-refactor into curtin:master

[Michael Hudson-Doyle]

block_meta: refactor partition verification slightly

I want to call these functions from the new partitioning code and this
tweaking makes this easier.

[Server Team CI bot (server-team-bot)]

PASSED: Continuous integration, rev:99470e340fc2a024f57f6dda52a66880be4940c3
https://jenkins.ubuntu.com/server/job/curtin-ci/202/
Executed test runs:
    SUCCESS: https://jenkins.ubuntu.com/server/job/curtin-ci/nodes=metal-amd64/202/
    SUCCESS: https://jenkins.ubuntu.com/server/job/curtin-ci/nodes=metal-arm64/202/
    SUCCESS: https://jenkins.ubuntu.com/server/job/curtin-ci/nodes=metal-ppc64el/202/
    SUCCESS: https://jenkins.ubuntu.com/server/job/curtin-ci/nodes=metal-s390x/202/

Click here to trigger a rebuild:
https://jenkins.ubuntu.com/server/job/curtin-ci/202//rebuild

review: Approve (continuous-integration)

[Dan Bungert (dbungert)]

review: Approve

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):