filesystem: autoinstall reformat_disk match raids

Raise AutoinstallError on failed match while we're at it, better than
the "disk is None" scenario today.

subiquity/server/controllers/filesystem.py

@@ -74,6 +74,7 @@ from subiquity.models.filesystem import (
     humanize_size,
 )
 from subiquity.server import snapdapi
+from subiquity.server.autoinstall import AutoinstallError
 from subiquity.server.controller import SubiquityController
 from subiquity.server.controllers.source import SEARCH_DRIVERS_AUTOINSTALL_DEFAULT
 from subiquity.server.mounter import Mounter
@@ -1327,6 +1328,17 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                 self.start_monitor()
             break
 
+    def get_bootable_matching_disk(self, match: dict[str, str]):
+        """given a match directive, find disks or disk-like devices for which
+        we have a plan to boot, and return the best matching one of those.
+        As match directives are autoinstall-supplied, raise AutoinstallError if
+        no matching disk is found."""
+        disks = self.potential_boot_disks(with_reformatting=True)
+        disk = self.model.disk_for_match(disks, match)
+        if disk is None:
+            raise AutoinstallError(f"Failed to find matching device for {match}")
+        return disk
+
     async def run_autoinstall_guided(self, layout):
         name = layout["name"]
         password = None
@@ -1368,7 +1380,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
                     raise Exception("cannot install this model unencrypted")
                 capability = GC.CORE_BOOT_UNENCRYPTED
             match = layout.get("match", {"size": "largest"})
-            disk = self.model.disk_for_match(self.model.all_disks(), match)
+            disk = self.get_bootable_matching_disk(match)
             mode = "reformat_disk"
         else:
             # this check is conceptually unnecessary but results in a
@@ -1413,7 +1425,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
 
         if mode == "reformat_disk":
             match = layout.get("match", {"size": "largest"})
-            disk = self.model.disk_for_match(self.model.all_disks(), match)
+            disk = self.get_bootable_matching_disk(match)
             target = GuidedStorageTargetReformat(disk_id=disk.id, allowed=[])
         elif mode == "use_gap":
             bootable = [

subiquity/server/controllers/tests/test_filesystem.py

@@ -51,8 +51,10 @@ from subiquity.models.tests.test_filesystem import (
     make_model,
     make_nvme_controller,
     make_partition,
+    make_raid,
 )
 from subiquity.server import snapdapi
+from subiquity.server.autoinstall import AutoinstallError
 from subiquity.server.controllers.filesystem import (
     DRY_RUN_RESET_SIZE,
     FilesystemController,
@@ -1549,3 +1551,31 @@ class TestCoreBootInstallMethods(IsolatedAsyncioTestCase):
             disallowed.reason,
             GuidedDisallowedCapabilityReason.CORE_BOOT_ENCRYPTION_UNAVAILABLE,
         )
+
+
+class TestMatchingDisks(IsolatedAsyncioTestCase):
+    def setUp(self):
+        bootloader = Bootloader.UEFI
+        self.app = make_app()
+        self.app.opts.bootloader = bootloader.value
+        self.fsc = FilesystemController(app=self.app)
+        self.fsc.model = make_model(bootloader)
+
+    def test_no_match_raises_AutoinstallError(self):
+        with self.assertRaises(AutoinstallError):
+            self.fsc.get_bootable_matching_disk({"size": "largest"})
+
+    def test_two_matches(self):
+        make_disk(self.fsc.model, size=10 << 30)
+        d2 = make_disk(self.fsc.model, size=20 << 30)
+        actual = self.fsc.get_bootable_matching_disk({"size": "largest"})
+        self.assertEqual(d2, actual)
+
+    @mock.patch("subiquity.common.filesystem.boot.can_be_boot_device")
+    def test_actually_match_raid(self, m_cbb):
+        r1 = make_raid(self.fsc.model)
+        m_cbb.return_value = True
+        # a size based check will make the raid not largest because of 65MiB of
+        # overhead
+        actual = self.fsc.get_bootable_matching_disk({"path": "/dev/md/*"})
+        self.assertEqual(r1, actual)