Move all LVM helpers to dedicated module (#4283)

archinstall/lib/disk/device_handler.py

@@ -1,11 +1,10 @@
 import logging
 import os
-from collections.abc import Iterable
 from pathlib import Path
 
 from parted import Device, Disk, DiskException, FileSystem, Geometry, IOException, Partition, PartitionException, freshDisk, getAllDevices, getDevice, newDisk
 
-from archinstall.lib.command import SysCommand, SysCommandWorker
+from archinstall.lib.command import SysCommand
 from archinstall.lib.disk.utils import (
 	find_lsblk_info,
 	get_all_lsblk_info,
@@ -24,14 +23,11 @@ from archinstall.lib.models.device import (
 	DiskEncryption,
 	FilesystemType,
 	LsblkInfo,
-	LvmVolume,
-	LvmVolumeGroup,
 	ModificationStatus,
 	PartitionFlag,
 	PartitionGUID,
 	PartitionModification,
 	PartitionTable,
-	Size,
 	SubvolumeModification,
 	Unit,
 	_BtrfsSubvolumeInfo,
@@ -339,58 +335,6 @@ class DeviceHandler:
 		info(f'luks2 locking device: {dev_path}')
 		luks_handler.lock()
 
-	def lvm_export_vg(self, vg: LvmVolumeGroup) -> None:
-		cmd = f'vgexport {vg.name}'
-
-		debug(f'vgexport: {cmd}')
-		SysCommand(cmd)
-
-	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]) -> None:
-		pvs_str = ' '.join(str(pv) for pv in pvs)
-		# Signatures are already wiped by wipefs, -f is just for safety
-		cmd = f'pvcreate -f --yes {pvs_str}'
-		# note flags used in scripting
-		debug(f'Creating LVM PVS: {cmd}')
-		SysCommand(cmd)
-
-		# Sync with udev to ensure the PVs are visible
-		udev_sync()
-
-	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 --force {vg_name} {pvs_str}'
-
-		debug(f'Creating LVM group: {cmd}')
-		SysCommand(cmd)
-
-		# Sync with udev to ensure the VG is visible
-		udev_sync()
-
-	def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Size | None = None) -> None:
-		if offset is not None:
-			length = volume.length - offset
-		else:
-			length = volume.length
-
-		length_str = length.format_size(Unit.B, include_unit=False)
-		cmd = f'lvcreate --yes -L {length_str}B {vg_name} -n {volume.name}'
-
-		debug(f'Creating volume: {cmd}')
-
-		worker = SysCommandWorker(cmd)
-		worker.poll()
-		worker.write(b'y\n', line_ending=False)
-
-		volume.vg_name = vg_name
-		volume.dev_path = Path(f'/dev/{vg_name}/{volume.name}')
-
 	def _setup_partition(
 		self,
 		part_mod: PartitionModification,

archinstall/lib/disk/filesystem.py

@@ -3,7 +3,14 @@ import time
 from pathlib import Path
 
 from archinstall.lib.disk.device_handler import device_handler
-from archinstall.lib.disk.lvm import lvm_group_info, lvm_vol_info
+from archinstall.lib.disk.lvm import (
+	lvm_group_info,
+	lvm_pv_create,
+	lvm_vg_create,
+	lvm_vol_create,
+	lvm_vol_info,
+	lvm_vol_reduce,
+)
 from archinstall.lib.disk.utils import udev_sync
 from archinstall.lib.interactions.general_conf import confirm_abort
 from archinstall.lib.luks import Luks2
@@ -166,7 +173,7 @@ class FilesystemHandler:
 		for vg in lvm_config.vol_groups:
 			pv_dev_paths = self._get_all_pv_dev_paths(vg.pvs, enc_mods)
 
-			device_handler.lvm_vg_create(pv_dev_paths, vg.name)
+			lvm_vg_create(pv_dev_paths, vg.name)
 
 			# figure out what the actual available size in the group is
 			vg_info = lvm_group_info(vg.name)
@@ -199,7 +206,7 @@ class FilesystemHandler:
 				offset = max_vol_offset if lv == max_vol else None
 
 				debug(f'vg: {vg.name}, vol: {lv.name}, offset: {offset}')
-				device_handler.lvm_vol_create(vg.name, lv, offset)
+				lvm_vol_create(vg.name, lv, offset)
 
 				while True:
 					debug('Fetching LVM volume info')
@@ -241,7 +248,7 @@ class FilesystemHandler:
 		for vg in lvm_config.vol_groups:
 			pv_paths |= self._get_all_pv_dev_paths(vg.pvs, enc_mods)
 
-		device_handler.lvm_pv_create(pv_paths)
+		lvm_pv_create(pv_paths)
 
 	def _get_all_pv_dev_paths(
 		self,
@@ -319,7 +326,7 @@ class FilesystemHandler:
 		if any([vol.fs_type == FilesystemType.Ext4 for vol in vol_gp.volumes]):
 			largest_vol = max(vol_gp.volumes, key=lambda x: x.length)
 
-			device_handler.lvm_vol_reduce(
+			lvm_vol_reduce(
 				largest_vol.safe_dev_path,
 				Size(256, Unit.MiB, SectorSize.default()),
 			)

archinstall/lib/disk/lvm.py

@@ -1,9 +1,11 @@
 import json
 import time
+from collections.abc import Iterable
 from pathlib import Path
 from typing import Literal, overload
 
-from archinstall.lib.command import SysCommand
+from archinstall.lib.command import SysCommand, SysCommandWorker
+from archinstall.lib.disk.utils import udev_sync
 from archinstall.lib.exceptions import SysCallError
 from archinstall.lib.models.device import (
 	LvmGroupInfo,
@@ -116,6 +118,13 @@ def lvm_vol_change(vol: LvmVolume, activate: bool) -> None:
 	SysCommand(cmd)
 
 
+def lvm_export_vg(vg: LvmVolumeGroup) -> None:
+	cmd = f'vgexport {vg.name}'
+
+	debug(f'vgexport: {cmd}')
+	SysCommand(cmd)
+
+
 def lvm_import_vg(vg: LvmVolumeGroup) -> None:
 	# Check if the VG is actually exported before trying to import it
 	check_cmd = f'vgs --noheadings -o vg_exported {vg.name}'
@@ -135,3 +144,53 @@ def lvm_import_vg(vg: LvmVolumeGroup) -> None:
 	cmd = f'vgimport {vg.name}'
 	debug(f'vgimport: {cmd}')
 	SysCommand(cmd)
+
+
+def lvm_vol_reduce(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(pvs: Iterable[Path]) -> None:
+	pvs_str = ' '.join(str(pv) for pv in pvs)
+	# Signatures are already wiped by wipefs, -f is just for safety
+	cmd = f'pvcreate -f --yes {pvs_str}'
+	# note flags used in scripting
+	debug(f'Creating LVM PVS: {cmd}')
+	SysCommand(cmd)
+
+	# Sync with udev to ensure the PVs are visible
+	udev_sync()
+
+
+def lvm_vg_create(pvs: Iterable[Path], vg_name: str) -> None:
+	pvs_str = ' '.join(str(pv) for pv in pvs)
+	cmd = f'vgcreate --yes --force {vg_name} {pvs_str}'
+
+	debug(f'Creating LVM group: {cmd}')
+	SysCommand(cmd)
+
+	# Sync with udev to ensure the VG is visible
+	udev_sync()
+
+
+def lvm_vol_create(vg_name: str, volume: LvmVolume, offset: Size | None = None) -> None:
+	if offset is not None:
+		length = volume.length - offset
+	else:
+		length = volume.length
+
+	length_str = length.format_size(Unit.B, include_unit=False)
+	cmd = f'lvcreate --yes -L {length_str}B {vg_name} -n {volume.name}'
+
+	debug(f'Creating volume: {cmd}')
+
+	worker = SysCommandWorker(cmd)
+	worker.poll()
+	worker.write(b'y\n', line_ending=False)
+
+	volume.vg_name = vg_name
+	volume.dev_path = Path(f'/dev/{vg_name}/{volume.name}')