convert all "is this action possible on this device" code over to new style

This commit is contained in:
Michael Hudson-Doyle 2021-05-31 13:57:11 +12:00
parent 22bf1a0321
commit b3e124d26f
5 changed files with 186 additions and 169 deletions

View File

@ -25,6 +25,7 @@ from subiquity.models.filesystem import (
LVM_VolGroup,
Partition,
Raid,
raidlevels_by_value,
)
@ -62,7 +63,14 @@ class DeviceAction(enum.Enum):
return _supported_actions(device)
def can(self, device):
return _checkers[self](device)
assert self in self.supported(device)
r = _checkers[self](device)
if isinstance(r, bool):
return r, None
elif isinstance(r, str):
return False, r
else:
return r
@functools.singledispatch
@ -134,7 +142,7 @@ def _disk_info(disk):
_can_edit = make_checker(DeviceAction.EDIT)
def _generic_edit(device):
def _can_edit_generic(device):
cd = device.constructed_device()
if cd is None:
return True
@ -146,8 +154,8 @@ def _generic_edit(device):
cdname=cd.label)
_can_edit.register(Partition, _generic_edit)
_can_edit.register(LVM_LogicalVolume, _generic_edit)
_can_edit.register(Partition, _can_edit_generic)
_can_edit.register(LVM_LogicalVolume, _can_edit_generic)
@_can_edit.register(Raid)
@ -159,7 +167,7 @@ def _can_edit_raid(raid):
"Cannot edit {raidlabel} because it has partitions.").format(
raidlabel=raid.label)
else:
return _generic_edit(raid)
return _can_edit_generic(raid)
@_can_edit.register(LVM_VolGroup)
@ -172,4 +180,172 @@ def _can_edit_vg(vg):
"volumes.").format(
vglabel=vg.label)
else:
return _generic_edit(vg)
return _can_edit_generic(vg)
_can_reformat = make_checker(DeviceAction.REFORMAT)
@_can_reformat.register(Disk)
@_can_reformat.register(Raid)
def _can_reformat_device(device):
if len(device._partitions) == 0:
return False
for p in device._partitions:
if p._constructed_device is not None:
return False
return True
_can_partition = make_checker(DeviceAction.PARTITION)
@_can_partition.register(Disk)
@_can_partition.register(Raid)
def _can_partition_device(device):
if device._has_preexisting_partition():
return False
if device.free_for_partitions <= 0:
return False
# We only create msdos partition tables with FBA dasds, which
# only support 3 partitions. As and when we support editing
# partition msdos tables we'll need to be more clever here.
if device.ptable in ['vtoc', 'msdos'] and len(device._partitions) >= 3:
return False
return True
_can_create_lv = make_checker(DeviceAction.CREATE_LV)
@_can_create_lv.register(LVM_VolGroup)
def _can_create_lv_vg(vg):
return not vg.preserve and vg.free_for_partitions > 0
_can_format = make_checker(DeviceAction.FORMAT)
@_can_format.register(Disk)
@_can_format.register(Raid)
def _can_format_device(device):
return len(device._partitions) == 0 and device._constructed_device is None
_can_remove = make_checker(DeviceAction.REMOVE)
@_can_remove.register(Disk)
@_can_remove.register(Partition)
@_can_remove.register(Raid)
def _can_remove_device(device):
cd = device.constructed_device()
if cd is None:
return False
if cd.preserve:
return _("Cannot remove {selflabel} from pre-existing {cdtype} "
"{cdlabel}.").format(
selflabel=device.label,
cdtype=cd.desc(),
cdlabel=cd.label)
if isinstance(cd, Raid):
if device in cd.spare_devices:
return True
min_devices = raidlevels_by_value[cd.raidlevel].min_devices
if len(cd.devices) == min_devices:
return _(
"Removing {selflabel} would leave the {cdtype} {cdlabel} with "
"less than {min_devices} devices.").format(
selflabel=device.label,
cdtype=cd.desc(),
cdlabel=cd.label,
min_devices=min_devices)
elif isinstance(cd, LVM_VolGroup):
if len(cd.devices) == 1:
return _(
"Removing {selflabel} would leave the {cdtype} {cdlabel} with "
"no devices.").format(
selflabel=device.label,
cdtype=cd.desc(),
cdlabel=cd.label)
return True
_can_delete = make_checker(DeviceAction.DELETE)
def _can_delete_generic(device):
cd = device.constructed_device()
if cd is None:
return True
return _(
"Cannot delete {selflabel} as it is part of the {cdtype} "
"{cdname}.").format(
selflabel=device.label,
cdtype=cd.desc(),
cdname=cd.label)
@_can_delete.register(Partition)
def _can_delete_partition(partition):
if partition.device._has_preexisting_partition():
return _("Cannot delete a single partition from a device that "
"already has partitions.")
if partition.is_bootloader_partition:
return _("Cannot delete required bootloader partition")
return _can_delete_generic(partition)
@_can_delete.register(Raid)
@_can_delete.register(LVM_VolGroup)
def _can_delete_raid_vg(device):
mounted_partitions = 0
for p in device._partitions:
if p.fs() and p.fs().mount():
mounted_partitions += 1
elif p.constructed_device():
cd = p.constructed_device()
return _(
"Cannot delete {devicelabel} as partition {partnum} is part "
"of the {cdtype} {cdname}.").format(
devicelabel=device.label,
partnum=p._number,
cdtype=cd.desc(),
cdname=cd.label,
)
if mounted_partitions > 1:
return _(
"Cannot delete {devicelabel} because it has {count} mounted "
"partitions.").format(
devicelabel=device.label,
count=mounted_partitions)
elif mounted_partitions == 1:
return _(
"Cannot delete {devicelabel} because it has 1 mounted partition."
).format(devicelabel=device.label)
else:
return _can_delete_generic(device)
@_can_delete.register(LVM_LogicalVolume)
def _can_delete_lv(lv):
if lv.volgroup._has_preexisting_partition():
return _("Cannot delete a single logical volume from a volume "
"group that already has logical volumes.")
return True
_can_toggle_boot = make_checker(DeviceAction.TOGGLE_BOOT)
@_can_toggle_boot.register(Disk)
def _can_toggle_boot_disk(disk):
if disk._is_boot_device():
for disk2 in disk._m.all_disks():
if disk2 is not disk and disk2._is_boot_device():
return True
return False
elif disk._fs is not None or disk._constructed_device is not None:
return False
else:
return disk._can_be_boot_disk()

View File

@ -66,7 +66,7 @@ class TestFilesystemManipulator(unittest.TestCase):
continue
manipulator.add_boot_disk(disk)
self.assertFalse(
disk._can_TOGGLE_BOOT,
DeviceAction.TOGGLE_BOOT.can(disk)[0],
"add_boot_disk(disk) did not make _can_TOGGLE_BOOT false "
"with bootloader {}".format(bl))

View File

@ -419,51 +419,6 @@ def asdict(inst):
# in the FilesystemModel or FilesystemController classes.
def _generic_can_REMOVE(obj):
cd = obj.constructed_device()
if cd is None:
return False
if cd.preserve:
return _("Cannot remove {selflabel} from pre-existing {cdtype} "
"{cdlabel}.").format(
selflabel=obj.label,
cdtype=cd.desc(),
cdlabel=cd.label)
if isinstance(cd, Raid):
if obj in cd.spare_devices:
return True
min_devices = raidlevels_by_value[cd.raidlevel].min_devices
if len(cd.devices) == min_devices:
return _(
"Removing {selflabel} would leave the {cdtype} {cdlabel} with "
"less than {min_devices} devices.").format(
selflabel=obj.label,
cdtype=cd.desc(),
cdlabel=cd.label,
min_devices=min_devices)
elif isinstance(cd, LVM_VolGroup):
if len(cd.devices) == 1:
return _(
"Removing {selflabel} would leave the {cdtype} {cdlabel} with "
"no devices.").format(
selflabel=obj.label,
cdtype=cd.desc(),
cdlabel=cd.label)
return True
def _generic_can_DELETE(obj):
cd = obj.constructed_device()
if cd is None:
return True
return _(
"Cannot delete {selflabel} as it is part of the {cdtype} "
"{cdname}.").format(
selflabel=obj.label,
cdtype=cd.desc(),
cdname=cd.label)
@attr.s(cmp=False)
class _Formattable(ABC):
# Base class for anything that can be formatted and mounted,
@ -554,22 +509,6 @@ class _Formattable(ABC):
else:
return cd
def action_possible(self, action):
from subiquity.common.filesystem.actions import (
DeviceAction,
)
assert action in DeviceAction.supported(self)
if action == DeviceAction.EDIT:
r = action.can(self)
else:
r = getattr(self, "_can_" + action.name)
if isinstance(r, bool):
return r, None
elif isinstance(r, str):
return False, r
else:
return r
@property
@abstractmethod
def ok_for_raid(self):
@ -655,35 +594,6 @@ class _Device(_Formattable, ABC):
def _has_preexisting_partition(self):
return any(p.preserve for p in self._partitions)
@property
def _can_DELETE(self):
mounted_partitions = 0
for p in self._partitions:
if p.fs() and p.fs().mount():
mounted_partitions += 1
elif p.constructed_device():
cd = p.constructed_device()
return _(
"Cannot delete {selflabel} as partition {partnum} is part "
"of the {cdtype} {cdname}.").format(
selflabel=self.label,
partnum=p._number,
cdtype=cd.desc(),
cdname=cd.label,
)
if mounted_partitions > 1:
return _(
"Cannot delete {selflabel} because it has {count} mounted "
"partitions.").format(
selflabel=self.label,
count=mounted_partitions)
elif mounted_partitions == 1:
return _(
"Cannot delete {selflabel} because it has 1 mounted partition."
).format(selflabel=self.label)
else:
return _generic_can_DELETE(self)
@fsobj("dasd")
class Dasd:
@ -791,35 +701,6 @@ class Disk(_Device):
else:
return True
_can_INFO = True
@property
def _can_REFORMAT(self):
if len(self._partitions) == 0:
return False
for p in self._partitions:
if p._constructed_device is not None:
return False
return True
@property
def _can_PARTITION(self):
if self._has_preexisting_partition():
return False
if self.free_for_partitions <= 0:
return False
# We only create msdos partition tables with FBA dasds, which
# only support 3 partitions. As and when we support editing
# partition msdos tables we'll need to be more clever here.
if self.ptable in ['vtoc', 'msdos'] and len(self._partitions) >= 3:
return False
return True
_can_FORMAT = property(
lambda self: len(self._partitions) == 0 and
self._constructed_device is None)
_can_REMOVE = property(_generic_can_REMOVE)
def _is_boot_device(self):
bl = self._m.bootloader
if bl == Bootloader.NONE:
@ -829,18 +710,6 @@ class Disk(_Device):
elif bl in [Bootloader.PREP, Bootloader.UEFI]:
return any(p.grub_device for p in self._partitions)
@property
def _can_TOGGLE_BOOT(self):
if self._is_boot_device():
for disk in self._m.all_disks():
if disk is not self and disk._is_boot_device():
return True
return False
elif self._fs is not None or self._constructed_device is not None:
return False
else:
return self._can_be_boot_disk()
@property
def ok_for_raid(self):
if self._fs is not None:
@ -983,17 +852,6 @@ class Partition(_Formattable):
else:
return False
_can_REMOVE = property(_generic_can_REMOVE)
@property
def _can_DELETE(self):
if self.device._has_preexisting_partition():
return _("Cannot delete a single partition from a device that "
"already has partitions.")
if self.is_bootloader_partition:
return _("Cannot delete required bootloader partition")
return _generic_can_DELETE(self)
@property
def ok_for_raid(self):
if self.is_bootloader_partition:
@ -1054,13 +912,6 @@ class Raid(_Device):
def desc(self):
return _("software RAID {level}").format(level=self.raidlevel[4:])
_can_PARTITION = Disk._can_PARTITION
_can_REFORMAT = Disk._can_REFORMAT
_can_FORMAT = property(
lambda self: len(self._partitions) == 0 and
self._constructed_device is None)
_can_REMOVE = property(_generic_can_REMOVE)
@property
def ok_for_raid(self):
if self._fs is not None:
@ -1111,9 +962,6 @@ class LVM_VolGroup(_Device):
def desc(self):
return _("LVM volume group")
_can_CREATE_LV = property(
lambda self: not self.preserve and self.free_for_partitions > 0)
ok_for_raid = False
ok_for_lvm_vg = False
@ -1157,13 +1005,6 @@ class LVM_LogicalVolume(_Formattable):
label = short_label
@property
def _can_DELETE(self):
if self.volgroup._has_preexisting_partition():
return _("Cannot delete a single logical volume from a volume "
"group that already has logical volumes.")
return True
ok_for_raid = False
ok_for_lvm_vg = False

View File

@ -382,11 +382,11 @@ class TestFilesystemModel(unittest.TestCase):
def assertActionPossible(self, obj, action):
self.assertIn(action, DeviceAction.supported(obj))
self.assertTrue(obj.action_possible(action)[0])
self.assertTrue(action.can(obj)[0])
def assertActionNotPossible(self, obj, action):
self.assertIn(action, DeviceAction.supported(obj))
self.assertFalse(obj.action_possible(action)[0])
self.assertFalse(action.can(obj)[0])
def _test_remove_action(self, model, objects):
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)

View File

@ -343,7 +343,7 @@ class DeviceList(WidgetWrap):
label_meth = getattr(
self, '_label_{}'.format(action.name), lambda a, d: a.str())
label = label_meth(action, device)
enabled, whynot = device.action_possible(action)
enabled, whynot = action.can(device)
if whynot:
assert not enabled
enabled = True