[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

otaclient/_utils/__init__.py

@@ -29,9 +29,11 @@ def copy_callable_typehint(_source: Callable[P, Any]):
     """This helper function return a decorator that can type hint the target
     function as the _source function.
 
-    At runtime, this decorator actually does nothing, but just return the input function as it.
-    But the returned function will have the same type hint as the source function in ide.
-    It will not impact the runtime behavior of the decorated function.
+    At runtime, this decorator actually does nothing, but just return
+    the input function as it. But the returned function will have the
+    same type hint as the source function in ide. It will not impact the
+    runtime behavior of the decorated function.
+
     """
 
     def _decorator(target) -> Callable[P, Any]:
@@ -76,6 +78,7 @@ def replace_root(path: str | Path, old_root: str | Path, new_root: str | Path) -
 
     For example, if path="/abc", old_root="/", new_root="/new_root",
     then we will have "/new_root/abc".
+
     """
     # normalize all the input args
     path = os.path.normpath(path)

otaclient/_utils/linux.py

@@ -41,6 +41,7 @@ def create_swapfile(
     Raises:
         ValueError on file already exists at <swapfile_fpath>, SubprocessCallFailed
             on failed swapfile creation.
+
     """
     swapfile_fpath = Path(swapfile_fpath)
     if swapfile_fpath.exists():
@@ -86,6 +87,7 @@ class ParsedPasswd:
     Attrs:
         _by_name (dict[str, int]): name:uid mapping.
         _by_uid (dict[int, str]): uid:name mapping.
+
     """
 
     __slots__ = ["_by_name", "_by_uid"]
@@ -112,6 +114,7 @@ class ParsedGroup:
     Attrs:
         _by_name (dict[str, int]): name:gid mapping.
         _by_gid (dict[int, str]): gid:name mapping.
+
     """
 
     __slots__ = ["_by_name", "_by_gid"]
@@ -133,6 +136,7 @@ def map_uid_by_pwnam(*, src_db: ParsedPasswd, dst_db: ParsedPasswd, uid: int) ->
 
     Raises:
         ValueError on failed mapping.
+
     """
     try:
         return dst_db._by_name[src_db._by_uid[uid]]
@@ -145,6 +149,7 @@ def map_gid_by_grpnam(*, src_db: ParsedGroup, dst_db: ParsedGroup, gid: int) ->
 
     Raises:
         ValueError on failed mapping.
+
     """
     try:
         return dst_db._by_name[src_db._by_gid[gid]]

otaclient/_utils/typing.py

@@ -33,11 +33,12 @@
 
 
 def gen_strenum_validator(enum_type: type[T]) -> Callable[[T | str | Any], T]:
-    """A before validator generator that converts input value into enum
-    before passing it to pydantic validator.
+    """A before validator generator that converts input value into enum before
+    passing it to pydantic validator.
 
     NOTE(20240129): as upto pydantic v2.5.3, (str, Enum) field cannot
-                    pass strict validation if input is str.
+    pass strict validation if input is str.
+
     """
 
     def _inner(value: T | str | Any) -> T:

otaclient/app/boot_control/_common.py

@@ -42,9 +42,10 @@
 class CMDHelperFuncs:
     """HelperFuncs bundle for wrapped linux cmd.
 
-    When underlying subprocess call failed and <raise_exception> is True,
-        functions defined in this class will raise the original exception
-        to the upper caller.
+    When underlying subprocess call failed and <raise_exception> is
+    True,     functions defined in this class will raise the original
+    exception     to the upper caller.
+
     """
 
     @classmethod
@@ -64,6 +65,7 @@ def get_attrs_by_dev(
 
         Returns:
             str: <attr> of <dev>.
+
         """
         cmd = ["lsblk", "-ino", attr, str(dev)]
         return subprocess_check_output(cmd, raise_exception=raise_exception)
@@ -86,6 +88,7 @@ def get_dev_by_token(
         Returns:
             Optional[list[str]]: If there is at least one device found, return a list
                 contains all found device(s), otherwise None.
+
         """
         cmd = ["blkid", "-o", "device", "-t", f"{token}={value}"]
         if res := subprocess_check_output(cmd, raise_exception=raise_exception):
@@ -104,6 +107,7 @@ def get_current_rootfs_dev(cls, *, raise_exception: bool = True) -> str:
 
         Returns:
             str: the devpath of current rootfs device.
+
         """
         cmd = ["findmnt", "-nfco", "SOURCE", cfg.ACTIVE_ROOTFS_PATH]
         return subprocess_check_output(cmd, raise_exception=raise_exception)
@@ -125,6 +129,7 @@ def get_mount_point_by_dev(cls, dev: str, *, raise_exception: bool = True) -> st
         Returns:
             str: the FIRST mountpint of the <dev>, or empty string if <raise_exception> is False
                 and the subprocess call failed(due to dev is not mounted or other reasons).
+
         """
         cmd = ["findmnt", "-nfo", "TARGET", dev]
         return subprocess_check_output(cmd, raise_exception=raise_exception)
@@ -145,6 +150,7 @@ def get_dev_by_mount_point(
 
         Returns:
             str: the source device of <mount_point>.
+
         """
         cmd = ["findmnt", "-no", "SOURCE", mount_point]
         return subprocess_check_output(cmd, raise_exception=raise_exception)
@@ -153,7 +159,8 @@ def get_dev_by_mount_point(
     def is_target_mounted(
         cls, target: Path | str, *, raise_exception: bool = True
     ) -> bool:
-        """Check if <target> is mounted or not. <target> can be a dev or a mount point.
+        """Check if <target> is mounted or not. <target> can be a dev or a
+        mount point.
 
         This is implemented by calling:
             findmnt <target>
@@ -165,6 +172,7 @@ def is_target_mounted(
 
         Returns:
             bool: return True if the target has at least one mount_point.
+
         """
         cmd = ["findmnt", target]
         return bool(subprocess_check_output(cmd, raise_exception=raise_exception))
@@ -185,6 +193,7 @@ def get_parent_dev(cls, child_device: str, *, raise_exception: bool = True) -> s
 
         Returns:
             str: the parent device of the specific <child_device>.
+
         """
         cmd = ["lsblk", "-idpno", "PKNAME", child_device]
         return subprocess_check_output(cmd, raise_exception=raise_exception)
@@ -201,6 +210,7 @@ def set_ext4_fslabel(cls, dev: str, fslabel: str, *, raise_exception: bool = Tru
             fslabel (str): the fslabel to be set.
             raise_exception (bool, optional): raise exception on subprocess call failed.
                 Defaults to True.
+
         """
         cmd = ["e2label", dev, fslabel]
         subprocess_call(cmd, raise_exception=raise_exception)
@@ -222,6 +232,7 @@ def mount_rw(
             mount_point (Path | str): mount point to mount to.
             raise_exception (bool, optional): raise exception on subprocess call failed.
                 Defaults to True.
+
         """
         # fmt: off
         cmd = [
@@ -248,6 +259,7 @@ def bind_mount_ro(
             mount_point (Path | str): mount point to mount to.
             raise_exception (bool, optional): raise exception on subprocess call failed.
                 Defaults to True.
+
         """
         # fmt: off
         cmd = [
@@ -274,6 +286,7 @@ def umount(cls, target: Path | str, *, raise_exception: bool = True):
             target (Path | str): target to be umounted.
             raise_exception (bool, optional): raise exception on subprocess call failed.
                 Defaults to True.
+
         """
         # first try to check whether the target(either a mount point or a dev)
         # is mounted
@@ -293,8 +306,8 @@ def mkfs_ext4(
         fsuuid: Optional[str] = None,
         raise_exception: bool = True,
     ):
-        """Create new ext4 formatted filesystem on <dev>, optionally with <fslabel>
-            and/or <fsuuid>.
+        """Create new ext4 formatted filesystem on <dev>, optionally with
+        <fslabel> and/or <fsuuid>.
 
         Args:
             dev (str): device to be formatted to ext4.
@@ -304,6 +317,7 @@ def mkfs_ext4(
                 When it is None, this function will try to preserve the previous fsuuid.
             raise_exception (bool, optional): raise exception on subprocess call failed.
                 Defaults to True.
+
         """
         cmd = ["mkfs.ext4", "-F"]
 
@@ -347,6 +361,7 @@ def mount_ro(
             mount_point (str | Path): mount point to mount to.
             raise_exception (bool, optional): raise exception on subprocess call failed.
                 Defaults to True.
+
         """
         # NOTE: set raise_exception to false to allow not mounted
         #       not mounted dev will have empty return str
@@ -384,6 +399,7 @@ def reboot(cls, args: Optional[list[str]] = None) -> NoReturn:
         Args:
             args (Optional[list[str]], optional): args passed to reboot command.
                 Defaults to None, not passing any args.
+
         """
         cmd = ["reboot"]
         if args:
@@ -410,6 +426,7 @@ class OTAStatusFilesControl:
 
     If status or slot_in_use file is missing, initialization is required.
     status will be set to INITIALIZED, and slot_in_use will be set to current slot.
+
     """
 
     def __init__(
@@ -570,7 +587,8 @@ def _store_standby_version(self, _version: str):
     # helper methods
 
     def _is_switching_boot(self, active_slot: str) -> bool:
-        """Detect whether we should switch boot or not with ota_status files."""
+        """Detect whether we should switch boot or not with ota_status
+        files."""
         # evidence: ota_status
         _is_updating_or_rollbacking = self._load_current_status() in [
             wrapper.StatusOta.UPDATING,
@@ -592,16 +610,17 @@ def _is_switching_boot(self, active_slot: str) -> bool:
     # boot control used methods
 
     def pre_update_current(self):
-        """On pre_update stage, set current slot's status to FAILURE
-        and set slot_in_use to standby slot."""
+        """On pre_update stage, set current slot's status to FAILURE and set
+        slot_in_use to standby slot."""
         self._store_current_status(wrapper.StatusOta.FAILURE)
         self._store_current_slot_in_use(self.standby_slot)
 
     def pre_update_standby(self, *, version: str):
-        """On pre_update stage, set standby slot's status to UPDATING,
-        set slot_in_use to standby slot, and set version.
+        """On pre_update stage, set standby slot's status to UPDATING, set
+        slot_in_use to standby slot, and set version.
 
         NOTE: expecting standby slot to be mounted and ready for use!
+
         """
         # create the ota-status folder unconditionally
         self.standby_ota_status_dir.mkdir(exist_ok=True, parents=True)
@@ -640,6 +659,7 @@ def booted_ota_status(self) -> wrapper.StatusOta:
 
         This property is only meant to be used once when otaclient starts up,
         switch to use live_ota_status by otaclient after otaclient is running.
+
         """
         return self._ota_status
 
@@ -696,8 +716,9 @@ def mount_active(self) -> None:
     def preserve_ota_folder_to_standby(self):
         """Copy the /boot/ota folder to standby slot to preserve it.
 
-        /boot/ota folder contains the ota setting for this device,
-        so we should preserve it for each slot, accross each update.
+        /boot/ota folder contains the ota setting for this device, so we
+        should preserve it for each slot, accross each update.
+
         """
         logger.debug("copy /boot/ota from active to standby.")
         try: