Skip to content

Commit

Permalink
boot_control.common: update a lot of docstrings
Browse files Browse the repository at this point in the history
  • Loading branch information
Bodong-Yang committed Apr 25, 2024
1 parent afcbf33 commit 192e36b
Showing 1 changed file with 103 additions and 41 deletions.
144 changes: 103 additions & 41 deletions otaclient/app/boot_control/_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,12 @@


class CMDHelperFuncs:
"""HelperFuncs bundle for wrapped linux cmd."""
"""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.
"""

@classmethod
def get_attrs_by_dev(
Expand All @@ -61,9 +66,6 @@ def get_attrs_by_dev(
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
Raises:
subprocess.CalledProcessError on failed command call.
Returns:
str: <attr> of <dev>.
"""
Expand All @@ -85,9 +87,6 @@ def get_dev_by_token(
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
Raises:
subprocess.CalledProcessError on failed command call.
Returns:
Optional[list[str]]: If there is at least one device found, return a list
contains all found device(s), otherwise None.
Expand All @@ -107,11 +106,8 @@ def get_current_rootfs_dev(cls, *, raise_exception: bool = True) -> str:
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
Raises:
subprocess.CalledProcessError on failed command call.
Returns:
str: _description_
str: the devpath of current rootfs device.
"""
cmd = ["findmnt", "-nfco", "SOURCE", cfg.ACTIVE_ROOTFS_PATH]
return subprocess_check_output(cmd, raise_exception=raise_exception)
Expand All @@ -123,8 +119,16 @@ def get_mount_point_by_dev(cls, dev: str, *, raise_exception: bool = True) -> st
This is implemented by calling:
findmnt <dev> -nfo TARGET <dev>
NOTE: findmnt raw result might have multiple lines if target dev is bind mounted.
use option -f to only show the first file system.
NOTE: option -f is used to only show the first file system.
Args:
dev (str): the device to check against.
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
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)
Expand All @@ -133,10 +137,18 @@ def get_mount_point_by_dev(cls, dev: str, *, raise_exception: bool = True) -> st
def get_dev_by_mount_point(
cls, mount_point: str, *, raise_exception: bool = True
) -> str:
"""Return the underlying mounted dev of the given mount_point.
"""Return the source dev of the given <mount_point>.
This is implemented by calling:
findmnt -no SOURCE <mount_point>
Args:
mount_point (str): mount_point to check against.
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
Returns:
str: the source device of <mount_point>.
"""
cmd = ["findmnt", "-no", "SOURCE", mount_point]
return subprocess_check_output(cmd, raise_exception=raise_exception)
Expand All @@ -150,41 +162,70 @@ def is_target_mounted(
This is implemented by calling:
findmnt <target>
and see if there is any matching device.
Args:
target (Path | str): the target to check against. Could be a device or a mount point.
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
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))

@classmethod
def get_parent_dev(cls, child_device: str, *, raise_exception: bool = True) -> str:
"""Get the parent devpath from <child_device>.
When `/dev/nvme0n1p1` is specified as child_device, /dev/nvme0n1 is returned.
This function is implemented by calling:
lsblk -idpno PKNAME <child_device>
Args:
child_device (str): the device to find parent device from.
raise_exception (bool, optional): raise exception on subprocess call failed.
Defaults to True.
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)

@classmethod
def set_ext4_fslabel(cls, dev: str, fslabel: str, *, raise_exception: bool = True):
"""Set <fslabel> to ext4 formatted <dev>.
This is implemented by calling:
e2label <dev> <fslabel>
Args:
dev (str): the ext4 partition device.
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)

@classmethod
def mount_rw(
cls, target: str, mount_point: Path | str, *, raise_exception: bool = True
):
"""Mount the target to the mount_point read-write.
"""Mount the <target> to <mount_point> read-write.
This is implemented by calling:
mount -o rw --make-private --make-unbindable <target> <mount_point>
NOTE: pass args = ["--make-private", "--make-unbindable"] to prevent
mount events propagation to/from this mount point.
Raises:
CalledProcessError on failed mount.
Args:
target (str): target to be mounted.
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 = [
Expand All @@ -201,13 +242,16 @@ def mount_rw(
def bind_mount_ro(
cls, target: str, mount_point: Path | str, *, raise_exception: bool = True
):
"""Bind mount the target to the mount_point read-only.
"""Bind mount the <target> to <mount_point> read-only.
This is implemented by calling:
mount -o bind,ro --make-private --make-unbindable <target> <mount_point>
Raises:
CalledProcessError on failed mount.
Args:
target (str): target to be mounted.
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 = [
Expand All @@ -222,9 +266,18 @@ def bind_mount_ro(

@classmethod
def umount(cls, target: Path | str, *, raise_exception: bool = True):
"""Try to unmount the <target>.
"""Try to umount the <target>.
This function will first check whether the <target> is mounted.
This is implemented by calling:
umount <target>
Before calling umount, the <target> will be check whether it is mounted,
if it is not mounted, this function will return directly.
Args:
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
Expand All @@ -244,15 +297,17 @@ def mkfs_ext4(
fsuuid: Optional[str] = None,
raise_exception: bool = True,
):
"""Call mkfs.ext4 on <dev>.
If fslabel and/or fsuuid provided, use them in prior, otherwise
try to preserve fsuuid and fslabel if possible.
NOTE: -F is used to bypass interactive confirmation.
"""Create new ext4 formatted filesystem on <dev>, optionally with <fslabel>
and/or <fsuuid>.
Raises:
Original CalledProcessError from the failed command execution.
Args:
dev (str): device to be formatted to ext4.
fslabel (Optional[str], optional): fslabel of the new ext4 filesystem. Defaults to None.
When it is None, this function will try to preserve the previous fslabel.
fsuuid (Optional[str], optional): fsuuid of the new ext4 filesystem. Defaults to None.
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"]

Expand All @@ -279,23 +334,23 @@ def mkfs_ext4(
cmd.extend(["-L", fslabel])

cmd.append(dev)
logger.warning(f"execute {cmd=}")
logger.warning(f"format {dev} to ext4: {cmd=}")
subprocess_call(cmd, raise_exception=raise_exception)

@classmethod
def mount_ro(
cls, *, target: str, mount_point: str | Path, raise_exception: bool = True
):
"""Mount target on mount_point read-only.
"""Mount <target> to <mount_point> read-only.
If the target device is mounted, we bind mount the target device to mount_point,
If the target device is mounted, we bind mount the target device to mount_point.
if the target device is not mounted, we directly mount it to the mount_point.
This method mount the target as ro with make-private flag and make-unbindable flag,
to prevent ANY accidental writes/changes to the target.
Raises:
CalledProcessError on failed mounting.
Args:
target (str): target to be mounted.
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
Expand All @@ -322,10 +377,17 @@ def mount_ro(

@classmethod
def reboot(cls, args: Optional[list[str]] = None) -> NoReturn:
"""Reboot the whole system otaclient running at and terminate otaclient.
"""Reboot the system, with optional args passed to reboot command.
NOTE(20230614): this command MUST also make otaclient exit immediately.
This is implemented by calling:
reboot [args[0], args[1], ...]
NOTE(20230614): this command makes otaclient exit immediately.
NOTE(20240421): rpi_boot's reboot takes args.
Args:
args (Optional[list[str]], optional): args passed to reboot command.
Defaults to None, not passing any args.
"""
cmd = ["reboot"]
if args:
Expand Down

0 comments on commit 192e36b

Please sign in to comment.