refine(v3.7.x): refine bootctrl.common CMDHelperFuncs (#289)

This PR does the following things:
1. refines the common module's subprocess wrapper methods.
2. refines and simplify the boot_control.common.CMDHelperFuncs implementation a lot.
3. cleanup and remove boot_contro.errors, now each boot controller implementation has their own error types.
4. use mount slot helper's prepare_standby_dev instead of each boot controller's prepare_standby_dev method.
5. integrate the new CMDHelperFuncs to each boot controller and fix up test files accordingly.

NOTE that cboot boot controller implementation is temporarily removed, it will be added back in #287 .

otaclient/app/boot_control/_grub.py

@@ -31,6 +31,7 @@
 """
 
 
+from __future__ import annotations
 import logging
 import re
 import shutil
@@ -327,9 +328,16 @@ def _get_sibling_dev(self, active_dev: str) -> str:
             raise ValueError(_err_msg)
 
         # list children device file from parent device
-        cmd = f"-Pp -o NAME,FSTYPE {parent}"
+        cmd = ["lsblk", "-Ppo", "NAME,FSTYPE", parent]
+        try:
+            cmd_result = subprocess_check_output(cmd, raise_exception=True)
+        except Exception as e:
+            _err_msg = f"failed to detect boot device family tree: {e!r}"
+            logger.error(_err_msg)
+            raise _GrubBootControllerError(_err_msg) from e
+
         # exclude parent dev
-        output = CMDHelperFuncs._lsblk(cmd).splitlines()[1:]
+        output = cmd_result.splitlines()[1:]
         # FSTYPE="ext4" and
         # not (parent_device_file, root_device_file and boot_device_file)
         for blk in output:
@@ -352,7 +360,14 @@ def _detect_active_slot(self) -> Tuple[str, str]:
             A tuple contains the slot_name and the full dev path
             of the active slot.
         """
-        dev_path = CMDHelperFuncs.get_current_rootfs_dev()
+        try:
+            dev_path = CMDHelperFuncs.get_current_rootfs_dev()
+            assert dev_path
+        except Exception as e:
+            _err_msg = f"failed to detect current rootfs dev: {e!r}"
+            logger.error(_err_msg)
+            raise _GrubBootControllerError(_err_msg) from e
+
         _dev_path_ma = self.DEV_PATH_PA.match(dev_path)
         assert _dev_path_ma, f"dev path is invalid for OTA: {dev_path}"
 
@@ -605,7 +620,17 @@ def _grub_update_on_booted_slot(self, *, abort_on_standby_missed=True):
         active_slot_grub_file = self.active_ota_partition_folder / cfg.GRUB_CFG_FNAME
 
         grub_cfg_content = GrubHelper.grub_mkconfig()
-        standby_uuid_str = CMDHelperFuncs.get_uuid_str_by_dev(self.standby_root_dev)
+        try:
+            standby_uuid = CMDHelperFuncs.get_attrs_by_dev(
+                "UUID", self.standby_root_dev
+            )
+            assert standby_uuid
+        except Exception as e:
+            _err_msg = f"failed to get UUID of {self.standby_root_dev}: {e!r}"
+            logger.error(_err_msg)
+            raise _GrubBootControllerError(_err_msg) from e
+
+        standby_uuid_str = f"UUID={standby_uuid}"
         if grub_cfg_updated := GrubHelper.update_entry_rootfs(
             grub_cfg_content,
             kernel_ver=GrubHelper.SUFFIX_OTA_STANDBY,
@@ -667,27 +692,6 @@ def _ensure_standby_slot_boot_files_symlinks(self, standby_slot: str):
 
     # API
 
-    def prepare_standby_dev(self, *, erase_standby: bool):
-        """
-        Args:
-            erase_standby: indicate boot_controller whether to format the
-                standby slot's file system or not. This value is indicated and
-                passed to boot controller by the standby slot creator.
-        """
-        try:
-            # try to unmount the standby root dev unconditionally
-            if CMDHelperFuncs.is_target_mounted(self.standby_root_dev):
-                CMDHelperFuncs.umount(self.standby_root_dev)
-
-            if erase_standby:
-                CMDHelperFuncs.mkfs_ext4(self.standby_root_dev)
-            # TODO: check the standby file system status
-            #       if not erase the standby slot
-        except Exception as e:
-            _err_msg = f"failed to prepare standby dev: {e!r}"
-            logger.error(_err_msg)
-            raise _GrubBootControllerError(_err_msg) from e
-
     def finalize_update_switch_boot(self):
         """Finalize switch boot and use boot files from current booted slot."""
         # NOTE: since we have not yet switched boot, the active/standby relationship is
@@ -769,9 +773,19 @@ def _update_fstab(self, *, active_slot_fstab: Path, standby_slot_fstab: Path):
 
         Override existed entries in standby fstab, merge new entries from active fstab.
         """
-        standby_uuid_str = CMDHelperFuncs.get_uuid_str_by_dev(
-            self._boot_control.standby_root_dev
-        )
+        try:
+            standby_uuid = CMDHelperFuncs.get_attrs_by_dev(
+                "UUID", self._boot_control.standby_root_dev
+            )
+            assert standby_uuid
+        except Exception as e:
+            _err_msg = (
+                f"failed to get UUID of {self._boot_control.standby_root_dev}: {e!r}"
+            )
+            logger.error(_err_msg)
+            raise _GrubBootControllerError(_err_msg) from e
+
+        standby_uuid_str = f"UUID={standby_uuid}"
         fstab_entry_pa = re.compile(
             r"^\s*(?P<file_system>[^# ]*)\s+"
             r"(?P<mount_point>[^ ]*)\s+"
@@ -869,7 +883,7 @@ def pre_update(self, version: str, *, standby_as_ref: bool, erase_standby=False)
             self._ota_status_control.pre_update_current()
 
             ### mount slots ###
-            self._boot_control.prepare_standby_dev(erase_standby=erase_standby)
+            self._mp_control.prepare_standby_dev(erase_standby=erase_standby)
             self._mp_control.mount_standby()
             self._mp_control.mount_active()
 

otaclient/app/boot_control/_rpi_boot.py

@@ -14,6 +14,7 @@
 """Boot control support for Raspberry pi 4 Model B."""
 
 
+from __future__ import annotations
 import logging
 import os
 import re
@@ -23,7 +24,7 @@
 
 from .. import errors as ota_errors
 from ..proto import wrapper
-from ..common import replace_atomic, subprocess_call
+from ..common import replace_atomic, subprocess_call, subprocess_check_output
 
 from ._common import (
     OTAStatusFilesControl,
@@ -77,9 +78,8 @@ class _RPIBootControl:
 
     def __init__(self) -> None:
         self.system_boot_path = Path(cfg.SYSTEM_BOOT_MOUNT_POINT)
-        if not (
-            self.system_boot_path.is_dir()
-            and CMDHelperFuncs.is_target_mounted(self.system_boot_path)
+        if not CMDHelperFuncs.is_target_mounted(
+            self.system_boot_path, raise_exception=False
         ):
             _err_msg = "system-boot is not presented or not mounted!"
             logger.error(_err_msg)
@@ -92,29 +92,34 @@ def _init_slots_info(self):
         logger.debug("checking and initializing slots info...")
         try:
             # detect active slot
-            self._active_slot_dev = CMDHelperFuncs.get_dev_by_mount_point(
-                cfg.ACTIVE_ROOTFS_PATH
+            _active_slot_dev = CMDHelperFuncs.get_current_rootfs_dev()
+            assert _active_slot_dev
+            self._active_slot_dev = _active_slot_dev
+
+            _active_slot = CMDHelperFuncs.get_attrs_by_dev(
+                "LABEL", str(self._active_slot_dev)
             )
-            if not (
-                _active_slot := CMDHelperFuncs.get_fslabel_by_dev(self._active_slot_dev)
-            ):
-                raise ValueError(
-                    f"failed to get slot_id(fslabel) for active slot dev({self._active_slot_dev})"
-                )
+            assert _active_slot
             self._active_slot = _active_slot
 
             # detect standby slot
             # NOTE: using the similar logic like grub, detect the silibing dev
             #       of the active slot as standby slot
-            _parent = CMDHelperFuncs.get_parent_dev(self._active_slot_dev)
+            _parent = CMDHelperFuncs.get_parent_dev(str(self._active_slot_dev))
+            assert _parent
+
             # list children device file from parent device
             # exclude parent dev(always in the front)
             # expected raw result from lsblk:
             # NAME="/dev/sdx"
             # NAME="/dev/sdx1" # system-boot
             # NAME="/dev/sdx2" # slot_a
             # NAME="/dev/sdx3" # slot_b
-            _raw_child_partitions = CMDHelperFuncs._lsblk(f"-Pp -o NAME {_parent}")
+            _check_dev_family_cmd = ["lsblk", "-Ppo", "NAME", _parent]
+            _raw_child_partitions = subprocess_check_output(
+                _check_dev_family_cmd, raise_exception=True
+            )
+
             try:
                 # NOTE: exclude the first 2 lines(parent and system-boot)
                 _child_partitions = list(
@@ -315,38 +320,12 @@ def finalize_switching_boot(self) -> bool:
                 # reboot to the same slot to apply the new firmware
 
                 logger.info("reboot to apply new firmware...")
-                CMDHelperFuncs.reboot()
-
-                # NOTE(20230614): after calling reboot, otaclient SHOULD be terminated or exception raised
-                #                 on failed reboot command subprocess call. But if somehow it doesn't,
-                #                 it should be treated as failure and return False here.
-                return False
-
+                CMDHelperFuncs.reboot()  # this function will make otaclient exit immediately
         except Exception as e:
             _err_msg = f"failed to finalize boot switching: {e!r}"
             logger.error(_err_msg)
             return False
 
-    def prepare_standby_dev(self, *, erase_standby: bool):
-        # try umount and dev
-        if CMDHelperFuncs.is_target_mounted(self.standby_slot_dev):
-            CMDHelperFuncs.umount(self.standby_slot_dev)
-        try:
-            if erase_standby:
-                CMDHelperFuncs.mkfs_ext4(
-                    self.standby_slot_dev,
-                    fslabel=self.standby_slot,
-                )
-            else:
-                # TODO: check the standby file system status
-                #       if not erase the standby slot
-                # set the standby file system label with standby slot id
-                CMDHelperFuncs.set_dev_fslabel(self.active_slot_dev, self.standby_slot)
-        except Exception as e:
-            _err_msg = f"failed to prepare standby dev: {e!r}"
-            logger.error(_err_msg)
-            raise _RPIBootControllerError(_err_msg) from e
-
     def prepare_tryboot_txt(self):
         """Copy the standby slot's config.txt as tryboot.txt."""
         logger.debug("prepare tryboot.txt...")
@@ -364,8 +343,7 @@ def reboot_tryboot(self):
         """Reboot with tryboot flag."""
         logger.info(f"tryboot reboot to standby slot({self.standby_slot})...")
         try:
-            _cmd = "reboot '0 tryboot'"
-            subprocess_call(_cmd, raise_exception=True)
+            CMDHelperFuncs.reboot(args=["0 tryboot"])
         except Exception as e:
             _err_msg = "failed to reboot"
             logger.exception(_err_msg)
@@ -496,7 +474,10 @@ def pre_update(self, version: str, *, standby_as_ref: bool, erase_standby: bool)
             self._ota_status_control.pre_update_current()
 
             ### mount slots ###
-            self._rpiboot_control.prepare_standby_dev(erase_standby=erase_standby)
+            self._mp_control.prepare_standby_dev(
+                erase_standby=erase_standby,
+                fslabel=self._rpiboot_control.standby_slot,
+            )
             self._mp_control.mount_standby()
             self._mp_control.mount_active()
 

otaclient/app/boot_control/selecter.py

@@ -20,7 +20,6 @@
 from typing_extensions import deprecated
 
 from .configs import BootloaderType
-from ._errors import BootControlError
 from .protocol import BootControllerProtocol
 
 from ..common import read_str_from_file
@@ -82,13 +81,9 @@ def get_boot_controller(
         from ._grub import GrubController
 
         return GrubController
-    if bootloader_type == BootloaderType.CBOOT:
-        from ._cboot import CBootController
-
-        return CBootController
     if bootloader_type == BootloaderType.RPI_BOOT:
         from ._rpi_boot import RPIBootController
 
         return RPIBootController
 
-    raise BootControlError from NotImplementedError(f"unsupported: {bootloader_type=}")
+    raise NotImplementedError(f"unsupported: {bootloader_type=}")