From a19735b4d8f2fee1e15c1982a6f31663d47ccc96 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Fri, 13 Sep 2024 10:55:03 +0200 Subject: [PATCH 1/4] storage: add MatchDirective type and use it as a type hint Instead of using inconsistent type hints for match directives, we now declare a MatchDirective type and use it everywhere. Signed-off-by: Olivier Gayot --- subiquity/models/filesystem.py | 46 +++++++++++++++++----- subiquity/server/controllers/filesystem.py | 3 +- 2 files changed, 39 insertions(+), 10 deletions(-) diff --git a/subiquity/models/filesystem.py b/subiquity/models/filesystem.py index 72bffc8d5..1cadeb11b 100644 --- a/subiquity/models/filesystem.py +++ b/subiquity/models/filesystem.py @@ -25,7 +25,18 @@ import secrets import tempfile from abc import ABC, abstractmethod -from typing import Callable, Dict, List, Optional, Sequence, Set, Tuple, Union +from typing import ( + Callable, + Dict, + List, + Literal, + Optional, + Sequence, + Set, + Tuple, + TypedDict, + Union, +) import attr import more_itertools @@ -55,6 +66,26 @@ class NotFinalPartitionError(Exception): the last one.""" +# a match directive is a dict that specifies +# * zero or more keys to filter on +# * an optional sort on size +MatchDirective = TypedDict( + "MatchDirective", + { + "serial": str, + "model": str, + "vendor": str, + "path": str, + "id_path": str, + "devpath": str, + "ssd": bool, + "size": Literal["smallest", "largest"], + "install-media": bool, + }, + total=False, +) + + @attr.s(auto_attribs=True) class RecoveryKeyHandler: # Where to store the key on the live system @@ -1650,7 +1681,7 @@ def load_server_data(self, status: StorageResponse): status.config, blockdevs=None, is_probe_data=False ) - def _make_matchers(self, match: dict) -> Sequence[Callable]: + def _make_matchers(self, match: MatchDirective) -> Sequence[Callable]: def _udev_val(disk, key): return self._probe_data["blockdev"].get(disk.path, {}).get(key, "") @@ -1709,7 +1740,7 @@ def match_nonzero_size(disk): return matchers - def _sorted_matches(self, disks: Sequence[_Device], match: dict): + def _sorted_matches(self, disks: Sequence[_Device], match: MatchDirective): # sort first on the sort_key. Objective here is that if we are falling # back to arbitrary disk selection, we're at least consistent in what # disk we arbitrarily select across runs @@ -1723,18 +1754,15 @@ def _sorted_matches(self, disks: Sequence[_Device], match: dict): disks.sort(key=lambda d: d.size, reverse=True) return disks - def _filtered_matches(self, disks: Sequence[_Device], match: dict): + def _filtered_matches(self, disks: Sequence[_Device], match: MatchDirective): matchers = self._make_matchers(match) return [disk for disk in disks if all(match_fn(disk) for match_fn in matchers)] def disk_for_match( - self, disks: Sequence[_Device], match: dict | Sequence[dict] + self, disks: Sequence[_Device], match: MatchDirective | Sequence[MatchDirective] ) -> _Device: - # a match directive is a dict, or a list of dicts, that specify - # * zero or more keys to filter on - # * an optional sort on size log.info(f"considering {disks} for {match}") - if isinstance(match, dict): + if not isinstance(match, Sequence): match = [match] for m in match: candidates = self._filtered_matches(disks, m) diff --git a/subiquity/server/controllers/filesystem.py b/subiquity/server/controllers/filesystem.py index 209b6d6f2..68c52e0b6 100644 --- a/subiquity/server/controllers/filesystem.py +++ b/subiquity/server/controllers/filesystem.py @@ -71,6 +71,7 @@ from subiquity.models.filesystem import ( LVM_LogicalVolume, LVM_VolGroup, + MatchDirective, MiB, Partition, Raid, @@ -1511,7 +1512,7 @@ async def _probe_firmware(self) -> None: self.model.detected_supports_nvme_tcp_booting = assume_supported def get_bootable_matching_disk( - self, match: dict[str, str] | Sequence[dict[str, str]] + self, match: MatchDirective | Sequence[MatchDirective] ): """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. From 991f6629b264c684f4652692421f712efa4bfab7 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Fri, 13 Sep 2024 11:31:02 +0200 Subject: [PATCH 2/4] storage: fix wrong return type for disk_for_match disk_for_match either returns a matching disk ; or returns None. Ensure that the return type hint is "Optional". Signed-off-by: Olivier Gayot --- subiquity/models/filesystem.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/subiquity/models/filesystem.py b/subiquity/models/filesystem.py index 1cadeb11b..f95e99d18 100644 --- a/subiquity/models/filesystem.py +++ b/subiquity/models/filesystem.py @@ -1760,7 +1760,7 @@ def _filtered_matches(self, disks: Sequence[_Device], match: MatchDirective): def disk_for_match( self, disks: Sequence[_Device], match: MatchDirective | Sequence[MatchDirective] - ) -> _Device: + ) -> Optional[_Device]: log.info(f"considering {disks} for {match}") if not isinstance(match, Sequence): match = [match] From fc5e4bbc497f69ccd99664c69795c4399fab34c5 Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Fri, 13 Sep 2024 12:04:09 +0200 Subject: [PATCH 3/4] storage: add helper functions to match multiple disks Signed-off-by: Olivier Gayot --- subiquity/models/filesystem.py | 25 ++++-- subiquity/models/tests/test_filesystem.py | 90 +++++++++++-------- subiquity/server/controllers/filesystem.py | 23 +++-- .../controllers/tests/test_filesystem.py | 14 ++- 4 files changed, 102 insertions(+), 50 deletions(-) diff --git a/subiquity/models/filesystem.py b/subiquity/models/filesystem.py index f95e99d18..fc4e28157 100644 --- a/subiquity/models/filesystem.py +++ b/subiquity/models/filesystem.py @@ -1758,9 +1758,9 @@ def _filtered_matches(self, disks: Sequence[_Device], match: MatchDirective): matchers = self._make_matchers(match) return [disk for disk in disks if all(match_fn(disk) for match_fn in matchers)] - def disk_for_match( + def _matching_disks( self, disks: Sequence[_Device], match: MatchDirective | Sequence[MatchDirective] - ) -> Optional[_Device]: + ) -> tuple[list[_Device], Optional[MatchDirective]]: log.info(f"considering {disks} for {match}") if not isinstance(match, Sequence): match = [match] @@ -1768,10 +1768,25 @@ def disk_for_match( candidates = self._filtered_matches(disks, m) candidates = self._sorted_matches(candidates, m) if candidates: - log.info(f"For match {m}, using the first candidate from {candidates}") - return candidates[0] + return candidates, m log.info(f"No devices satisfy criteria {match}") - return None + return [], None + + def disks_for_match( + self, disks: Sequence[_Device], match: MatchDirective | Sequence[MatchDirective] + ) -> list[_Device]: + candidates, _ = self._matching_disks(disks, match) + return candidates + + def disk_for_match( + self, disks: Sequence[_Device], match: MatchDirective | Sequence[MatchDirective] + ) -> Optional[_Device]: + candidates, m = self._matching_disks(disks, match) + if candidates: + log.info(f"For match {m}, using the first candidate from {candidates}") + return candidates[0] + else: + return None def assign_omitted_offsets(self): """Assign offsets to partitions that do not already have one. diff --git a/subiquity/models/tests/test_filesystem.py b/subiquity/models/tests/test_filesystem.py index f3c781fe7..e2f13bfaa 100644 --- a/subiquity/models/tests/test_filesystem.py +++ b/subiquity/models/tests/test_filesystem.py @@ -1870,24 +1870,28 @@ def test_empty_match_directive(self): d2 = make_disk(m) # this test relies heavily on the assumptions in make_disk - actual = m.disk_for_match([d1, d2], {}) - self.assertEqual(d1, actual) - actual = m.disk_for_match([d2, d1], {}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d1, d2], {})) + self.assertEqual([d1, d2], m.disks_for_match([d1, d2], {})) + self.assertEqual(d1, m.disk_for_match([d2, d1], {})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {})) def test_sort_largest(self): m = make_model() d100 = make_disk(m, size=100 << 30, serial="s1", path="/dev/d1") d200 = make_disk(m, size=200 << 30, serial="s2", path="/dev/d2") - actual = m.disk_for_match([d100, d200], {"size": "largest"}) - self.assertEqual(d200, actual) + self.assertEqual(d200, m.disk_for_match([d100, d200], {"size": "largest"})) + self.assertEqual( + [d200, d100], m.disks_for_match([d100, d200], {"size": "largest"}) + ) def test_sort_smallest(self): m = make_model() d200 = make_disk(m, size=200 << 30) d100 = make_disk(m, size=100 << 30) - actual = m.disk_for_match([d200, d100], {"size": "smallest"}) - self.assertEqual(d100, actual) + self.assertEqual(d100, m.disk_for_match([d200, d100], {"size": "smallest"})) + self.assertEqual( + [d100, d200], m.disks_for_match([d200, d100], {"size": "smallest"}) + ) @parameterized.expand(match_sort_criteria) def test_sort_serial(self, sort_criteria: str): @@ -1896,48 +1900,48 @@ def test_sort_serial(self, sort_criteria: str): d2 = make_disk(m, serial="s2", path=None, wwn=None) # while the size sort is reversed when doing largest, # we pre-sort on the other criteria, and stable sort helps out - actual = m.disk_for_match([d2, d1], {"size": sort_criteria}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d2, d1], {"size": sort_criteria})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {"size": sort_criteria})) @parameterized.expand(match_sort_criteria) def test_sort_path(self, sort_criteria: str): m = make_model() d1 = make_disk(m, serial=None, path="/dev/d1", wwn=None) d2 = make_disk(m, serial=None, path="/dev/d2", wwn=None) - actual = m.disk_for_match([d2, d1], {"size": sort_criteria}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d2, d1], {"size": sort_criteria})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {"size": sort_criteria})) @parameterized.expand(match_sort_criteria) def test_sort_wwn(self, sort_criteria: str): m = make_model() d1 = make_disk(m, serial=None, path=None, wwn="w1") d2 = make_disk(m, serial=None, path=None, wwn="w2") - actual = m.disk_for_match([d2, d1], {"size": sort_criteria}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d2, d1], {"size": sort_criteria})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {"size": sort_criteria})) @parameterized.expand(match_sort_criteria) def test_sort_wwn_wins(self, sort_criteria: str): m = make_model() d1 = make_disk(m, serial="s2", path="/dev/d2", wwn="w1") d2 = make_disk(m, serial="s1", path="/dev/d1", wwn="w2") - actual = m.disk_for_match([d2, d1], {"size": sort_criteria}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d2, d1], {"size": sort_criteria})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {"size": sort_criteria})) @parameterized.expand(match_sort_criteria) def test_sort_serial_wins(self, sort_criteria: str): m = make_model() d1 = make_disk(m, serial="s1", path="/dev/d2", wwn="w") d2 = make_disk(m, serial="s2", path="/dev/d1", wwn="w") - actual = m.disk_for_match([d2, d1], {"size": sort_criteria}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d2, d1], {"size": sort_criteria})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {"size": sort_criteria})) @parameterized.expand(match_sort_criteria) def test_sort_path_wins(self, sort_criteria: str): m = make_model() d1 = make_disk(m, serial="s", path="/dev/d1", wwn="w") d2 = make_disk(m, serial="s", path="/dev/d2", wwn="w") - actual = m.disk_for_match([d2, d1], {"size": sort_criteria}) - self.assertEqual(d1, actual) + self.assertEqual(d1, m.disk_for_match([d2, d1], {"size": sort_criteria})) + self.assertEqual([d1, d2], m.disks_for_match([d2, d1], {"size": sort_criteria})) def test_sort_raid(self): m = make_model() @@ -1947,8 +1951,8 @@ def test_sort_raid(self): d2_2 = make_disk(m, size=200 << 30) r1 = make_raid(m, disks={d1_1, d1_2}) r2 = make_raid(m, disks={d2_1, d2_2}) - actual = m.disk_for_match([r1, r2], {"size": "largest"}) - self.assertEqual(r2, actual) + self.assertEqual(r2, m.disk_for_match([r1, r2], {"size": "largest"})) + self.assertEqual([r2, r1], m.disks_for_match([r1, r2], {"size": "largest"})) @parameterized.expand(match_sort_criteria) def test_sort_raid_on_disks(self, sort_criteria: str): @@ -1959,23 +1963,23 @@ def test_sort_raid_on_disks(self, sort_criteria: str): d2_2 = make_disk(m, serial=None, path=None, wwn="w2_2") r1 = make_raid(m, disks={d1_1, d1_2}) r2 = make_raid(m, disks={d2_1, d2_2}) - actual = m.disk_for_match([r1, r2], {"size": sort_criteria}) - self.assertEqual(r1, actual) + self.assertEqual(r1, m.disk_for_match([r1, r2], {"size": sort_criteria})) + self.assertEqual([r1, r2], m.disks_for_match([r1, r2], {"size": sort_criteria})) def test_skip_empty(self): m = make_model() d0 = make_disk(m, size=0) d100 = make_disk(m, size=100 << 30) - actual = m.disk_for_match([d0, d100], {"size": "smallest"}) - self.assertEqual(d100, actual) + self.assertEqual(d100, m.disk_for_match([d0, d100], {"size": "smallest"})) + self.assertEqual([d100], m.disks_for_match([d0, d100], {"size": "smallest"})) def test_skip_in_use_size(self): m = make_model() d100 = make_disk(m, size=100 << 30) d200 = make_disk(m, size=200 << 30) d100._has_in_use_partition = True - actual = m.disk_for_match([d100, d200], {"size": "smallest"}) - self.assertEqual(d200, actual) + self.assertEqual(d200, m.disk_for_match([d100, d200], {"size": "smallest"})) + self.assertEqual([d200], m.disks_for_match([d100, d200], {"size": "smallest"})) def test_skip_in_use_ssd(self): m = make_model() @@ -1984,8 +1988,12 @@ def test_skip_in_use_ssd(self): d_in_use._has_in_use_partition = True d_in_use.info_for_display = Mock(return_value={"rotational": "false"}) d_not_used.info_for_display = Mock(return_value={"rotational": "false"}) - actual = m.disk_for_match([d_in_use, d_not_used], {"ssd": True}) - self.assertEqual(d_not_used, actual) + self.assertEqual( + d_not_used, m.disk_for_match([d_in_use, d_not_used], {"ssd": True}) + ) + self.assertEqual( + [d_not_used], m.disks_for_match([d_in_use, d_not_used], {"ssd": True}) + ) def test_matcher_serial(self): m = make_model() @@ -1993,7 +2001,9 @@ def test_matcher_serial(self): d2 = make_disk(m, serial="2") fake_up_blockdata(m) self.assertEqual(d1, m.disk_for_match([d1, d2], {"serial": "1"})) + self.assertEqual([d1], m.disks_for_match([d1, d2], {"serial": "1"})) self.assertEqual(d2, m.disk_for_match([d1, d2], {"serial": "2"})) + self.assertEqual([d2], m.disks_for_match([d1, d2], {"serial": "2"})) def test_matcher_model(self): m = make_model() @@ -2002,7 +2012,9 @@ def test_matcher_model(self): d2 = make_disk(m) fake_up_blockdata_disk(d2, ID_MODEL="m2") self.assertEqual(d1, m.disk_for_match([d1, d2], {"model": "m1"})) + self.assertEqual([d1], m.disks_for_match([d1, d2], {"model": "m1"})) self.assertEqual(d2, m.disk_for_match([d1, d2], {"model": "m2"})) + self.assertEqual([d2], m.disks_for_match([d1, d2], {"model": "m2"})) def test_matcher_vendor(self): m = make_model() @@ -2011,7 +2023,9 @@ def test_matcher_vendor(self): d2 = make_disk(m) fake_up_blockdata_disk(d2, ID_VENDOR="v2") self.assertEqual(d1, m.disk_for_match([d1, d2], {"vendor": "v1"})) + self.assertEqual([d1], m.disks_for_match([d1, d2], {"vendor": "v1"})) self.assertEqual(d2, m.disk_for_match([d1, d2], {"vendor": "v2"})) + self.assertEqual([d2], m.disks_for_match([d1, d2], {"vendor": "v2"})) def test_matcher_path(self): m = make_model() @@ -2019,7 +2033,9 @@ def test_matcher_path(self): vdb = make_disk(m, path="/dev/vdb") fake_up_blockdata(m) self.assertEqual(vda, m.disk_for_match([vda, vdb], {"path": "/dev/vda"})) + self.assertEqual([vda], m.disks_for_match([vda, vdb], {"path": "/dev/vda"})) self.assertEqual(vdb, m.disk_for_match([vda, vdb], {"path": "/dev/vdb"})) + self.assertEqual([vdb], m.disks_for_match([vda, vdb], {"path": "/dev/vdb"})) def test_matcher_id_path(self): m = make_model() @@ -2027,10 +2043,10 @@ def test_matcher_id_path(self): fake_up_blockdata_disk(vda, ID_PATH="pci-0000:00:00.0-nvme-vda") vdb = make_disk(m) fake_up_blockdata_disk(vdb, ID_PATH="pci-0000:00:00.0-nvme-vdb") - actual = m.disk_for_match([vda, vdb], {"id_path": "*vda"}) - self.assertEqual(vda, actual) - actual = m.disk_for_match([vda, vdb], {"id_path": "*vdb"}) - self.assertEqual(vdb, actual) + self.assertEqual(vda, m.disk_for_match([vda, vdb], {"id_path": "*vda"})) + self.assertEqual([vda], m.disks_for_match([vda, vdb], {"id_path": "*vda"})) + self.assertEqual(vdb, m.disk_for_match([vda, vdb], {"id_path": "*vdb"})) + self.assertEqual([vdb], m.disks_for_match([vda, vdb], {"id_path": "*vdb"})) def test_matcher_install_media(self): m = make_model() @@ -2038,8 +2054,8 @@ def test_matcher_install_media(self): iso._has_in_use_partition = True disk = make_disk(m) fake_up_blockdata(m) - actual = m.disk_for_match([iso, disk], {"install-media": True}) - self.assertEqual(iso, actual) + self.assertEqual(iso, m.disk_for_match([iso, disk], {"install-media": True})) + self.assertEqual([iso], m.disks_for_match([iso, disk], {"install-media": True})) def test_match_from_list_first(self): m = make_model() @@ -2051,6 +2067,7 @@ def test_match_from_list_first(self): {"path": "/dev/vdb"}, ] self.assertEqual(vda, m.disk_for_match([vda, vdb], match)) + self.assertEqual([vda], m.disks_for_match([vda, vdb], match)) def test_match_from_list_second(self): m = make_model() @@ -2062,3 +2079,4 @@ def test_match_from_list_second(self): {"path": "/dev/vdb"}, ] self.assertEqual(vdb, m.disk_for_match([vda, vdb], match)) + self.assertEqual([vdb], m.disks_for_match([vda, vdb], match)) diff --git a/subiquity/server/controllers/filesystem.py b/subiquity/server/controllers/filesystem.py index 68c52e0b6..97fd95545 100644 --- a/subiquity/server/controllers/filesystem.py +++ b/subiquity/server/controllers/filesystem.py @@ -1511,18 +1511,29 @@ async def _probe_firmware(self) -> None: self.model.detected_supports_nvme_tcp_booting = assume_supported - def get_bootable_matching_disk( + def get_bootable_matching_disks( self, match: MatchDirective | Sequence[MatchDirective] - ): + ) -> list[_Device]: """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. + we have a plan to boot, and return them. 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: + matching_disks = self.model.disks_for_match(disks, match) + if not matching_disks: raise AutoinstallError(f"Failed to find matching device for {match}") - return disk + return matching_disks + + def get_bootable_matching_disk( + self, match: MatchDirective | Sequence[MatchDirective] + ) -> _Device: + """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.""" + matching_disks = self.get_bootable_matching_disks(match) + assert matching_disks + return matching_disks[0] async def run_autoinstall_guided(self, layout): name = layout["name"] diff --git a/subiquity/server/controllers/tests/test_filesystem.py b/subiquity/server/controllers/tests/test_filesystem.py index ba0b9ec03..5a7fa64d8 100644 --- a/subiquity/server/controllers/tests/test_filesystem.py +++ b/subiquity/server/controllers/tests/test_filesystem.py @@ -1860,12 +1860,20 @@ def setUp(self): def test_no_match_raises_AutoinstallError(self): with self.assertRaises(AutoinstallError): self.fsc.get_bootable_matching_disk({"size": "largest"}) + with self.assertRaises(AutoinstallError): + self.fsc.get_bootable_matching_disks({"size": "largest"}) def test_two_matches(self): - make_disk(self.fsc.model, size=10 << 30) + d1 = 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) + self.assertEqual(d2, self.fsc.get_bootable_matching_disk({"size": "largest"})) + self.assertEqual(d1, self.fsc.get_bootable_matching_disk({"size": "smallest"})) + self.assertEqual( + [d2, d1], self.fsc.get_bootable_matching_disks({"size": "largest"}) + ) + self.assertEqual( + [d1, d2], self.fsc.get_bootable_matching_disks({"size": "smallest"}) + ) @mock.patch("subiquity.common.filesystem.boot.can_be_boot_device") def test_actually_match_raid(self, m_cbb): From 9fd3c924791b737d6b6cce04a35e3b82d45b535b Mon Sep 17 00:00:00 2001 From: Olivier Gayot Date: Fri, 13 Sep 2024 12:04:27 +0200 Subject: [PATCH 4/4] storage: honor matchers in use_gap autoinstall mode When using the use_gap mode in autoinstall, one may want to ensure that we use a gap on a specific disk. Unfortunately, the match directives were silently ignored. Let's make sure they are honored. LP: #2080608 Signed-off-by: Olivier Gayot --- subiquity/server/controllers/filesystem.py | 5 +- .../controllers/tests/test_filesystem.py | 72 +++++++++++++++++++ 2 files changed, 75 insertions(+), 2 deletions(-) diff --git a/subiquity/server/controllers/filesystem.py b/subiquity/server/controllers/filesystem.py index 97fd95545..ca202d281 100644 --- a/subiquity/server/controllers/filesystem.py +++ b/subiquity/server/controllers/filesystem.py @@ -1624,8 +1624,9 @@ async def run_autoinstall_guided(self, layout): disk = self.get_bootable_matching_disk(match) target = GuidedStorageTargetReformat(disk_id=disk.id, allowed=[]) elif mode == "use_gap": - bootable = self.potential_boot_disks(with_reformatting=False) - gap = gaps.largest_gap(bootable) + match = layout.get("match", {}) + bootable_disks = self.get_bootable_matching_disks(match) + gap = gaps.largest_gap(bootable_disks) if not gap: raise Exception( "autoinstall cannot configure storage " diff --git a/subiquity/server/controllers/tests/test_filesystem.py b/subiquity/server/controllers/tests/test_filesystem.py index 5a7fa64d8..32edaa4bf 100644 --- a/subiquity/server/controllers/tests/test_filesystem.py +++ b/subiquity/server/controllers/tests/test_filesystem.py @@ -605,6 +605,78 @@ async def test__get_system_api_error_logged(self): self.assertIn("cannot load assertions for label", logs.output[0]) +class TestRunAutoinstallGuided(IsolatedAsyncioTestCase): + def setUp(self): + self.app = make_app() + self.app.opts.bootloader = None + self.fsc = FilesystemController(self.app) + self.model = self.fsc.model = make_model() + + # This is needed for examine_systems_task + self.app.base_model.source.current.type = "fsimage" + self.app.base_model.source.current.variations = { + "default": CatalogEntryVariation(path="", size=1), + } + + async def asyncSetUp(self): + self.fsc._examine_systems_task.start_sync() + + await self.fsc._examine_systems_task.wait() + + async def test_direct_use_gap__install_media(self): + """Match directives were previously not honored when using mode: use_gap. + This made it not possible for the OEM team to install to the + installation media. LP: #2080608""" + layout = { + "name": "direct", + "mode": "use_gap", + "match": { + "install-media": True, + }, + } + + # The matcher for "install-media": True looks for + # _has_in_use_partition. + iso = make_disk(self.model) + iso._has_in_use_partition = True + + make_disk(self.model) + + p_guided = mock.patch.object(self.fsc, "guided") + p_guided_choice_v2 = mock.patch( + "subiquity.server.controllers.filesystem.GuidedChoiceV2", + wraps=GuidedChoiceV2, + ) + p_largest_gap = mock.patch( + "subiquity.server.controllers.filesystem.gaps.largest_gap", + wraps=gaps.largest_gap, + ) + + with ( + p_guided as m_guided, + p_guided_choice_v2 as m_guided_choice_v2, + p_largest_gap as m_largest_gap, + ): + await self.fsc.run_autoinstall_guided(layout) + + # largest_gap will call itself recursively, so we should not expect a + # single call to it. + m_largest_gap.mock_calls[0] = mock.call([iso]) + + m_guided.assert_called_once() + m_guided_choice_v2.assert_called_once_with( + target=GuidedStorageTargetUseGap( + disk_id=iso.id, gap=gaps.largest_gap([iso]), allowed=[] + ), + capability=GuidedCapability.DIRECT, + password=mock.ANY, + recovery_key=mock.ANY, + sizing_policy=mock.ANY, + reset_partition=mock.ANY, + reset_partition_size=mock.ANY, + ) + + class TestGuided(IsolatedAsyncioTestCase): boot_expectations = [ (Bootloader.UEFI, "gpt", "/boot/efi"),