Merge branch 'main' into refactor/delta_calculation_limit_concurrency

src/otaclient/app/boot_control/_common.py

@@ -32,6 +32,7 @@
     subprocess_check_output,
     write_str_to_file_sync,
 )
+from otaclient_common.typing import StrOrPath
 
 logger = logging.getLogger(__name__)
 
@@ -194,6 +195,30 @@ def get_parent_dev(cls, child_device: str, *, raise_exception: bool = True) -> s
         cmd = ["lsblk", "-idpno", "PKNAME", child_device]
         return subprocess_check_output(cmd, raise_exception=raise_exception)
 
+    @classmethod
+    def get_device_tree(
+        cls, parent_dev: str, *, raise_exception: bool = True
+    ) -> list[str]:
+        """Get the device tree of a parent device.
+
+        For example, for sda with 3 partitions, we will get:
+        ["/dev/sda", "/dev/sda1", "/dev/sda2", "/dev/sda3"]
+
+        This function is implemented by calling:
+            lsblk -lnpo NAME <parent_dev>
+
+        Args:
+            parent_dev (str): The parent device to be checked.
+            raise_exception (bool, optional): raise exception on subprocess call failed.
+                Defaults to True.
+
+        Returns:
+            str: _description_
+        """
+        cmd = ["lsblk", "-lnpo", "NAME", parent_dev]
+        raw_res = subprocess_check_output(cmd, raise_exception=raise_exception)
+        return raw_res.splitlines()
+
     @classmethod
     def set_ext4_fslabel(cls, dev: str, fslabel: str, *, raise_exception: bool = True):
         """Set <fslabel> to ext4 formatted <dev>.
@@ -210,6 +235,36 @@ def set_ext4_fslabel(cls, dev: str, fslabel: str, *, raise_exception: bool = Tru
         cmd = ["e2label", dev, fslabel]
         subprocess_call(cmd, raise_exception=raise_exception)
 
+    @classmethod
+    def mount(
+        cls,
+        target: StrOrPath,
+        mount_point: StrOrPath,
+        *,
+        options: list[str] | None = None,
+        params: list[str] | None = None,
+        raise_exception: bool = True,
+    ) -> None:
+        """Thin wrapper to call mount using subprocess.
+
+        This will call the following:
+            mount [-o <option1>,[<option2>[,...]] [<param1> [<param2>[...]]] <target> <mount_point>
+
+        Args:
+            target (StrOrPath): The target device to mount.
+            mount_point (StrOrPath): The mount point to mount to.
+            options (list[str] | None, optional): A list of options, append after -o. Defaults to None.
+            params (list[str] | None, optional): A list of params. Defaults to None.
+            raise_exception (bool, optional): Whether to raise exception on failed call. Defaults to True.
+        """
+        cmd = ["mount"]
+        if options:
+            cmd.extend(["-o", ",".join(options)])
+        if params:
+            cmd.extend(params)
+        cmd = [*cmd, str(target), str(mount_point)]
+        subprocess_call(cmd, raise_exception=raise_exception)
+
     @classmethod
     def mount_rw(
         cls, target: str, mount_point: Path | str, *, raise_exception: bool = True
@@ -724,7 +779,7 @@ def prepare_standby_dev(
         # TODO: in the future if in-place update mode is implemented, do a
         #   fschck over the standby slot file system.
         if fslabel:
-            CMDHelperFuncs.set_ext4_fslabel(self.active_slot_dev, fslabel=fslabel)
+            CMDHelperFuncs.set_ext4_fslabel(self.standby_slot_dev, fslabel=fslabel)
 
     def umount_all(self, *, ignore_error: bool = True):
         logger.debug("unmount standby slot and active slot mount point...")