From 63b4184f706ba1fe9a4eb5496d03c6c679cf91de Mon Sep 17 00:00:00 2001 From: correctmost <134317971+correctmost@users.noreply.github.com> Date: Tue, 27 Aug 2024 15:40:51 -0400 Subject: [PATCH] Fix most of the mypy errors in archinstall/lib/disk/ (#2639) --- archinstall/lib/disk/device_handler.py | 42 +++++++++++------------ archinstall/lib/disk/device_model.py | 42 +++++++++++------------ archinstall/lib/disk/disk_menu.py | 2 +- archinstall/lib/disk/encryption_menu.py | 2 +- archinstall/lib/disk/fido.py | 2 +- archinstall/lib/disk/filesystem.py | 20 +++++------ archinstall/lib/disk/partitioning_menu.py | 6 ++-- 7 files changed, 58 insertions(+), 58 deletions(-) diff --git a/archinstall/lib/disk/device_handler.py b/archinstall/lib/disk/device_handler.py index a98dd2c1..3adf3671 100644 --- a/archinstall/lib/disk/device_handler.py +++ b/archinstall/lib/disk/device_handler.py @@ -36,7 +36,7 @@ if TYPE_CHECKING: class DeviceHandler(object): _TMP_BTRFS_MOUNT = Path('/mnt/arch_btrfs') - def __init__(self): + def __init__(self) -> None: self._devices: Dict[Path, BDevice] = {} self.load_devices() @@ -44,7 +44,7 @@ class DeviceHandler(object): def devices(self) -> List[BDevice]: return list(self._devices.values()) - def load_devices(self): + def load_devices(self) -> None: block_devices = {} SysCommand('udevadm settle') @@ -216,7 +216,7 @@ class DeviceHandler(object): fs_type: FilesystemType, path: Path, additional_parted_options: List[str] = [] - ): + ) -> None: mkfs_type = fs_type.value options = [] @@ -282,7 +282,7 @@ class DeviceHandler(object): mapper_name: Optional[str], fs_type: FilesystemType, enc_conf: DiskEncryption - ): + ) -> None: luks_handler = Luks2( dev_path, mapper_name=mapper_name, @@ -384,33 +384,33 @@ class DeviceHandler(object): return self._lvm_info_with_retry(cmd, 'pvseg') - def lvm_vol_change(self, vol: LvmVolume, activate: bool): + def lvm_vol_change(self, vol: LvmVolume, activate: bool) -> None: active_flag = 'y' if activate else 'n' cmd = f'lvchange -a {active_flag} {vol.safe_dev_path}' debug(f'lvchange volume: {cmd}') SysCommand(cmd) - def lvm_export_vg(self, vg: LvmVolumeGroup): + def lvm_export_vg(self, vg: LvmVolumeGroup) -> None: cmd = f'vgexport {vg.name}' debug(f'vgexport: {cmd}') SysCommand(cmd) - def lvm_import_vg(self, vg: LvmVolumeGroup): + def lvm_import_vg(self, vg: LvmVolumeGroup) -> None: cmd = f'vgimport {vg.name}' debug(f'vgimport: {cmd}') SysCommand(cmd) - def lvm_vol_reduce(self, vol_path: Path, amount: Size): + def lvm_vol_reduce(self, vol_path: Path, amount: Size) -> None: val = amount.format_size(Unit.B, include_unit=False) cmd = f'lvreduce -L -{val}B {vol_path}' debug(f'Reducing LVM volume size: {cmd}') SysCommand(cmd) - def lvm_pv_create(self, pvs: Iterable[Path]): + def lvm_pv_create(self, pvs: Iterable[Path]) -> None: cmd = 'pvcreate ' + ' '.join([str(pv) for pv in pvs]) debug(f'Creating LVM PVS: {cmd}') @@ -418,7 +418,7 @@ class DeviceHandler(object): worker.poll() worker.write(b'y\n', line_ending=False) - def lvm_vg_create(self, pvs: Iterable[Path], vg_name: str): + def lvm_vg_create(self, pvs: Iterable[Path], vg_name: str) -> None: pvs_str = ' '.join([str(pv) for pv in pvs]) cmd = f'vgcreate --yes {vg_name} {pvs_str}' @@ -428,7 +428,7 @@ class DeviceHandler(object): worker.poll() worker.write(b'y\n', line_ending=False) - def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Optional[Size] = None): + def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Optional[Size] = None) -> None: if offset is not None: length = volume.length - offset else: @@ -452,7 +452,7 @@ class DeviceHandler(object): block_device: BDevice, disk: Disk, requires_delete: bool - ): + ) -> None: # when we require a delete and the partition to be (re)created # already exists then we have to delete it first if requires_delete and part_mod.status in [ModificationStatus.Modify, ModificationStatus.Delete]: @@ -535,7 +535,7 @@ class DeviceHandler(object): path: Path, btrfs_subvols: List[SubvolumeModification], mount_options: List[str] - ): + ) -> None: info(f'Creating subvolumes: {path}') self.mount(path, self._TMP_BTRFS_MOUNT, create_target_mountpoint=True) @@ -565,7 +565,7 @@ class DeviceHandler(object): self, part_mod: PartitionModification, enc_conf: Optional['DiskEncryption'] = None - ): + ) -> None: info(f'Creating subvolumes: {part_mod.safe_dev_path}') # unlock the partition first if it's encrypted @@ -617,7 +617,7 @@ class DeviceHandler(object): return luks_handler - def umount_all_existing(self, device_path: Path): + def umount_all_existing(self, device_path: Path) -> None: debug(f'Unmounting all existing partitions: {device_path}') existing_partitions = self._devices[device_path].partition_infos @@ -635,7 +635,7 @@ class DeviceHandler(object): self, modification: DeviceModification, partition_table: Optional[PartitionTable] = None - ): + ) -> None: """ Create a partition table on the block device and create all partitions. """ @@ -675,7 +675,7 @@ class DeviceHandler(object): mount_fs: Optional[str] = None, create_target_mountpoint: bool = True, options: List[str] = [] - ): + ) -> None: if create_target_mountpoint and not target_mountpoint.exists(): target_mountpoint.mkdir(parents=True, exist_ok=True) @@ -705,7 +705,7 @@ class DeviceHandler(object): except SysCallError as err: raise DiskError(f'Could not mount {dev_path}: {command}\n{err.message}') - def umount(self, mountpoint: Path, recursive: bool = False): + def umount(self, mountpoint: Path, recursive: bool = False) -> None: lsblk_info = get_lsblk_info(mountpoint) if not lsblk_info.mountpoints: @@ -748,7 +748,7 @@ class DeviceHandler(object): return device_mods - def partprobe(self, path: Optional[Path] = None): + def partprobe(self, path: Optional[Path] = None) -> None: if path is not None: command = f'partprobe {path}' else: @@ -763,7 +763,7 @@ class DeviceHandler(object): else: error(f'"{command}" failed to run (continuing anyway): {err}') - def _wipe(self, dev_path: Path): + def _wipe(self, dev_path: Path) -> None: """ Wipe a device (partition or otherwise) of meta-data, be it file system, LVM, etc. @param dev_path: Device path of the partition to be wiped. @@ -772,7 +772,7 @@ class DeviceHandler(object): with open(dev_path, 'wb') as p: p.write(bytearray(1024)) - def wipe_dev(self, block_device: BDevice): + def wipe_dev(self, block_device: BDevice) -> None: """ Wipe the block device of meta-data, be it file system, LVM, etc. This is not intended to be secure, but rather to ensure that diff --git a/archinstall/lib/disk/device_model.py b/archinstall/lib/disk/device_model.py index c2d095d2..4f7e7dfb 100644 --- a/archinstall/lib/disk/device_model.py +++ b/archinstall/lib/disk/device_model.py @@ -178,7 +178,7 @@ class SectorSize: value: int unit: Unit - def __post_init__(self): + def __post_init__(self) -> None: match self.unit: case Unit.sectors: raise ValueError('Unit type sector not allowed for SectorSize') @@ -204,7 +204,7 @@ class SectorSize: """ will normalize the value of the unit to Byte """ - return int(self.value * self.unit.value) # type: ignore + return int(self.value * self.unit.value) @dataclass @@ -213,7 +213,7 @@ class Size: unit: Unit sector_size: SectorSize - def __post_init__(self): + def __post_init__(self) -> None: if not isinstance(self.sector_size, SectorSize): raise ValueError('sector size must be of type SectorSize') @@ -253,7 +253,7 @@ class Size: sectors = math.ceil(norm / sector_size.value) return Size(sectors, Unit.sectors, sector_size) else: - value = int(self._normalize() / target_unit.value) # type: ignore + value = int(self._normalize() / target_unit.value) return Size(value, target_unit, self.sector_size) def as_text(self) -> str: @@ -293,7 +293,7 @@ class Size: """ if self.unit == Unit.sectors and self.sector_size is not None: return self.value * self.sector_size.normalize() - return int(self.value * self.unit.value) # type: ignore + return int(self.value * self.unit.value) def __sub__(self, other: Size) -> Size: src_norm = self._normalize() @@ -305,22 +305,22 @@ class Size: dest_norm = other._normalize() return Size(abs(src_norm + dest_norm), Unit.B, self.sector_size) - def __lt__(self, other): + def __lt__(self, other: Size) -> bool: return self._normalize() < other._normalize() - def __le__(self, other): + def __le__(self, other: Size) -> bool: return self._normalize() <= other._normalize() - def __eq__(self, other): + def __eq__(self, other) -> bool: return self._normalize() == other._normalize() - def __ne__(self, other): + def __ne__(self, other) -> bool: return self._normalize() != other._normalize() - def __gt__(self, other): + def __gt__(self, other: Size) -> bool: return self._normalize() > other._normalize() - def __ge__(self, other): + def __ge__(self, other: Size) -> bool: return self._normalize() >= other._normalize() @@ -556,7 +556,7 @@ class BDevice: device_info: _DeviceInfo partition_infos: List[_PartitionInfo] - def __hash__(self): + def __hash__(self) -> int: return hash(self.disk.device.path) @@ -680,7 +680,7 @@ class PartitionModification: _efi_indicator_flags = (PartitionFlag.Boot, PartitionFlag.ESP) _boot_indicator_flags = (PartitionFlag.Boot, PartitionFlag.XBOOTLDR) - def __post_init__(self): + def __post_init__(self) -> None: # needed to use the object as a dictionary key due to hash func if not hasattr(self, '_obj_id'): self._obj_id = uuid.uuid4() @@ -691,7 +691,7 @@ class PartitionModification: if self.fs_type is None and self.status == ModificationStatus.Modify: raise ValueError('FS type must not be empty on modifications with status type modify') - def __hash__(self): + def __hash__(self) -> int: return hash(self._obj_id) @property @@ -796,11 +796,11 @@ class PartitionModification: return f'{storage.get("ENC_IDENTIFIER", "ai")}{self.dev_path.name}' return None - def set_flag(self, flag: PartitionFlag): + def set_flag(self, flag: PartitionFlag) -> None: if flag not in self.flags: self.flags.append(flag) - def invert_flag(self, flag: PartitionFlag): + def invert_flag(self, flag: PartitionFlag) -> None: if flag in self.flags: self.flags = [f for f in self.flags if f != flag] else: @@ -915,12 +915,12 @@ class LvmVolume: # mapper device path /dev// dev_path: Optional[Path] = None - def __post_init__(self): + def __post_init__(self) -> None: # needed to use the object as a dictionary key due to hash func if not hasattr(self, '_obj_id'): self._obj_id = uuid.uuid4() - def __hash__(self): + def __hash__(self) -> int: return hash(self._obj_id) @property @@ -1050,7 +1050,7 @@ class LvmConfiguration: config_type: LvmLayoutType vol_groups: List[LvmVolumeGroup] - def __post_init__(self): + def __post_init__(self) -> None: # make sure all volume groups have unique PVs pvs = [] for group in self.vol_groups: @@ -1121,7 +1121,7 @@ class DeviceModification: def device_path(self) -> Path: return self.device.device_info.path - def add_partition(self, partition: PartitionModification): + def add_partition(self, partition: PartitionModification) -> None: self.partitions.append(partition) def get_efi_partition(self) -> Optional[PartitionModification]: @@ -1196,7 +1196,7 @@ class DiskEncryption: lvm_volumes: List[LvmVolume] = field(default_factory=list) hsm_device: Optional[Fido2Device] = None - def __post_init__(self): + def __post_init__(self) -> None: if self.encryption_type in [EncryptionType.Luks, EncryptionType.LvmOnLuks] and not self.partitions: raise ValueError('Luks or LvmOnLuks encryption require partitions to be defined') diff --git a/archinstall/lib/disk/disk_menu.py b/archinstall/lib/disk/disk_menu.py index 6de02b38..214526e7 100644 --- a/archinstall/lib/disk/disk_menu.py +++ b/archinstall/lib/disk/disk_menu.py @@ -29,7 +29,7 @@ class DiskLayoutConfigurationMenu(AbstractSubMenu): super().__init__(data_store=data_store, preview_size=0.5) - def setup_selection_menu_options(self): + def setup_selection_menu_options(self) -> None: self._menu_options['disk_config'] = \ Selector( _('Partitioning'), diff --git a/archinstall/lib/disk/encryption_menu.py b/archinstall/lib/disk/encryption_menu.py index b0e292ce..ccc1bace 100644 --- a/archinstall/lib/disk/encryption_menu.py +++ b/archinstall/lib/disk/encryption_menu.py @@ -40,7 +40,7 @@ class DiskEncryptionMenu(AbstractSubMenu): self._disk_config = disk_config super().__init__(data_store=data_store) - def setup_selection_menu_options(self): + def setup_selection_menu_options(self) -> None: self._menu_options['encryption_type'] = \ Selector( _('Encryption type'), diff --git a/archinstall/lib/disk/fido.py b/archinstall/lib/disk/fido.py index 5a139534..ea9d812a 100644 --- a/archinstall/lib/disk/fido.py +++ b/archinstall/lib/disk/fido.py @@ -74,7 +74,7 @@ class Fido2: hsm_device: Fido2Device, dev_path: Path, password: str - ): + ) -> None: worker = SysCommandWorker(f"systemd-cryptenroll --fido2-device={hsm_device.path} {dev_path}", peek_output=True) pw_inputted = False pin_inputted = False diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py index a4db1d4d..9af2e00c 100644 --- a/archinstall/lib/disk/filesystem.py +++ b/archinstall/lib/disk/filesystem.py @@ -32,7 +32,7 @@ class FilesystemHandler: self._disk_config = disk_config self._enc_config = enc_conf - def perform_filesystem_operations(self, show_countdown: bool = True): + def perform_filesystem_operations(self, show_countdown: bool = True) -> None: if self._disk_config.config_type == DiskLayoutType.Pre_mount: debug('Disk layout configuration is set to pre-mount, not performing any operations') return @@ -90,7 +90,7 @@ class FilesystemHandler: self, partitions: List[PartitionModification], device_path: Path - ): + ) -> None: """ Format can be given an overriding path, for instance /dev/null to test the formatting functionality and in essence the support for the given filesystem. @@ -122,7 +122,7 @@ class FilesystemHandler: part_mod.partuuid = lsblk_info.partuuid part_mod.uuid = lsblk_info.uuid - def _validate_partitions(self, partitions: List[PartitionModification]): + def _validate_partitions(self, partitions: List[PartitionModification]) -> None: checks = { # verify that all partitions have a path set (which implies that they have been created) lambda x: x.dev_path is None: ValueError('When formatting, all partitions must have a path set'), @@ -138,7 +138,7 @@ class FilesystemHandler: if found is not None: raise exc - def perform_lvm_operations(self): + def perform_lvm_operations(self) -> None: info('Setting up LVM config...') if not self._disk_config.lvm_config: @@ -153,7 +153,7 @@ class FilesystemHandler: self._setup_lvm(self._disk_config.lvm_config) self._format_lvm_vols(self._disk_config.lvm_config) - def _setup_lvm_encrypted(self, lvm_config: LvmConfiguration, enc_config: DiskEncryption): + def _setup_lvm_encrypted(self, lvm_config: LvmConfiguration, enc_config: DiskEncryption) -> None: if enc_config.encryption_type == EncryptionType.LvmOnLuks: enc_mods = self._encrypt_partitions(enc_config, lock_after_create=False) @@ -175,7 +175,7 @@ class FilesystemHandler: self._safely_close_lvm(lvm_config) - def _safely_close_lvm(self, lvm_config: LvmConfiguration): + def _safely_close_lvm(self, lvm_config: LvmConfiguration) -> None: for vg in lvm_config.vol_groups: for vol in vg.volumes: device_handler.lvm_vol_change(vol, False) @@ -186,7 +186,7 @@ class FilesystemHandler: self, lvm_config: LvmConfiguration, enc_mods: Dict[PartitionModification, Luks2] = {} - ): + ) -> None: self._lvm_create_pvs(lvm_config, enc_mods) for vg in lvm_config.vol_groups: @@ -233,7 +233,7 @@ class FilesystemHandler: self, lvm_config: LvmConfiguration, enc_vols: Dict[LvmVolume, Luks2] = {} - ): + ) -> None: for vol in lvm_config.get_all_volumes(): if enc_vol := enc_vols.get(vol, None): if not enc_vol.mapper_dev: @@ -253,7 +253,7 @@ class FilesystemHandler: self, lvm_config: LvmConfiguration, enc_mods: Dict[PartitionModification, Luks2] = {} - ): + ) -> None: pv_paths: Set[Path] = set() for vg in lvm_config.vol_groups: @@ -328,7 +328,7 @@ class FilesystemHandler: return enc_mods - def _lvm_vol_handle_e2scrub(self, vol_gp: LvmVolumeGroup): + def _lvm_vol_handle_e2scrub(self, vol_gp: LvmVolumeGroup) -> None: # from arch wiki: # If a logical volume will be formatted with ext4, leave at least 256 MiB # free space in the volume group to allow using e2scrub diff --git a/archinstall/lib/disk/partitioning_menu.py b/archinstall/lib/disk/partitioning_menu.py index fb1eb74b..68427213 100644 --- a/archinstall/lib/disk/partitioning_menu.py +++ b/archinstall/lib/disk/partitioning_menu.py @@ -148,7 +148,7 @@ class PartitioningList(ListManager): self, partition: PartitionModification, option: BtrfsMountOption - ): + ) -> None: if option.value not in partition.mount_options: if option == BtrfsMountOption.compress: partition.mount_options = [ @@ -167,13 +167,13 @@ class PartitioningList(ListManager): o for o in partition.mount_options if o != option.value ] - def _set_btrfs_subvolumes(self, partition: PartitionModification): + def _set_btrfs_subvolumes(self, partition: PartitionModification) -> None: partition.btrfs_subvols = SubvolumeMenu( _("Manage btrfs subvolumes for current partition"), partition.btrfs_subvols ).run() - def _prompt_formatting(self, partition: PartitionModification): + def _prompt_formatting(self, partition: PartitionModification) -> None: # an existing partition can toggle between Exist or Modify if partition.is_modify(): partition.status = ModificationStatus.Exist