Refactor

Author: svartkanin

archinstall/lib/disk/__init__.py

@@ -46,10 +46,3 @@
 )
 
 from .disk_menu import DiskLayoutConfigurationMenu
-
-from .lvm_menu import (
-	LvmConfigurationMenu,
-	LvmVolumeList,
-	select_lvm_pvs,
-	select_lvm_vol_group
-)

archinstall/lib/disk/device_handler.py

@@ -5,7 +5,7 @@
 import time
 import logging
 from pathlib import Path
-from typing import List, Dict, Any, Optional, TYPE_CHECKING
+from typing import List, Dict, Any, Optional, TYPE_CHECKING, Literal
 
 from parted import (  # type: ignore
 	Disk, Geometry, FileSystem,
@@ -272,76 +272,54 @@ def format_enc(
 		info(f'luks2 locking device: {dev_path}')
 		luks_handler.lock()
 
-	def lvm_vol_info(self, lv_name: str) -> Optional[LvmVolumeInfo]:
-		cmd = 'lvs --reportformat json ' \
-			  '--unit B ' \
-			  f'-S lv_name={lv_name}'
-
+	def _lvm_info(self, cmd: str, name: str, info_type: Literal['lv', 'vg']) -> Optional[Any]:
 		raw_info = SysCommand(cmd).decode().split('\n')
 
 		# for whatever reason the output sometimes contains
 		# "File descriptor X leaked leaked on vgs invocation
-		data = '\n'.join([l for l in raw_info if 'File descriptor' not in l])
+		data = '\n'.join([raw for raw in raw_info if 'File descriptor' not in raw])
 
-		debug(f'LVM info: {data}')
+		debug(f'LVM raw: {data}')
 
 		reports = json.loads(data)
 
-		lv = None
+		type_name = f'{info_type}_name'
 
 		for report in reports['report']:
-			lvs = report.get('lv', [])
-			for lv_data in lvs:
-				if lv_data['lv_name'] == lv_name:
-					lv = lv_data
-
-		if lv is None:
-			return None
+			entries = report.get(info_type, [])
+			for entry in entries:
+				if entry[type_name] == name:
+					lv_size = int(entry[f'{info_type}_size'][:-1])
+					size = Size(lv_size, Unit.B, SectorSize.default())
+
+					match info_type:
+						case 'lv':
+							return LvmVolumeInfo(
+								lv_name=entry['lv_name'],
+								vg_name=entry['vg_name'],
+								lv_size=size
+							)
+						case 'vg':
+							return LvmGroupInfo(
+								vg_size=size,
+								vg_uuid=entry['vg_uuid']
+							)
+		return None
 
-		lv_size = int(lv['lv_size'][:-1])
-		size = Size(lv_size, Unit.B, SectorSize.default())
+	def lvm_vol_info(self, lv_name: str) -> Optional[LvmVolumeInfo]:
+		cmd = 'lvs --reportformat json ' \
+			'--unit B ' \
+			f'-S lv_name={lv_name}'
 
-		return LvmVolumeInfo(
-			lv_name=lv['lv_name'],
-			vg_name=lv['vg_name'],
-			lv_size=size
-		)
+		return self._lvm_info(cmd, lv_name, 'lv')
 
 	def lvm_group_info(self, vg_name: str) -> Optional[LvmGroupInfo]:
 		cmd = 'vgs --reportformat json ' \
-			  '--unit B ' \
-			  '-o vg_name,vg_uuid,vg_size ' \
-			  f'-S vg_name={vg_name}'
+			'--unit B ' \
+			'-o vg_name,vg_uuid,vg_size ' \
+			f'-S vg_name={vg_name}'
 
-		debug(f'LVM group info: {cmd}')
-		raw_info = SysCommand(cmd).decode().strip().split('\n')
-
-		# for whatever reason the output sometimes contains
-		# "File descriptor X leaked leaked on vgs invocation
-		data = '\n'.join([l for l in raw_info if 'File descriptor' not in l])
-
-		debug(f'VG info: {data}')
-
-		reports = json.loads(data)
-
-		vg = None
-
-		for report in reports['report']:
-			vgs = report.get('vg', [])
-			for vg_data in vgs:
-				if vg_data['vg_name'] == vg_name:
-					vg = vg_data
-
-		if vg is None:
-			return None
-
-		vg_size = int(vg['vg_size'][:-1])
-		size = Size(vg_size, Unit.B, SectorSize.default())
-
-		return LvmGroupInfo(
-			vg_size=size,
-			vg_uuid=vg['vg_uuid']
-		)
+		return self._lvm_info(cmd, vg_name, 'vg')
 
 	def lvm_vol_reduce(self, vol_path: Path, amount: Size):
 		val = amount.format_size(Unit.B, include_unit=False)
@@ -383,7 +361,7 @@ def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Optional[Size]
 		worker.poll()
 		worker.write(b'y\n', line_ending=False)
 
-		volume.dev_path = f'/dev/{vg_name}/{volume.name}'
+		volume.dev_path = Path(f'/dev/{vg_name}/{volume.name}')
 
 	def _perform_partitioning(
 		self,

archinstall/lib/disk/device_model.py

@@ -68,7 +68,7 @@ def json(self) -> Dict[str, Any]:
 			return config
 
 	@classmethod
-	def parse_arg(cls, disk_config: Dict[str, Dict[str, Any]]) -> Optional[DiskLayoutConfiguration]:
+	def parse_arg(cls, disk_config: Dict[str, Any]) -> Optional[DiskLayoutConfiguration]:
 		from .device_handler import device_handler
 
 		device_modifications: List[DeviceModification] = []
@@ -136,7 +136,7 @@ def parse_arg(cls, disk_config: Dict[str, Dict[str, Any]]) -> Optional[DiskLayou
 			device_modifications.append(device_modification)
 
 		# Parse LVM configuration from settings
-		if lvm_arg := disk_config.get('lvm_config', None):
+		if (lvm_arg := disk_config.get('lvm_config', None)) is not None:
 			config.lvm_config = LvmConfiguration.parse_arg(lvm_arg, config)
 
 		return config

archinstall/lib/disk/filesystem.py

@@ -148,6 +148,9 @@ def setup_lvm(
 				# figure out what the actual available size in the group is
 				lvm_gp_info = device_handler.lvm_group_info(vol_gp.name)
 
+				if not lvm_gp_info:
+					raise ValueError('Unable to fetch VG info')
+
 				# the actual available LVM Group size will be smaller than the
 				# total PVs size due to reserved metadata storage etc.
 				# so we'll have a look at the total avail. size, check the delta
@@ -184,7 +187,7 @@ def setup_lvm(
 					device_handler.format(lv.fs_type, lv.safe_dev_path)
 
 					if lv.fs_type == FilesystemType.Btrfs:
-						device_handler.create_btrfs_volumes(lv, enc_conf=self._enc_config)
+						device_handler.create_lvm_btrfs_subvolumes(lv, enc_conf=self._enc_config)
 
 	def _lvm_vol_handle_e2scrub(self, vol_gp: LvmVolumeGroup):
 		# from arch wiki: