Merge branch 'main' into refactor/otaclient_config

.github/tools/gen_requirements_txt.py

@@ -4,7 +4,7 @@
 import sys
 from typing import Any
 
-import tomllib
+import tomllib  # type: ignore[import]
 
 
 def gen_requirements_txt(pyproject_cfg: dict[str, Any]) -> str:

src/otaclient/app/main.py

@@ -33,7 +33,7 @@
 from otaclient.log_setting import configure_logging
 from otaclient_api.v2 import otaclient_v2_pb2_grpc as v2_grpc
 from otaclient_api.v2.api_stub import OtaClientServiceV2
-from otaclient_common.common import read_str_from_file, write_str_to_file_sync
+from otaclient_common._io import read_str_from_file, write_str_to_file_atomic
 
 # configure logging before any code being executed
 configure_logging()
@@ -43,7 +43,7 @@
 def _check_other_otaclient():
     """Check if there is another otaclient instance running."""
     # create a lock file to prevent multiple ota-client instances start
-    if pid := read_str_from_file(cfg.OTACLIENT_PID_FILE):
+    if pid := read_str_from_file(cfg.OTACLIENT_PID_FILE, _default=""):
         # running process will have a folder under /proc
         if Path(f"/proc/{pid}").is_dir():
             logger.error(f"another instance of ota-client({pid=}) is running, abort")
@@ -56,7 +56,7 @@ def _check_other_otaclient():
     _run_dir.mkdir(parents=True, exist_ok=True)
     os.chmod(_run_dir, 0o550)
     # write our pid to the lock file
-    write_str_to_file_sync(cfg.OTACLIENT_PID_FILE, f"{os.getpid()}")
+    write_str_to_file_atomic(cfg.OTACLIENT_PID_FILE, f"{os.getpid()}")
 
 
 def create_otaclient_grpc_server():

src/otaclient/boot_control/_common.py

@@ -26,12 +26,8 @@
 
 from otaclient.app.configs import config as cfg
 from otaclient_api.v2 import types as api_types
-from otaclient_common.common import (
-    read_str_from_file,
-    subprocess_call,
-    subprocess_check_output,
-    write_str_to_file_sync,
-)
+from otaclient_common._io import read_str_from_file, write_str_to_file_atomic
+from otaclient_common.common import subprocess_call, subprocess_check_output
 from otaclient_common.typing import StrOrPath
 
 logger = logging.getLogger(__name__)
@@ -583,36 +579,36 @@ def _load_slot_in_use_file(self):
     # slot_in_use control
 
     def _store_current_slot_in_use(self, _slot: str):
-        write_str_to_file_sync(
+        write_str_to_file_atomic(
             self.current_ota_status_dir / cfg.SLOT_IN_USE_FNAME, _slot
         )
 
     def _store_standby_slot_in_use(self, _slot: str):
-        write_str_to_file_sync(
+        write_str_to_file_atomic(
             self.standby_ota_status_dir / cfg.SLOT_IN_USE_FNAME, _slot
         )
 
     def _load_current_slot_in_use(self) -> Optional[str]:
         if res := read_str_from_file(
-            self.current_ota_status_dir / cfg.SLOT_IN_USE_FNAME, default=""
+            self.current_ota_status_dir / cfg.SLOT_IN_USE_FNAME, _default=""
         ):
             return res
 
     # status control
 
     def _store_current_status(self, _status: api_types.StatusOta):
-        write_str_to_file_sync(
+        write_str_to_file_atomic(
             self.current_ota_status_dir / cfg.OTA_STATUS_FNAME, _status.name
         )
 
     def _store_standby_status(self, _status: api_types.StatusOta):
-        write_str_to_file_sync(
+        write_str_to_file_atomic(
             self.standby_ota_status_dir / cfg.OTA_STATUS_FNAME, _status.name
         )
 
     def _load_current_status(self) -> Optional[api_types.StatusOta]:
         if _status_str := read_str_from_file(
-            self.current_ota_status_dir / cfg.OTA_STATUS_FNAME
+            self.current_ota_status_dir / cfg.OTA_STATUS_FNAME, _default=""
         ).upper():
             with contextlib.suppress(KeyError):
                 # invalid status string
@@ -621,7 +617,7 @@ def _load_current_status(self) -> Optional[api_types.StatusOta]:
     # version control
 
     def _store_standby_version(self, _version: str):
-        write_str_to_file_sync(
+        write_str_to_file_atomic(
             self.standby_ota_status_dir / cfg.OTA_VERSION_FNAME,
             _version,
         )
@@ -680,8 +676,7 @@ def pre_rollback_standby(self):
     def load_active_slot_version(self) -> str:
         return read_str_from_file(
             self.current_ota_status_dir / cfg.OTA_VERSION_FNAME,
-            missing_ok=True,
-            default=cfg.DEFAULT_VERSION_STR,
+            _default=cfg.DEFAULT_VERSION_STR,
         )
 
     def on_failure(self):
@@ -794,4 +789,4 @@ def umount_all(self, *, ignore_error: bool = True):
 
 
 def cat_proc_cmdline(target: str = "/proc/cmdline") -> str:
-    return read_str_from_file(target, missing_ok=False)
+    return read_str_from_file(target, _default="")

src/otaclient/boot_control/_grub.py

@@ -44,13 +44,12 @@
 
 from otaclient.app import errors as ota_errors
 from otaclient_api.v2 import types as api_types
-from otaclient_common.common import (
-    re_symlink_atomic,
+from otaclient_common._io import (
     read_str_from_file,
-    subprocess_call,
-    subprocess_check_output,
-    write_str_to_file_sync,
+    symlink_atomic,
+    write_str_to_file_atomic,
 )
+from otaclient_common.common import subprocess_call, subprocess_check_output
 
 from ._common import (
     CMDHelperFuncs,
@@ -558,8 +557,8 @@ def _prepare_kernel_initrd_links(target_folder: Path):
         if not (kernel and initrd):
             raise ValueError(f"vmlinuz and/or initrd.img not found at {target_folder}")
 
-        re_symlink_atomic(target_folder / GrubHelper.KERNEL_OTA, kernel)
-        re_symlink_atomic(target_folder / GrubHelper.INITRD_OTA, initrd)
+        symlink_atomic(target_folder / GrubHelper.KERNEL_OTA, kernel)
+        symlink_atomic(target_folder / GrubHelper.INITRD_OTA, initrd)
         logger.info(f"finished generate ota symlinks under {target_folder}")
 
     def _grub_update_on_booted_slot(self, *, abort_on_standby_missed=True):
@@ -588,7 +587,7 @@ def _grub_update_on_booted_slot(self, *, abort_on_standby_missed=True):
         # step1: update grub_default file
         _in = grub_default_file.read_text()
         _out = GrubHelper.update_grub_default(_in)
-        write_str_to_file_sync(grub_default_file, _out)
+        write_str_to_file_atomic(grub_default_file, _out)
 
         # step2: generate grub_cfg by grub-mkconfig
         # parse the output and find the active slot boot entry idx
@@ -612,7 +611,7 @@ def _grub_update_on_booted_slot(self, *, abort_on_standby_missed=True):
             default_entry_idx=active_slot_entry_idx,
         )
         logger.debug(f"generated grub_default: {pformat(_out)}")
-        write_str_to_file_sync(grub_default_file, _out)
+        write_str_to_file_atomic(grub_default_file, _out)
 
         # step4: populate new active grub_file
         # update the ota.standby entry's rootfs uuid to standby slot's uuid
@@ -635,7 +634,7 @@ def _grub_update_on_booted_slot(self, *, abort_on_standby_missed=True):
             kernel_ver=GrubHelper.SUFFIX_OTA_STANDBY,
             rootfs_str=f"root={standby_uuid_str}",
         ):
-            write_str_to_file_sync(active_slot_grub_file, grub_cfg_updated)
+            write_str_to_file_atomic(active_slot_grub_file, grub_cfg_updated)
             logger.info(f"standby rootfs: {standby_uuid_str}")
             logger.debug(f"generated grub_cfg: {pformat(grub_cfg_updated)}")
         else:
@@ -649,10 +648,10 @@ def _grub_update_on_booted_slot(self, *, abort_on_standby_missed=True):
 
             logger.warning(msg)
             logger.info(f"generated grub_cfg: {pformat(grub_cfg_content)}")
-            write_str_to_file_sync(active_slot_grub_file, grub_cfg_content)
+            write_str_to_file_atomic(active_slot_grub_file, grub_cfg_content)
 
         # finally, point grub.cfg to active slot's grub.cfg
-        re_symlink_atomic(  # /boot/grub/grub.cfg -> ../ota-partition/grub.cfg
+        symlink_atomic(  # /boot/grub/grub.cfg -> ../ota-partition/grub.cfg
             self.grub_file,
             Path("../") / cfg.BOOT_OTA_PARTITION_FILE / "grub.cfg",
         )
@@ -662,28 +661,28 @@ def _ensure_ota_partition_symlinks(self, active_slot: str):
         """Ensure /boot/{ota_partition,vmlinuz-ota,initrd.img-ota} symlinks from
         specified <active_slot> point's of view."""
         ota_partition_folder = Path(cfg.BOOT_OTA_PARTITION_FILE)  # ota-partition
-        re_symlink_atomic(  # /boot/ota-partition -> ota-partition.<active_slot>
+        symlink_atomic(  # /boot/ota-partition -> ota-partition.<active_slot>
             self.boot_dir / ota_partition_folder,
             ota_partition_folder.with_suffix(f".{active_slot}"),
         )
-        re_symlink_atomic(  # /boot/vmlinuz-ota -> ota-partition/vmlinuz-ota
+        symlink_atomic(  # /boot/vmlinuz-ota -> ota-partition/vmlinuz-ota
             self.boot_dir / GrubHelper.KERNEL_OTA,
             ota_partition_folder / GrubHelper.KERNEL_OTA,
         )
-        re_symlink_atomic(  # /boot/initrd.img-ota -> ota-partition/initrd.img-ota
+        symlink_atomic(  # /boot/initrd.img-ota -> ota-partition/initrd.img-ota
             self.boot_dir / GrubHelper.INITRD_OTA,
             ota_partition_folder / GrubHelper.INITRD_OTA,
         )
 
     def _ensure_standby_slot_boot_files_symlinks(self, standby_slot: str):
         """Ensure boot files symlinks for specified <standby_slot>."""
         ota_partition_folder = Path(cfg.BOOT_OTA_PARTITION_FILE)  # ota-partition
-        re_symlink_atomic(  # /boot/vmlinuz-ota.standby -> ota-partition.<standby_slot>/vmlinuz-ota
+        symlink_atomic(  # /boot/vmlinuz-ota.standby -> ota-partition.<standby_slot>/vmlinuz-ota
             self.boot_dir / GrubHelper.KERNEL_OTA_STANDBY,
             ota_partition_folder.with_suffix(f".{standby_slot}")
             / GrubHelper.KERNEL_OTA,
         )
-        re_symlink_atomic(  # /boot/initrd.img-ota.standby -> ota-partition.<standby_slot>/initrd.img-ota
+        symlink_atomic(  # /boot/initrd.img-ota.standby -> ota-partition.<standby_slot>/initrd.img-ota
             self.boot_dir / GrubHelper.INITRD_OTA_STANDBY,
             ota_partition_folder.with_suffix(f".{standby_slot}")
             / GrubHelper.INITRD_OTA,
@@ -795,7 +794,7 @@ def _update_fstab(self, *, active_slot_fstab: Path, standby_slot_fstab: Path):
         )
 
         # standby partition fstab (to be merged)
-        fstab_standby = read_str_from_file(standby_slot_fstab, missing_ok=False)
+        fstab_standby = read_str_from_file(standby_slot_fstab)
         fstab_standby_dict: Dict[str, re.Match] = {}
 
         for line in fstab_standby.splitlines():
@@ -805,7 +804,7 @@ def _update_fstab(self, *, active_slot_fstab: Path, standby_slot_fstab: Path):
 
         # merge entries
         merged: List[str] = []
-        fstab_active = read_str_from_file(active_slot_fstab, missing_ok=False)
+        fstab_active = read_str_from_file(active_slot_fstab)
         for line in fstab_active.splitlines():
             if ma := fstab_entry_pa.match(line):
                 mp = ma.group("mount_point")
@@ -827,7 +826,7 @@ def _update_fstab(self, *, active_slot_fstab: Path, standby_slot_fstab: Path):
         merged.append("")  # add a new line at the end of file
 
         # write to standby fstab
-        write_str_to_file_sync(standby_slot_fstab, "\n".join(merged))
+        write_str_to_file_atomic(standby_slot_fstab, "\n".join(merged))
 
     def _cleanup_standby_ota_partition_folder(self):
         """Cleanup old files under the standby ota-partition folder."""

src/otaclient/boot_control/_jetson_cboot.py

@@ -34,7 +34,8 @@
 )
 from otaclient_api.v2 import types as api_types
 from otaclient_common import replace_root
-from otaclient_common.common import file_digest, subprocess_run_wrapper
+from otaclient_common._io import cal_file_digest
+from otaclient_common.common import subprocess_run_wrapper
 from otaclient_common.typing import StrOrPath
 
 from ._common import CMDHelperFuncs, OTAStatusFilesControl, SlotMountHelper
@@ -249,7 +250,7 @@ def firmware_update(self) -> bool:
                 logger.warning(f"{bup_fpath=} doesn't exist! skip...")
                 continue
 
-            _digest = file_digest(bup_fpath, algorithm=payload_digest_alg)
+            _digest = cal_file_digest(bup_fpath, algorithm=payload_digest_alg)
             if _digest != payload_digest_value:
                 logger.warning(
                     f"{payload_digest_alg} validation failed, "

src/otaclient/boot_control/_jetson_common.py

@@ -30,7 +30,8 @@
 from typing_extensions import Annotated, Literal, Self
 
 from otaclient_common import replace_root
-from otaclient_common.common import copytree_identical, write_str_to_file_sync
+from otaclient_common._io import write_str_to_file_atomic
+from otaclient_common.common import copytree_identical
 from otaclient_common.typing import StrOrPath
 
 from ._common import CMDHelperFuncs
@@ -299,7 +300,7 @@ def __init__(
 
     def write_to_file(self, fw_bsp_fpath: StrOrPath) -> None:
         """Write firmware_bsp_version from memory to firmware_bsp_version file."""
-        write_str_to_file_sync(fw_bsp_fpath, self._version.model_dump_json())
+        write_str_to_file_atomic(fw_bsp_fpath, self._version.model_dump_json())
 
     @property
     def current_slot_bsp_ver(self) -> BSPVersion:
@@ -459,7 +460,7 @@ def update_standby_slot_extlinux_cfg(
         )
         src = active_slot_extlinux_fpath
 
-    write_str_to_file_sync(
+    write_str_to_file_atomic(
         standby_slot_extlinux_fpath,
         update_extlinux_cfg(
             src.read_text(),

src/otaclient/boot_control/_jetson_uefi.py

@@ -42,12 +42,8 @@
 )
 from otaclient_api.v2 import types as api_types
 from otaclient_common import replace_root
-from otaclient_common.common import (
-    file_digest,
-    file_sha256,
-    subprocess_call,
-    write_str_to_file_sync,
-)
+from otaclient_common._io import cal_file_digest, file_sha256, write_str_to_file_atomic
+from otaclient_common.common import subprocess_call
 from otaclient_common.typing import StrOrPath
 
 from ._common import CMDHelperFuncs, OTAStatusFilesControl, SlotMountHelper
@@ -403,7 +399,7 @@ def _l4tlauncher_version_control(
         _ver_control = L4TLauncherBSPVersionControl(
             bsp_ver=l4tlauncher_bsp_ver, sha256_digest=l4tlauncher_sha256_digest
         )
-        write_str_to_file_sync(l4tlauncher_ver_fpath, _ver_control.dump())
+        write_str_to_file_atomic(l4tlauncher_ver_fpath, _ver_control.dump())
         return l4tlauncher_bsp_ver
 
     # NOTE(20240624): if we failed to detect the l4tlauncher's version,
@@ -418,7 +414,7 @@ def _l4tlauncher_version_control(
     _ver_control = L4TLauncherBSPVersionControl(
         bsp_ver=current_slot_bsp_ver, sha256_digest=l4tlauncher_sha256_digest
     )
-    write_str_to_file_sync(l4tlauncher_ver_fpath, _ver_control.dump())
+    write_str_to_file_atomic(l4tlauncher_ver_fpath, _ver_control.dump())
     return current_slot_bsp_ver
 
 
@@ -509,7 +505,7 @@ def _prepare_fwupdate_capsule(self) -> bool:
             )
 
             try:
-                _digest = file_digest(capsule_fpath, algorithm=capsule_digest_alg)
+                _digest = cal_file_digest(capsule_fpath, algorithm=capsule_digest_alg)
                 assert (
                     _digest == capsule_digest_value
                 ), f"{capsule_digest_alg} validation failed, expect {capsule_digest_value}, get {_digest}"
@@ -561,7 +557,9 @@ def _update_l4tlauncher(self) -> bool:
                 sha256_digest=_payload.digest.digest,
             )
             try:
-                _digest = file_digest(ota_image_bootaa64, algorithm=payload_digest_alg)
+                _digest = cal_file_digest(
+                    ota_image_bootaa64, algorithm=payload_digest_alg
+                )
                 assert (
                     _digest == payload_digest_value
                 ), f"{payload_digest_alg} validation failed, expect {payload_digest_value}, get {_digest}"
@@ -570,7 +568,7 @@ def _update_l4tlauncher(self) -> bool:
                 shutil.copy(ota_image_bootaa64, self.bootaa64_at_esp)
                 os.sync()  # ensure the boot application is written to the disk
 
-                write_str_to_file_sync(
+                write_str_to_file_atomic(
                     self.l4tlauncher_ver_fpath,
                     new_l4tlauncher_ver_ctrl.dump(),
                 )