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

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=}")