Merge pull request #962 from mwhudson/fs-action-reorg

move DeviceAction stuff to its own file
This commit is contained in:
Michael Hudson-Doyle 2021-06-02 13:15:16 +12:00 committed by GitHub
commit 6aaa0b70d9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 902 additions and 729 deletions

View File

@ -10,6 +10,7 @@ subiquity/client/controllers/network.py
subiquity/client/controllers/progress.py
subiquity/client/controllers/proxy.py
subiquity/client/controllers/refresh.py
subiquity/client/controllers/serial.py
subiquity/client/controllers/snaplist.py
subiquity/client/controllers/ssh.py
subiquity/client/controllers/welcome.py
@ -31,11 +32,15 @@ subiquity/common/api/tests/test_client.py
subiquity/common/api/tests/test_endtoend.py
subiquity/common/api/tests/test_server.py
subiquity/common/errorreport.py
subiquity/common/filesystem.py
subiquity/common/filesystem/actions.py
subiquity/common/filesystem/__init__.py
subiquity/common/filesystem/manipulator.py
subiquity/common/filesystem/tests/__init__.py
subiquity/common/filesystem/tests/test_actions.py
subiquity/common/filesystem/tests/test_manipulator.py
subiquity/common/__init__.py
subiquity/common/serialize.py
subiquity/common/tests/__init__.py
subiquity/common/tests/test_filesystem.py
subiquity/common/tests/test_serialization.py
subiquity/common/types.py
subiquitycore/async_helpers.py
@ -168,6 +173,7 @@ subiquity/ui/views/keyboard.py
subiquity/ui/views/mirror.py
subiquity/ui/views/proxy.py
subiquity/ui/views/refresh.py
subiquity/ui/views/serial.py
subiquity/ui/views/snaplist.py
subiquity/ui/views/ssh.py
subiquity/ui/views/tests/__init__.py

View File

@ -19,7 +19,7 @@ import logging
from subiquitycore.lsb_release import lsb_release
from subiquity.client.controller import SubiquityTuiController
from subiquity.common.filesystem import FilesystemManipulator
from subiquity.common.filesystem.manipulator import FilesystemManipulator
from subiquity.common.types import ProbeStatus
from subiquity.models.filesystem import (
Bootloader,

View File

@ -0,0 +1,14 @@
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

View File

@ -0,0 +1,351 @@
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import enum
import functools
from subiquitycore.gettext38 import pgettext
from subiquity.models.filesystem import (
Bootloader,
Disk,
LVM_LogicalVolume,
LVM_VolGroup,
Partition,
Raid,
raidlevels_by_value,
)
_checkers = {}
def make_checker(action):
@functools.singledispatch
def impl(device):
raise NotImplementedError(
"checker for %s on %s not implemented" % (
action, device))
_checkers[action] = impl
return impl
class DeviceAction(enum.Enum):
# Information about a drive
INFO = pgettext("DeviceAction", "Info")
# Edit a device (partition, logical volume, RAID, etc)
EDIT = pgettext("DeviceAction", "Edit")
REFORMAT = pgettext("DeviceAction", "Reformat")
PARTITION = pgettext("DeviceAction", "Add Partition")
CREATE_LV = pgettext("DeviceAction", "Create Logical Volume")
FORMAT = pgettext("DeviceAction", "Format")
REMOVE = pgettext("DeviceAction", "Remove from RAID/LVM")
DELETE = pgettext("DeviceAction", "Delete")
TOGGLE_BOOT = pgettext("DeviceAction", "Make Boot Device")
def str(self):
return pgettext(type(self).__name__, self.value)
@classmethod
def supported(self, device):
return _supported_actions(device)
def can(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
def _supported_actions(device):
raise NotImplementedError(
"_supported_actions({}) not defined".format(device))
@_supported_actions.register(Disk)
def _disk_actions(disk):
actions = [
DeviceAction.INFO,
DeviceAction.REFORMAT,
DeviceAction.PARTITION,
DeviceAction.FORMAT,
DeviceAction.REMOVE,
]
if disk._m.bootloader != Bootloader.NONE:
actions.append(DeviceAction.TOGGLE_BOOT)
return actions
@_supported_actions.register(Partition)
def _part_actions(part):
return [
DeviceAction.EDIT,
DeviceAction.REMOVE,
DeviceAction.DELETE,
]
@_supported_actions.register(Raid)
def _raid_actions(raid):
return [
DeviceAction.EDIT,
DeviceAction.PARTITION,
DeviceAction.FORMAT,
DeviceAction.REMOVE,
DeviceAction.DELETE,
DeviceAction.REFORMAT,
]
@_supported_actions.register(LVM_VolGroup)
def _vg_actions(vg):
return [
DeviceAction.EDIT,
DeviceAction.CREATE_LV,
DeviceAction.DELETE,
]
@_supported_actions.register(LVM_LogicalVolume)
def _lv_actions(lv):
return [
DeviceAction.EDIT,
DeviceAction.DELETE,
]
_can_info = make_checker(DeviceAction.INFO)
@_can_info.register(Disk)
def _disk_info(disk):
return True
_can_edit = make_checker(DeviceAction.EDIT)
def _can_edit_generic(device):
cd = device.constructed_device()
if cd is None:
return True
return _(
"Cannot edit {selflabel} as it is part of the {cdtype} "
"{cdname}.").format(
selflabel=device.label,
cdtype=cd.desc(),
cdname=cd.label)
_can_edit.register(Partition, _can_edit_generic)
_can_edit.register(LVM_LogicalVolume, _can_edit_generic)
@_can_edit.register(Raid)
def _can_edit_raid(raid):
if raid.preserve:
return _("Cannot edit pre-existing RAIDs.")
elif len(raid._partitions) > 0:
return _(
"Cannot edit {raidlabel} because it has partitions.").format(
raidlabel=raid.label)
else:
return _can_edit_generic(raid)
@_can_edit.register(LVM_VolGroup)
def _can_edit_vg(vg):
if vg.preserve:
return _("Cannot edit pre-existing volume groups.")
elif len(vg._partitions) > 0:
return _(
"Cannot edit {vglabel} because it has logical "
"volumes.").format(
vglabel=vg.label)
else:
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

@ -15,14 +15,16 @@
import logging
from subiquity.common.filesystem.actions import (
DeviceAction,
)
from subiquity.common.types import Bootloader
from subiquity.models.filesystem import (
align_up,
DeviceAction,
Partition,
)
log = logging.getLogger('subiquity.common.filesystem')
log = logging.getLogger('subiquity.common.filesystem.manipulator')
BIOS_GRUB_SIZE_BYTES = 1 * 1024 * 1024 # 1MiB
@ -214,7 +216,7 @@ class FilesystemManipulator:
needs_boot = self.model.needs_bootloader_partition()
log.debug('model needs a bootloader partition? {}'.format(needs_boot))
can_be_boot = DeviceAction.TOGGLE_BOOT in disk.supported_actions
can_be_boot = DeviceAction.TOGGLE_BOOT in DeviceAction.supported(disk)
if needs_boot and len(disk.partitions()) == 0 and can_be_boot:
part = self._create_boot_partition(disk)

View File

@ -0,0 +1,14 @@
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

View File

@ -0,0 +1,493 @@
# Copyright 2021 Canonical, Ltd.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import unittest
from subiquity.common.filesystem.actions import (
DeviceAction,
)
from subiquity.models.filesystem import Bootloader
from subiquity.models.tests.test_filesystem import (
make_disk,
make_lv,
make_model,
make_model_and_disk,
make_model_and_lv,
make_model_and_partition,
make_model_and_raid,
make_model_and_vg,
make_partition,
make_raid,
make_vg,
)
class TestActions(unittest.TestCase):
def assertActionNotSupported(self, obj, action):
self.assertNotIn(action, DeviceAction.supported(obj))
def assertActionPossible(self, obj, action):
self.assertIn(action, DeviceAction.supported(obj))
self.assertTrue(action.can(obj)[0])
def assertActionNotPossible(self, obj, action):
self.assertIn(action, DeviceAction.supported(obj))
self.assertFalse(action.can(obj)[0])
def _test_remove_action(self, model, objects):
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)
vg = model.add_volgroup('vg1', {objects[0], objects[1]})
self.assertActionPossible(objects[0], DeviceAction.REMOVE)
# Cannot remove a device from a preexisting VG
vg.preserve = True
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)
vg.preserve = False
# Probably this removal should be a model method?
vg.devices.remove(objects[1])
objects[1]._constructed_device = None
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)
raid = model.add_raid('md0', 'raid1', set(objects[2:]), set())
self.assertActionPossible(objects[2], DeviceAction.REMOVE)
# Cannot remove a device from a preexisting RAID
raid.preserve = True
self.assertActionNotPossible(objects[2], DeviceAction.REMOVE)
raid.preserve = False
# Probably this removal should be a model method?
raid.devices.remove(objects[4])
objects[4]._constructed_device = None
self.assertActionNotPossible(objects[2], DeviceAction.REMOVE)
def test_disk_action_INFO(self):
model, disk = make_model_and_disk()
self.assertActionPossible(disk, DeviceAction.INFO)
def test_disk_action_EDIT(self):
model, disk = make_model_and_disk()
self.assertActionNotSupported(disk, DeviceAction.EDIT)
def test_disk_action_REFORMAT(self):
model = make_model()
disk1 = make_disk(model, preserve=False)
self.assertActionNotPossible(disk1, DeviceAction.REFORMAT)
disk1p1 = make_partition(model, disk1, preserve=False)
self.assertActionPossible(disk1, DeviceAction.REFORMAT)
model.add_volgroup('vg0', {disk1p1})
self.assertActionNotPossible(disk1, DeviceAction.REFORMAT)
disk2 = make_disk(model, preserve=True)
self.assertActionNotPossible(disk2, DeviceAction.REFORMAT)
disk2p1 = make_partition(model, disk2, preserve=True)
self.assertActionPossible(disk2, DeviceAction.REFORMAT)
model.add_volgroup('vg1', {disk2p1})
self.assertActionNotPossible(disk2, DeviceAction.REFORMAT)
disk3 = make_disk(model, preserve=False)
model.add_volgroup('vg2', {disk3})
self.assertActionNotPossible(disk3, DeviceAction.REFORMAT)
disk4 = make_disk(model, preserve=True)
model.add_volgroup('vg2', {disk4})
self.assertActionNotPossible(disk4, DeviceAction.REFORMAT)
def test_disk_action_PARTITION(self):
model, disk = make_model_and_disk()
self.assertActionPossible(disk, DeviceAction.PARTITION)
make_partition(model, disk, size=disk.free_for_partitions//2)
self.assertActionPossible(disk, DeviceAction.PARTITION)
make_partition(model, disk, size=disk.free_for_partitions)
self.assertActionNotPossible(disk, DeviceAction.PARTITION)
# Can partition a disk with .preserve=True
disk2 = make_disk(model)
disk2.preserve = True
self.assertActionPossible(disk2, DeviceAction.PARTITION)
# But not if it has a partition.
make_partition(model, disk2, preserve=True)
self.assertActionNotPossible(disk2, DeviceAction.PARTITION)
def test_disk_action_CREATE_LV(self):
model, disk = make_model_and_disk()
self.assertActionNotSupported(disk, DeviceAction.CREATE_LV)
def test_disk_action_FORMAT(self):
model, disk = make_model_and_disk()
self.assertActionPossible(disk, DeviceAction.FORMAT)
make_partition(model, disk)
self.assertActionNotPossible(disk, DeviceAction.FORMAT)
disk2 = make_disk(model)
model.add_volgroup('vg1', {disk2})
self.assertActionNotPossible(disk2, DeviceAction.FORMAT)
def test_disk_action_REMOVE(self):
model = make_model()
disks = [make_disk(model) for i in range(5)]
self._test_remove_action(model, disks)
def test_disk_action_DELETE(self):
model, disk = make_model_and_disk()
self.assertActionNotSupported(disk, DeviceAction.DELETE)
def test_disk_action_TOGGLE_BOOT_NONE(self):
model, disk = make_model_and_disk(Bootloader.NONE)
self.assertActionNotSupported(disk, DeviceAction.TOGGLE_BOOT)
def test_disk_action_TOGGLE_BOOT_BIOS(self):
model = make_model(Bootloader.BIOS)
# Disks with msdos partition tables can always be the BIOS boot disk.
dos_disk = make_disk(model, ptable='msdos', preserve=True)
self.assertActionPossible(dos_disk, DeviceAction.TOGGLE_BOOT)
# Even if they have existing partitions
make_partition(
model, dos_disk, size=dos_disk.free_for_partitions, preserve=True)
self.assertActionPossible(dos_disk, DeviceAction.TOGGLE_BOOT)
# (we never create dos partition tables so no need to test
# preserve=False case).
# GPT disks with new partition tables can always be the BIOS boot disk
gpt_disk = make_disk(model, ptable='gpt', preserve=False)
self.assertActionPossible(gpt_disk, DeviceAction.TOGGLE_BOOT)
# Even if they are filled with partitions (we resize partitions to fit)
make_partition(model, gpt_disk, size=dos_disk.free_for_partitions)
self.assertActionPossible(gpt_disk, DeviceAction.TOGGLE_BOOT)
# GPT disks with existing partition tables but no partitions can be the
# BIOS boot disk (in general we ignore existing empty partition tables)
gpt_disk2 = make_disk(model, ptable='gpt', preserve=True)
self.assertActionPossible(gpt_disk2, DeviceAction.TOGGLE_BOOT)
# If there is an existing *partition* though, it cannot be the boot
# disk
make_partition(model, gpt_disk2, preserve=True)
self.assertActionNotPossible(gpt_disk2, DeviceAction.TOGGLE_BOOT)
# Unless there is already a bios_grub partition we can reuse
gpt_disk3 = make_disk(model, ptable='gpt', preserve=True)
make_partition(
model, gpt_disk3, flag="bios_grub", preserve=True)
make_partition(
model, gpt_disk3, preserve=True)
self.assertActionPossible(gpt_disk3, DeviceAction.TOGGLE_BOOT)
# Edge case city: the bios_grub partition has to be first
gpt_disk4 = make_disk(model, ptable='gpt', preserve=True)
make_partition(
model, gpt_disk4, preserve=True)
make_partition(
model, gpt_disk4, flag="bios_grub", preserve=True)
self.assertActionNotPossible(gpt_disk4, DeviceAction.TOGGLE_BOOT)
def _test_TOGGLE_BOOT_boot_partition(self, bl, flag):
# The logic for when TOGGLE_BOOT is enabled for both UEFI and PREP
# bootloaders turns out to be the same, modulo the special flag that
# has to be present on a partition.
model = make_model(bl)
# A disk with a new partition table can always be the UEFI/PREP boot
# disk.
new_disk = make_disk(model, preserve=False)
self.assertActionPossible(new_disk, DeviceAction.TOGGLE_BOOT)
# Even if they are filled with partitions (we resize partitions to fit)
make_partition(model, new_disk, size=new_disk.free_for_partitions)
self.assertActionPossible(new_disk, DeviceAction.TOGGLE_BOOT)
# A disk with an existing but empty partitions can also be the
# UEFI/PREP boot disk.
old_disk = make_disk(model, preserve=True, ptable='gpt')
self.assertActionPossible(old_disk, DeviceAction.TOGGLE_BOOT)
# If there is an existing partition though, it cannot.
make_partition(model, old_disk, preserve=True)
self.assertActionNotPossible(old_disk, DeviceAction.TOGGLE_BOOT)
# If there is an existing ESP/PReP partition though, fine!
make_partition(model, old_disk, flag=flag, preserve=True)
self.assertActionPossible(old_disk, DeviceAction.TOGGLE_BOOT)
def test_disk_action_TOGGLE_BOOT_UEFI(self):
self._test_TOGGLE_BOOT_boot_partition(Bootloader.UEFI, "boot")
def test_disk_action_TOGGLE_BOOT_PREP(self):
self._test_TOGGLE_BOOT_boot_partition(Bootloader.PREP, "prep")
def test_partition_action_INFO(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.INFO)
def test_partition_action_EDIT(self):
model, part = make_model_and_partition()
self.assertActionPossible(part, DeviceAction.EDIT)
model.add_volgroup('vg1', {part})
self.assertActionNotPossible(part, DeviceAction.EDIT)
def test_partition_action_REFORMAT(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.REFORMAT)
def test_partition_action_PARTITION(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.PARTITION)
def test_partition_action_CREATE_LV(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.CREATE_LV)
def test_partition_action_FORMAT(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.FORMAT)
def test_partition_action_REMOVE(self):
model = make_model()
parts = []
for i in range(5):
parts.append(make_partition(model))
self._test_remove_action(model, parts)
def test_partition_action_DELETE(self):
model = make_model()
part1 = make_partition(model)
self.assertActionPossible(part1, DeviceAction.DELETE)
fs = model.add_filesystem(part1, 'ext4')
self.assertActionPossible(part1, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionPossible(part1, DeviceAction.DELETE)
part2 = make_partition(model)
model.add_volgroup('vg1', {part2})
self.assertActionNotPossible(part2, DeviceAction.DELETE)
part = make_partition(make_model(Bootloader.BIOS), flag='bios_grub')
self.assertActionNotPossible(part, DeviceAction.DELETE)
part = make_partition(make_model(Bootloader.UEFI), flag='boot')
self.assertActionNotPossible(part, DeviceAction.DELETE)
part = make_partition(make_model(Bootloader.PREP), flag='prep')
self.assertActionNotPossible(part, DeviceAction.DELETE)
# You cannot delete a partition from a disk that has
# pre-existing partitions (only reformat)
disk2 = make_disk(model, preserve=True)
disk2p1 = make_partition(model, disk2, preserve=True)
self.assertActionNotPossible(disk2p1, DeviceAction.DELETE)
def test_partition_action_TOGGLE_BOOT(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.TOGGLE_BOOT)
def test_raid_action_INFO(self):
model, raid = make_model_and_raid()
self.assertActionNotSupported(raid, DeviceAction.INFO)
def test_raid_action_EDIT(self):
model = make_model()
raid1 = make_raid(model)
self.assertActionPossible(raid1, DeviceAction.EDIT)
model.add_volgroup('vg1', {raid1})
self.assertActionNotPossible(raid1, DeviceAction.EDIT)
raid2 = make_raid(model)
make_partition(model, raid2)
self.assertActionNotPossible(raid2, DeviceAction.EDIT)
raid3 = make_raid(model)
raid3.preserve = True
self.assertActionNotPossible(raid3, DeviceAction.EDIT)
def test_raid_action_REFORMAT(self):
model = make_model()
raid1 = make_raid(model)
self.assertActionNotPossible(raid1, DeviceAction.REFORMAT)
raid1p1 = make_partition(model, raid1)
self.assertActionPossible(raid1, DeviceAction.REFORMAT)
model.add_volgroup('vg0', {raid1p1})
self.assertActionNotPossible(raid1, DeviceAction.REFORMAT)
raid2 = make_raid(model)
raid2.preserve = True
self.assertActionNotPossible(raid2, DeviceAction.REFORMAT)
raid2p1 = make_partition(model, raid2, preserve=True)
self.assertActionPossible(raid2, DeviceAction.REFORMAT)
model.add_volgroup('vg1', {raid2p1})
self.assertActionNotPossible(raid2, DeviceAction.REFORMAT)
raid3 = make_raid(model)
model.add_volgroup('vg2', {raid3})
self.assertActionNotPossible(raid3, DeviceAction.REFORMAT)
raid4 = make_raid(model)
raid4.preserve = True
model.add_volgroup('vg2', {raid4})
self.assertActionNotPossible(raid4, DeviceAction.REFORMAT)
def test_raid_action_PARTITION(self):
model, raid = make_model_and_raid()
self.assertActionPossible(raid, DeviceAction.PARTITION)
make_partition(model, raid, size=raid.free_for_partitions//2)
self.assertActionPossible(raid, DeviceAction.PARTITION)
make_partition(model, raid, size=raid.free_for_partitions)
self.assertActionNotPossible(raid, DeviceAction.PARTITION)
# Can partition a raid with .preserve=True
raid2 = make_raid(model)
raid2.preserve = True
self.assertActionPossible(raid2, DeviceAction.PARTITION)
# But not if it has a partition.
make_partition(model, raid2, preserve=True)
self.assertActionNotPossible(raid2, DeviceAction.PARTITION)
def test_raid_action_CREATE_LV(self):
model, raid = make_model_and_raid()
self.assertActionNotSupported(raid, DeviceAction.CREATE_LV)
def test_raid_action_FORMAT(self):
model, raid = make_model_and_raid()
self.assertActionPossible(raid, DeviceAction.FORMAT)
make_partition(model, raid)
self.assertActionNotPossible(raid, DeviceAction.FORMAT)
raid2 = make_raid(model)
model.add_volgroup('vg1', {raid2})
self.assertActionNotPossible(raid2, DeviceAction.FORMAT)
def test_raid_action_REMOVE(self):
model = make_model()
raids = [make_raid(model) for i in range(5)]
self._test_remove_action(model, raids)
def test_raid_action_DELETE(self):
model, raid = make_model_and_raid()
raid1 = make_raid(model)
self.assertActionPossible(raid1, DeviceAction.DELETE)
part = make_partition(model, raid1)
self.assertActionPossible(raid1, DeviceAction.DELETE)
fs = model.add_filesystem(part, 'ext4')
self.assertActionPossible(raid1, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionNotPossible(raid1, DeviceAction.DELETE)
raid2 = make_raid(model)
self.assertActionPossible(raid2, DeviceAction.DELETE)
fs = model.add_filesystem(raid2, 'ext4')
self.assertActionPossible(raid2, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionPossible(raid2, DeviceAction.DELETE)
raid2 = make_raid(model)
model.add_volgroup('vg0', {raid2})
self.assertActionNotPossible(raid2, DeviceAction.DELETE)
def test_raid_action_TOGGLE_BOOT(self):
model, raid = make_model_and_raid()
self.assertActionNotSupported(raid, DeviceAction.TOGGLE_BOOT)
def test_vg_action_INFO(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.INFO)
def test_vg_action_EDIT(self):
model, vg = make_model_and_vg()
self.assertActionPossible(vg, DeviceAction.EDIT)
model.add_logical_volume(vg, 'lv1', size=vg.free_for_partitions//2)
self.assertActionNotPossible(vg, DeviceAction.EDIT)
vg2 = make_vg(model)
vg2.preserve = True
self.assertActionNotPossible(vg2, DeviceAction.EDIT)
def test_vg_action_REFORMAT(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.REFORMAT)
def test_vg_action_PARTITION(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.PARTITION)
def test_vg_action_CREATE_LV(self):
model, vg = make_model_and_vg()
self.assertActionPossible(vg, DeviceAction.CREATE_LV)
model.add_logical_volume(vg, 'lv1', size=vg.free_for_partitions//2)
self.assertActionPossible(vg, DeviceAction.CREATE_LV)
model.add_logical_volume(vg, 'lv2', size=vg.free_for_partitions)
self.assertActionNotPossible(vg, DeviceAction.CREATE_LV)
vg2 = make_vg(model)
vg2.preserve = True
self.assertActionNotPossible(vg2, DeviceAction.CREATE_LV)
def test_vg_action_FORMAT(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.FORMAT)
def test_vg_action_REMOVE(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.REMOVE)
def test_vg_action_DELETE(self):
model, vg = make_model_and_vg()
self.assertActionPossible(vg, DeviceAction.DELETE)
self.assertActionPossible(vg, DeviceAction.DELETE)
lv = model.add_logical_volume(
vg, 'lv0', size=vg.free_for_partitions//2)
self.assertActionPossible(vg, DeviceAction.DELETE)
fs = model.add_filesystem(lv, 'ext4')
self.assertActionPossible(vg, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionNotPossible(vg, DeviceAction.DELETE)
def test_vg_action_TOGGLE_BOOT(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.TOGGLE_BOOT)
def test_lv_action_INFO(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.INFO)
def test_lv_action_EDIT(self):
model, lv = make_model_and_lv()
self.assertActionPossible(lv, DeviceAction.EDIT)
def test_lv_action_REFORMAT(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.REFORMAT)
def test_lv_action_PARTITION(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.PARTITION)
def test_lv_action_CREATE_LV(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.CREATE_LV)
def test_lv_action_FORMAT(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.FORMAT)
def test_lv_action_REMOVE(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.REMOVE)
def test_lv_action_DELETE(self):
model, lv = make_model_and_lv()
self.assertActionPossible(lv, DeviceAction.DELETE)
fs = model.add_filesystem(lv, 'ext4')
self.assertActionPossible(lv, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionPossible(lv, DeviceAction.DELETE)
lv2 = make_lv(model)
lv2.preserve = lv2.volgroup.preserve = True
self.assertActionNotPossible(lv2, DeviceAction.DELETE)
def test_lv_action_TOGGLE_BOOT(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.TOGGLE_BOOT)

View File

@ -15,7 +15,10 @@
import unittest
from subiquity.common.filesystem import (
from subiquity.common.filesystem.actions import (
DeviceAction,
)
from subiquity.common.filesystem.manipulator import (
FilesystemManipulator,
)
from subiquity.models.tests.test_filesystem import (
@ -24,7 +27,6 @@ from subiquity.models.tests.test_filesystem import (
)
from subiquity.models.filesystem import (
Bootloader,
DeviceAction,
)
@ -60,11 +62,11 @@ class TestFilesystemManipulator(unittest.TestCase):
# manipulator around.
for bl in Bootloader:
manipulator, disk = make_manipulator_and_disk(bl)
if DeviceAction.TOGGLE_BOOT not in disk.supported_actions:
if DeviceAction.TOGGLE_BOOT not in DeviceAction.supported(disk):
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

@ -16,7 +16,6 @@
from abc import ABC, abstractmethod
import attr
import collections
import enum
import fnmatch
import itertools
import logging
@ -32,8 +31,6 @@ from curtin.util import human2bytes
from probert.storage import StorageInfo
from subiquitycore.gettext38 import pgettext
from subiquity.common.types import Bootloader
@ -422,80 +419,6 @@ def asdict(inst):
# in the FilesystemModel or FilesystemController classes.
class DeviceAction(enum.Enum):
# Information about a drive
INFO = pgettext("DeviceAction", "Info")
# Edit a device (partition, logical volume, RAID, etc)
EDIT = pgettext("DeviceAction", "Edit")
REFORMAT = pgettext("DeviceAction", "Reformat")
PARTITION = pgettext("DeviceAction", "Add Partition")
CREATE_LV = pgettext("DeviceAction", "Create Logical Volume")
FORMAT = pgettext("DeviceAction", "Format")
REMOVE = pgettext("DeviceAction", "Remove from RAID/LVM")
DELETE = pgettext("DeviceAction", "Delete")
TOGGLE_BOOT = pgettext("DeviceAction", "Make Boot Device")
def str(self):
return pgettext(type(self).__name__, self.value)
def _generic_can_EDIT(obj):
cd = obj.constructed_device()
if cd is None:
return True
return _(
"Cannot edit {selflabel} as it is part of the {cdtype} "
"{cdname}.").format(
selflabel=obj.label,
cdtype=cd.desc(),
cdname=cd.label)
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,
@ -586,21 +509,6 @@ class _Formattable(ABC):
else:
return cd
@property
@abstractmethod
def supported_actions(self):
pass
def action_possible(self, action):
assert action in self.supported_actions
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):
@ -686,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:
@ -822,48 +701,6 @@ class Disk(_Device):
else:
return True
@property
def supported_actions(self):
actions = [
DeviceAction.INFO,
DeviceAction.REFORMAT,
DeviceAction.PARTITION,
DeviceAction.FORMAT,
DeviceAction.REMOVE,
]
if self._m.bootloader != Bootloader.NONE:
actions.append(DeviceAction.TOGGLE_BOOT)
return actions
_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:
@ -873,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:
@ -1027,25 +852,6 @@ class Partition(_Formattable):
else:
return False
supported_actions = [
DeviceAction.EDIT,
DeviceAction.REMOVE,
DeviceAction.DELETE,
]
_can_EDIT = property(_generic_can_EDIT)
_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:
@ -1106,33 +912,6 @@ class Raid(_Device):
def desc(self):
return _("software RAID {level}").format(level=self.raidlevel[4:])
supported_actions = [
DeviceAction.EDIT,
DeviceAction.PARTITION,
DeviceAction.FORMAT,
DeviceAction.REMOVE,
DeviceAction.DELETE,
DeviceAction.REFORMAT,
]
@property
def _can_EDIT(self):
if self.preserve:
return _("Cannot edit pre-existing RAIDs.")
elif len(self._partitions) > 0:
return _(
"Cannot edit {selflabel} because it has partitions.").format(
selflabel=self.label)
else:
return _generic_can_EDIT(self)
_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:
@ -1183,27 +962,6 @@ class LVM_VolGroup(_Device):
def desc(self):
return _("LVM volume group")
supported_actions = [
DeviceAction.EDIT,
DeviceAction.CREATE_LV,
DeviceAction.DELETE,
]
@property
def _can_EDIT(self):
if self.preserve:
return _("Cannot edit pre-existing volume groups.")
elif len(self._partitions) > 0:
return _(
"Cannot edit {selflabel} because it has logical "
"volumes.").format(
selflabel=self.label)
else:
return _generic_can_EDIT(self)
_can_CREATE_LV = property(
lambda self: not self.preserve and self.free_for_partitions > 0)
ok_for_raid = False
ok_for_lvm_vg = False
@ -1247,20 +1005,6 @@ class LVM_LogicalVolume(_Formattable):
label = short_label
supported_actions = [
DeviceAction.EDIT,
DeviceAction.DELETE,
]
_can_EDIT = True
@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

@ -20,7 +20,6 @@ import attr
from subiquity.models.filesystem import (
Bootloader,
dehumanize_size,
DeviceAction,
Disk,
FilesystemModel,
get_raid_size,
@ -375,462 +374,6 @@ class TestFilesystemModel(unittest.TestCase):
partition.usage_labels(),
["to be reformatted as ext4", "mounted at /"])
def assertActionNotSupported(self, obj, action):
self.assertNotIn(action, obj.supported_actions)
def assertActionPossible(self, obj, action):
self.assertIn(action, obj.supported_actions)
self.assertTrue(obj.action_possible(action)[0])
def assertActionNotPossible(self, obj, action):
self.assertIn(action, obj.supported_actions)
self.assertFalse(obj.action_possible(action)[0])
def _test_remove_action(self, model, objects):
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)
vg = model.add_volgroup('vg1', {objects[0], objects[1]})
self.assertActionPossible(objects[0], DeviceAction.REMOVE)
# Cannot remove a device from a preexisting VG
vg.preserve = True
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)
vg.preserve = False
# Probably this removal should be a model method?
vg.devices.remove(objects[1])
objects[1]._constructed_device = None
self.assertActionNotPossible(objects[0], DeviceAction.REMOVE)
raid = model.add_raid('md0', 'raid1', set(objects[2:]), set())
self.assertActionPossible(objects[2], DeviceAction.REMOVE)
# Cannot remove a device from a preexisting RAID
raid.preserve = True
self.assertActionNotPossible(objects[2], DeviceAction.REMOVE)
raid.preserve = False
# Probably this removal should be a model method?
raid.devices.remove(objects[4])
objects[4]._constructed_device = None
self.assertActionNotPossible(objects[2], DeviceAction.REMOVE)
def test_disk_action_INFO(self):
model, disk = make_model_and_disk()
self.assertActionPossible(disk, DeviceAction.INFO)
def test_disk_action_EDIT(self):
model, disk = make_model_and_disk()
self.assertActionNotSupported(disk, DeviceAction.EDIT)
def test_disk_action_REFORMAT(self):
model = make_model()
disk1 = make_disk(model, preserve=False)
self.assertActionNotPossible(disk1, DeviceAction.REFORMAT)
disk1p1 = make_partition(model, disk1, preserve=False)
self.assertActionPossible(disk1, DeviceAction.REFORMAT)
model.add_volgroup('vg0', {disk1p1})
self.assertActionNotPossible(disk1, DeviceAction.REFORMAT)
disk2 = make_disk(model, preserve=True)
self.assertActionNotPossible(disk2, DeviceAction.REFORMAT)
disk2p1 = make_partition(model, disk2, preserve=True)
self.assertActionPossible(disk2, DeviceAction.REFORMAT)
model.add_volgroup('vg1', {disk2p1})
self.assertActionNotPossible(disk2, DeviceAction.REFORMAT)
disk3 = make_disk(model, preserve=False)
model.add_volgroup('vg2', {disk3})
self.assertActionNotPossible(disk3, DeviceAction.REFORMAT)
disk4 = make_disk(model, preserve=True)
model.add_volgroup('vg2', {disk4})
self.assertActionNotPossible(disk4, DeviceAction.REFORMAT)
def test_disk_action_PARTITION(self):
model, disk = make_model_and_disk()
self.assertActionPossible(disk, DeviceAction.PARTITION)
make_partition(model, disk, size=disk.free_for_partitions//2)
self.assertActionPossible(disk, DeviceAction.PARTITION)
make_partition(model, disk, size=disk.free_for_partitions)
self.assertActionNotPossible(disk, DeviceAction.PARTITION)
# Can partition a disk with .preserve=True
disk2 = make_disk(model)
disk2.preserve = True
self.assertActionPossible(disk2, DeviceAction.PARTITION)
# But not if it has a partition.
make_partition(model, disk2, preserve=True)
self.assertActionNotPossible(disk2, DeviceAction.PARTITION)
def test_disk_action_CREATE_LV(self):
model, disk = make_model_and_disk()
self.assertActionNotSupported(disk, DeviceAction.CREATE_LV)
def test_disk_action_FORMAT(self):
model, disk = make_model_and_disk()
self.assertActionPossible(disk, DeviceAction.FORMAT)
make_partition(model, disk)
self.assertActionNotPossible(disk, DeviceAction.FORMAT)
disk2 = make_disk(model)
model.add_volgroup('vg1', {disk2})
self.assertActionNotPossible(disk2, DeviceAction.FORMAT)
def test_disk_action_REMOVE(self):
model = make_model()
disks = [make_disk(model) for i in range(5)]
self._test_remove_action(model, disks)
def test_disk_action_DELETE(self):
model, disk = make_model_and_disk()
self.assertActionNotSupported(disk, DeviceAction.DELETE)
def test_disk_action_TOGGLE_BOOT_NONE(self):
model, disk = make_model_and_disk(Bootloader.NONE)
self.assertActionNotSupported(disk, DeviceAction.TOGGLE_BOOT)
def test_disk_action_TOGGLE_BOOT_BIOS(self):
model = make_model(Bootloader.BIOS)
# Disks with msdos partition tables can always be the BIOS boot disk.
dos_disk = make_disk(model, ptable='msdos', preserve=True)
self.assertActionPossible(dos_disk, DeviceAction.TOGGLE_BOOT)
# Even if they have existing partitions
make_partition(
model, dos_disk, size=dos_disk.free_for_partitions, preserve=True)
self.assertActionPossible(dos_disk, DeviceAction.TOGGLE_BOOT)
# (we never create dos partition tables so no need to test
# preserve=False case).
# GPT disks with new partition tables can always be the BIOS boot disk
gpt_disk = make_disk(model, ptable='gpt', preserve=False)
self.assertActionPossible(gpt_disk, DeviceAction.TOGGLE_BOOT)
# Even if they are filled with partitions (we resize partitions to fit)
make_partition(model, gpt_disk, size=dos_disk.free_for_partitions)
self.assertActionPossible(gpt_disk, DeviceAction.TOGGLE_BOOT)
# GPT disks with existing partition tables but no partitions can be the
# BIOS boot disk (in general we ignore existing empty partition tables)
gpt_disk2 = make_disk(model, ptable='gpt', preserve=True)
self.assertActionPossible(gpt_disk2, DeviceAction.TOGGLE_BOOT)
# If there is an existing *partition* though, it cannot be the boot
# disk
make_partition(model, gpt_disk2, preserve=True)
self.assertActionNotPossible(gpt_disk2, DeviceAction.TOGGLE_BOOT)
# Unless there is already a bios_grub partition we can reuse
gpt_disk3 = make_disk(model, ptable='gpt', preserve=True)
make_partition(
model, gpt_disk3, flag="bios_grub", preserve=True)
make_partition(
model, gpt_disk3, preserve=True)
self.assertActionPossible(gpt_disk3, DeviceAction.TOGGLE_BOOT)
# Edge case city: the bios_grub partition has to be first
gpt_disk4 = make_disk(model, ptable='gpt', preserve=True)
make_partition(
model, gpt_disk4, preserve=True)
make_partition(
model, gpt_disk4, flag="bios_grub", preserve=True)
self.assertActionNotPossible(gpt_disk4, DeviceAction.TOGGLE_BOOT)
def _test_TOGGLE_BOOT_boot_partition(self, bl, flag):
# The logic for when TOGGLE_BOOT is enabled for both UEFI and PREP
# bootloaders turns out to be the same, modulo the special flag that
# has to be present on a partition.
model = make_model(bl)
# A disk with a new partition table can always be the UEFI/PREP boot
# disk.
new_disk = make_disk(model, preserve=False)
self.assertActionPossible(new_disk, DeviceAction.TOGGLE_BOOT)
# Even if they are filled with partitions (we resize partitions to fit)
make_partition(model, new_disk, size=new_disk.free_for_partitions)
self.assertActionPossible(new_disk, DeviceAction.TOGGLE_BOOT)
# A disk with an existing but empty partitions can also be the
# UEFI/PREP boot disk.
old_disk = make_disk(model, preserve=True, ptable='gpt')
self.assertActionPossible(old_disk, DeviceAction.TOGGLE_BOOT)
# If there is an existing partition though, it cannot.
make_partition(model, old_disk, preserve=True)
self.assertActionNotPossible(old_disk, DeviceAction.TOGGLE_BOOT)
# If there is an existing ESP/PReP partition though, fine!
make_partition(model, old_disk, flag=flag, preserve=True)
self.assertActionPossible(old_disk, DeviceAction.TOGGLE_BOOT)
def test_disk_action_TOGGLE_BOOT_UEFI(self):
self._test_TOGGLE_BOOT_boot_partition(Bootloader.UEFI, "boot")
def test_disk_action_TOGGLE_BOOT_PREP(self):
self._test_TOGGLE_BOOT_boot_partition(Bootloader.PREP, "prep")
def test_partition_action_INFO(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.INFO)
def test_partition_action_EDIT(self):
model, part = make_model_and_partition()
self.assertActionPossible(part, DeviceAction.EDIT)
model.add_volgroup('vg1', {part})
self.assertActionNotPossible(part, DeviceAction.EDIT)
def test_partition_action_REFORMAT(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.REFORMAT)
def test_partition_action_PARTITION(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.PARTITION)
def test_partition_action_CREATE_LV(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.CREATE_LV)
def test_partition_action_FORMAT(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.FORMAT)
def test_partition_action_REMOVE(self):
model = make_model()
parts = []
for i in range(5):
parts.append(make_partition(model))
self._test_remove_action(model, parts)
def test_partition_action_DELETE(self):
model = make_model()
part1 = make_partition(model)
self.assertActionPossible(part1, DeviceAction.DELETE)
fs = model.add_filesystem(part1, 'ext4')
self.assertActionPossible(part1, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionPossible(part1, DeviceAction.DELETE)
part2 = make_partition(model)
model.add_volgroup('vg1', {part2})
self.assertActionNotPossible(part2, DeviceAction.DELETE)
part = make_partition(make_model(Bootloader.BIOS), flag='bios_grub')
self.assertActionNotPossible(part, DeviceAction.DELETE)
part = make_partition(make_model(Bootloader.UEFI), flag='boot')
self.assertActionNotPossible(part, DeviceAction.DELETE)
part = make_partition(make_model(Bootloader.PREP), flag='prep')
self.assertActionNotPossible(part, DeviceAction.DELETE)
# You cannot delete a partition from a disk that has
# pre-existing partitions (only reformat)
disk2 = make_disk(model, preserve=True)
disk2p1 = make_partition(model, disk2, preserve=True)
self.assertActionNotPossible(disk2p1, DeviceAction.DELETE)
def test_partition_action_TOGGLE_BOOT(self):
model, part = make_model_and_partition()
self.assertActionNotSupported(part, DeviceAction.TOGGLE_BOOT)
def test_raid_action_INFO(self):
model, raid = make_model_and_raid()
self.assertActionNotSupported(raid, DeviceAction.INFO)
def test_raid_action_EDIT(self):
model = make_model()
raid1 = make_raid(model)
self.assertActionPossible(raid1, DeviceAction.EDIT)
model.add_volgroup('vg1', {raid1})
self.assertActionNotPossible(raid1, DeviceAction.EDIT)
raid2 = make_raid(model)
make_partition(model, raid2)
self.assertActionNotPossible(raid2, DeviceAction.EDIT)
raid3 = make_raid(model)
raid3.preserve = True
self.assertActionNotPossible(raid3, DeviceAction.EDIT)
def test_raid_action_REFORMAT(self):
model = make_model()
raid1 = make_raid(model)
self.assertActionNotPossible(raid1, DeviceAction.REFORMAT)
raid1p1 = make_partition(model, raid1)
self.assertActionPossible(raid1, DeviceAction.REFORMAT)
model.add_volgroup('vg0', {raid1p1})
self.assertActionNotPossible(raid1, DeviceAction.REFORMAT)
raid2 = make_raid(model)
raid2.preserve = True
self.assertActionNotPossible(raid2, DeviceAction.REFORMAT)
raid2p1 = make_partition(model, raid2, preserve=True)
self.assertActionPossible(raid2, DeviceAction.REFORMAT)
model.add_volgroup('vg1', {raid2p1})
self.assertActionNotPossible(raid2, DeviceAction.REFORMAT)
raid3 = make_raid(model)
model.add_volgroup('vg2', {raid3})
self.assertActionNotPossible(raid3, DeviceAction.REFORMAT)
raid4 = make_raid(model)
raid4.preserve = True
model.add_volgroup('vg2', {raid4})
self.assertActionNotPossible(raid4, DeviceAction.REFORMAT)
def test_raid_action_PARTITION(self):
model, raid = make_model_and_raid()
self.assertActionPossible(raid, DeviceAction.PARTITION)
make_partition(model, raid, size=raid.free_for_partitions//2)
self.assertActionPossible(raid, DeviceAction.PARTITION)
make_partition(model, raid, size=raid.free_for_partitions)
self.assertActionNotPossible(raid, DeviceAction.PARTITION)
# Can partition a raid with .preserve=True
raid2 = make_raid(model)
raid2.preserve = True
self.assertActionPossible(raid2, DeviceAction.PARTITION)
# But not if it has a partition.
make_partition(model, raid2, preserve=True)
self.assertActionNotPossible(raid2, DeviceAction.PARTITION)
def test_raid_action_CREATE_LV(self):
model, raid = make_model_and_raid()
self.assertActionNotSupported(raid, DeviceAction.CREATE_LV)
def test_raid_action_FORMAT(self):
model, raid = make_model_and_raid()
self.assertActionPossible(raid, DeviceAction.FORMAT)
make_partition(model, raid)
self.assertActionNotPossible(raid, DeviceAction.FORMAT)
raid2 = make_raid(model)
model.add_volgroup('vg1', {raid2})
self.assertActionNotPossible(raid2, DeviceAction.FORMAT)
def test_raid_action_REMOVE(self):
model = make_model()
raids = [make_raid(model) for i in range(5)]
self._test_remove_action(model, raids)
def test_raid_action_DELETE(self):
model, raid = make_model_and_raid()
raid1 = make_raid(model)
self.assertActionPossible(raid1, DeviceAction.DELETE)
part = make_partition(model, raid1)
self.assertActionPossible(raid1, DeviceAction.DELETE)
fs = model.add_filesystem(part, 'ext4')
self.assertActionPossible(raid1, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionNotPossible(raid1, DeviceAction.DELETE)
raid2 = make_raid(model)
self.assertActionPossible(raid2, DeviceAction.DELETE)
fs = model.add_filesystem(raid2, 'ext4')
self.assertActionPossible(raid2, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionPossible(raid2, DeviceAction.DELETE)
raid2 = make_raid(model)
model.add_volgroup('vg0', {raid2})
self.assertActionNotPossible(raid2, DeviceAction.DELETE)
def test_raid_action_TOGGLE_BOOT(self):
model, raid = make_model_and_raid()
self.assertActionNotSupported(raid, DeviceAction.TOGGLE_BOOT)
def test_vg_action_INFO(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.INFO)
def test_vg_action_EDIT(self):
model, vg = make_model_and_vg()
self.assertActionPossible(vg, DeviceAction.EDIT)
model.add_logical_volume(vg, 'lv1', size=vg.free_for_partitions//2)
self.assertActionNotPossible(vg, DeviceAction.EDIT)
vg2 = make_vg(model)
vg2.preserve = True
self.assertActionNotPossible(vg2, DeviceAction.EDIT)
def test_vg_action_REFORMAT(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.REFORMAT)
def test_vg_action_PARTITION(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.PARTITION)
def test_vg_action_CREATE_LV(self):
model, vg = make_model_and_vg()
self.assertActionPossible(vg, DeviceAction.CREATE_LV)
model.add_logical_volume(vg, 'lv1', size=vg.free_for_partitions//2)
self.assertActionPossible(vg, DeviceAction.CREATE_LV)
model.add_logical_volume(vg, 'lv2', size=vg.free_for_partitions)
self.assertActionNotPossible(vg, DeviceAction.CREATE_LV)
vg2 = make_vg(model)
vg2.preserve = True
self.assertActionNotPossible(vg2, DeviceAction.CREATE_LV)
def test_vg_action_FORMAT(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.FORMAT)
def test_vg_action_REMOVE(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.REMOVE)
def test_vg_action_DELETE(self):
model, vg = make_model_and_vg()
self.assertActionPossible(vg, DeviceAction.DELETE)
self.assertActionPossible(vg, DeviceAction.DELETE)
lv = model.add_logical_volume(
vg, 'lv0', size=vg.free_for_partitions//2)
self.assertActionPossible(vg, DeviceAction.DELETE)
fs = model.add_filesystem(lv, 'ext4')
self.assertActionPossible(vg, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionNotPossible(vg, DeviceAction.DELETE)
def test_vg_action_TOGGLE_BOOT(self):
model, vg = make_model_and_vg()
self.assertActionNotSupported(vg, DeviceAction.TOGGLE_BOOT)
def test_lv_action_INFO(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.INFO)
def test_lv_action_EDIT(self):
model, lv = make_model_and_lv()
self.assertActionPossible(lv, DeviceAction.EDIT)
def test_lv_action_REFORMAT(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.REFORMAT)
def test_lv_action_PARTITION(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.PARTITION)
def test_lv_action_CREATE_LV(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.CREATE_LV)
def test_lv_action_FORMAT(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.FORMAT)
def test_lv_action_REMOVE(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.REMOVE)
def test_lv_action_DELETE(self):
model, lv = make_model_and_lv()
self.assertActionPossible(lv, DeviceAction.DELETE)
fs = model.add_filesystem(lv, 'ext4')
self.assertActionPossible(lv, DeviceAction.DELETE)
model.add_mount(fs, '/')
self.assertActionPossible(lv, DeviceAction.DELETE)
lv2 = make_lv(model)
lv2.preserve = lv2.volgroup.preserve = True
self.assertActionNotPossible(lv2, DeviceAction.DELETE)
def test_lv_action_TOGGLE_BOOT(self):
model, lv = make_model_and_lv()
self.assertActionNotSupported(lv, DeviceAction.TOGGLE_BOOT)
def test_is_esp(self):
model = make_model(Bootloader.UEFI)
gpt_disk = make_disk(model, ptable='gpt')

View File

@ -37,7 +37,10 @@ from subiquitycore.lsb_release import lsb_release
from subiquity.common.apidef import API
from subiquity.common.errorreport import ErrorReportKind
from subiquity.common.filesystem import FilesystemManipulator
from subiquity.common.filesystem.actions import (
DeviceAction,
)
from subiquity.common.filesystem.manipulator import FilesystemManipulator
from subiquity.common.types import (
Bootloader,
GuidedChoice,
@ -47,7 +50,6 @@ from subiquity.common.types import (
)
from subiquity.models.filesystem import (
dehumanize_size,
DeviceAction,
)
from subiquity.server.controller import (
SubiquityController,
@ -129,7 +131,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
def guided_lvm(self, disk, lvm_options=None):
self.reformat(disk)
if DeviceAction.TOGGLE_BOOT in disk.supported_actions:
if DeviceAction.TOGGLE_BOOT in DeviceAction.supported(disk):
self.add_boot_disk(disk)
self.create_partition(
device=disk, spec=dict(

View File

@ -60,8 +60,10 @@ from subiquitycore.ui.utils import (
)
from subiquitycore.view import BaseView
from subiquity.models.filesystem import (
from subiquity.common.filesystem.actions import (
DeviceAction,
)
from subiquity.models.filesystem import (
humanize_size,
)
@ -337,11 +339,11 @@ class DeviceList(WidgetWrap):
def _action_menu_for_device(self, device):
device_actions = []
for action in device.supported_actions:
for action in DeviceAction.supported(device):
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