Break an import cycle between lib/luks and lib/disk/device_handler (#3438)

Author: correctmost

archinstall/lib/disk/device_handler.py

@@ -45,6 +45,7 @@
 	find_lsblk_info,
 	get_all_lsblk_info,
 	get_lsblk_info,
+	umount,
 )
 
 
@@ -257,7 +258,7 @@ def get_btrfs_info(
 			subvol_infos.append(_BtrfsSubvolumeInfo(name, sub_vol_mountpoint))
 
 		if not lsblk_info.mountpoint:
-			self.umount(dev_path)
+			umount(dev_path)
 
 		return subvol_infos
 
@@ -635,7 +636,7 @@ def create_lvm_btrfs_subvolumes(
 				except SysCallError as err:
 					raise DiskError(f'Could not set compress attribute at {subvol_path}: {err}')
 
-		self.umount(path)
+		umount(path)
 
 	def create_btrfs_volumes(
 		self,
@@ -677,7 +678,7 @@ def create_btrfs_volumes(
 
 			SysCommand(f"btrfs subvolume create -p {subvol_path}")
 
-		self.umount(dev_path)
+		umount(dev_path)
 
 		if luks_handler is not None and luks_handler.mapper_dev is not None:
 			luks_handler.lock()
@@ -710,7 +711,7 @@ def umount_all_existing(self, device_path: Path) -> None:
 			if partition.fs_type == FilesystemType.Crypto_luks:
 				Luks2(partition.path).lock()
 			else:
-				self.umount(partition.path, recursive=True)
+				umount(partition.path, recursive=True)
 
 	def partition(
 		self,
@@ -790,23 +791,6 @@ def mount(
 		except SysCallError as err:
 			raise DiskError(f'Could not mount {dev_path}: {command}\n{err.message}')
 
-	def umount(self, mountpoint: Path, recursive: bool = False) -> None:
-		lsblk_info = get_lsblk_info(mountpoint)
-
-		if not lsblk_info.mountpoints:
-			return
-
-		debug(f'Partition {mountpoint} is currently mounted at: {[str(m) for m in lsblk_info.mountpoints]}')
-
-		cmd = ['umount']
-
-		if recursive:
-			cmd.append('-R')
-
-		for path in lsblk_info.mountpoints:
-			debug(f'Unmounting mountpoint: {path}')
-			SysCommand(cmd + [str(path)])
-
 	def detect_pre_mounted_mods(self, base_mountpoint: Path) -> list[DeviceModification]:
 		part_mods: dict[Path, list[PartitionModification]] = {}
 

archinstall/lib/disk/utils.py

@@ -108,3 +108,21 @@ def disk_layouts() -> str:
 		return ''
 
 	return lsblk_output.model_dump_json(indent=4)
+
+
+def umount(mountpoint: Path, recursive: bool = False) -> None:
+	lsblk_info = get_lsblk_info(mountpoint)
+
+	if not lsblk_info.mountpoints:
+		return
+
+	debug(f'Partition {mountpoint} is currently mounted at: {[str(m) for m in lsblk_info.mountpoints]}')
+
+	cmd = ['umount']
+
+	if recursive:
+		cmd.append('-R')
+
+	for path in lsblk_info.mountpoints:
+		debug(f'Unmounting mountpoint: {path}')
+		SysCommand(cmd + [str(path)])

archinstall/lib/luks.py

@@ -5,7 +5,7 @@
 from pathlib import Path
 from subprocess import CalledProcessError
 
-from archinstall.lib.disk.utils import get_lsblk_info
+from archinstall.lib.disk.utils import get_lsblk_info, umount
 
 from .exceptions import DiskError, SysCallError
 from .general import SysCommand, SysCommandWorker, generate_password, run
@@ -153,8 +153,7 @@ def unlock(self, key_file: Path | None = None) -> None:
 			raise DiskError(f'Failed to open luks2 device: {self.luks_dev_path}')
 
 	def lock(self) -> None:
-		from archinstall.lib.disk.device_handler import device_handler
-		device_handler.umount(self.luks_dev_path)
+		umount(self.luks_dev_path)
 
 		# Get crypt-information about the device by doing a reverse lookup starting with the partition path
 		# For instance: /dev/sda
@@ -165,7 +164,7 @@ def lock(self) -> None:
 			# Unmount the child location
 			for mountpoint in child.mountpoints:
 				debug(f'Unmounting {mountpoint}')
-				device_handler.umount(mountpoint, recursive=True)
+				umount(mountpoint, recursive=True)
 
 			# And close it if possible.
 			debug(f"Closing crypt device {child.name}")