Verified Commit e12ade9b authored by Karel Koci's avatar Karel Koci 🤘
Browse files

Solidify public API with typing

This adds type hints on all public functions and methods.
parent 772939b0
import typing
from . import autorun, const
from .utils import check_exclusive_lock as _check_exclusive_lock
from .utils import daemonize as _daemonize
......@@ -7,14 +8,13 @@ from ._supervisor import run as _run
from .prerun import wait_for_network as _wait_for_network
def opkg_lock():
"""Returns True if opkg lock is taken. It can be taken by any other
process. It doesn't have to be updater.
def opkg_lock() -> bool:
"""Returns True if opkg lock is taken. It can be taken by any other process. It doesn't have to be updater.
"""
return _check_exclusive_lock(const.OPKG_LOCK, False)
def updater_supervised():
def updater_supervised() -> bool:
"""This returns True if there is running updater-supervisor instance.
(Running means as a running process not as a library in some other process)
"""
......@@ -22,16 +22,14 @@ def updater_supervised():
return _pid_locked()
def run(wait_for_network=False, ensure_run=False, timeout=const.PKGUPDATE_TIMEOUT,
timeout_kill=const.PKGUPDATE_TIMEOUT_KILL, hooklist=None):
def run(wait_for_network: bool = False, ensure_run: bool = False, timeout: int = const.PKGUPDATE_TIMEOUT,
timeout_kill: int = const.PKGUPDATE_TIMEOUT_KILL,
hooklist: typing.Union[None, typing.Iterable[str]] = None):
"""Run updater.
This call will spawn daemon process and returns. But be aware that at first
it checks if some other supervisor is not running and it takes file lock
because of that. If someone messed up that lock then it won't return
immediately. Calling this with timeout is advised for time sensitive
applications.
If there is already running daemon then it just sends signal to it and
exits.
This call will spawn daemon process and returns. But be aware that at first it checks if some other supervisor is
not running and it takes file lock because of that. If someone messed up that lock then it won't return immediately.
Calling this with timeout is advised for time sensitive applications.
If there is already running daemon then it just sends signal to it and exits.
You can pass hooks (single line shell scripts) to be run after updater.
"""
if not autorun.enabled():
......
......@@ -17,7 +17,7 @@ BOARD_MAP = {
__board = None
def board():
def board() -> str:
"""Returns board name as expected by updater components of current board host.
"""
global __board
......
import os
import time
import typing
from . import const, autorun, notify
from .utils import report
from .exceptions import UpdaterApproveInvalidError
# TODO do we want to have list of packages that are auto approved?
# This would be beneficial for packages such as base-files that are updated
# very often but do not change at all.
def current():
PlannedPackage = typing.Dict[str, str]
ApprovalRequest = typing.Dict[str, typing.Union[str, int, typing.List[PlannedPackage], bool]]
# We can't use TypedDict as it is available since Python 3.8 but we are still using Python 3.7. TypedDict implementation
# is kept here for future replacement.
#
#class PlannedPackage(typing.TypedDict):
# name: str
# op: str
# cur_ver: str
# new_ver: str
#
#class ApprovalRequest(typing.TypedDict):
# hash: str
# status: str
# time: int
# plan: typing.List[PlannedPackage]
# reboot: bool
def current() -> ApprovalRequest:
"""Returns currently existing aprroval request. If there is no approval
request pending then it returns None.
......@@ -101,7 +117,7 @@ def _set_stat(status, hsh):
file.write(' '.join(cols))
def approve(hsh):
def approve(hsh: str):
"""Approve current plan. Passed hash should match with hash returned from
current(). If it doesn't match then UpdaterApproveInvalidError is
thrown. You can pass None to skip this check.
......@@ -109,7 +125,7 @@ def approve(hsh):
_set_stat('granted', hsh)
def deny(hsh):
def deny(hsh: str):
"""Deny current plan. This makes it effectively never timeout
(automatically installed). Passed hash should be same as the one returned
from current(). If it doesn't match then UpdaterApproveInvalidError is
......
import typing
from euci import EUci
def get_os_branch_or_version():
def get_os_branch_or_version() -> typing.Tuple[str, str]:
"""Get OS branch or version from uci."""
with EUci() as uci:
mode = uci.get("updater", "turris", "mode", dtype=str, default="branch")
......
......@@ -3,6 +3,7 @@ import sys
import fcntl
import errno
import subprocess
import typing
from threading import Thread
from .utils import report
from ._pidlock import pid_locked
......@@ -33,7 +34,7 @@ def __run_command(command):
report('Command failed with exit code: ' + str(exit_code))
def register(command):
def register(command: str):
"""Add given command (format is expected to be same as if you call
subprocess.run) to be executed when updater exits. Note that this hook is
executed no matter if updater passed or failed or even if it just requested
......@@ -74,7 +75,7 @@ def register(command):
report('Postrun hook registered: ' + command)
def register_list(commands):
def register_list(commands: typing.Iterable[str]):
"""Same as register but it allows multiple commands to be registered at
once.
"""
......
......@@ -7,13 +7,31 @@ from .const import PKGLISTS_FILE, PKGLISTS_LABELS_FILE
from .exceptions import UpdaterNoSuchListError, UpdaterNoSuchListOptionError
from . import _board
__PKGLIST_ENTRIES_LABELS = typing.Dict[str, str]
__PKGLIST_ENTRIES_OPTIONS = typing.Dict[str, typing.Union[str, bool, __PKGLIST_ENTRIES_LABELS]]
__PKGLIST_ENTRIES = typing.Dict[
str, typing.Union[
str, bool, __PKGLIST_ENTRIES_OPTIONS, __PKGLIST_ENTRIES_LABELS
]
]
PkgListLabel = typing.Dict[str, str]
PkgListOption = typing.Dict[str, typing.Union[bool, str, None, PkgListLabel]]
PkgListEntry = typing.Dict[str, typing.Union[bool, str, None, PkgListOption, PkgListLabel]]
# We can't use TypedDict as it is available since Python 3.8 but we are still using Python 3.7. TypedDict implementation
# is kept here for future replacement.
#
#class PkgListLabel(typing.TypeDict):
# title: str
# desription: str
# severity: str
#
#class PkgListOption(typing.TypeDict):
# enabled: bool
# title: str
# description: str
# url: typing.Optional[str]
# labels: typing.Dict[str, PkgListLabel]
#
#class PkgListEntry(typing.TypeDict):
# enabled: bool
# title: str
# description: str
# url: typing.Optional[str]
# options: typing.Dict[str, PkgListOption]
# labels: typing.Dict[str, PkgListLabel]
def _load_json_dict(file_path):
......@@ -50,7 +68,7 @@ def __options(pkglist_name, trans, uci, known_labels, options):
}
def pkglists(lang=None) -> typing.Dict[str, __PKGLIST_ENTRIES]:
def pkglists(lang=None) -> typing.Dict[str, PkgListEntry]:
"""Returns dict of pkglists.
Argument lang is expected to be a string containing language code. This code is then used for gettext translations
of titles and descriptions of messages.
......@@ -106,8 +124,8 @@ def pkglists(lang=None) -> typing.Dict[str, __PKGLIST_ENTRIES]:
def update_pkglists(lists: typing.Dict[str, typing.Dict[str, bool]]):
"""
Lists is expected to be nested dictionary consisting of pklist names to be enabled
and sub-dictionary with their options.
Lists is expected to be nested dictionary consisting of pklist names to be enabled and sub-dictionary with their
options.
Anything omitted will be disabled.
"""
known_lists = _load_json_dict(PKGLISTS_FILE)
......
......@@ -19,7 +19,7 @@ def clear_logs():
os.remove(PKGUPDATE_CRASH_LOG)
def failure(exit_code, trace):
def failure(exit_code: int, trace: str):
"""Send notification about updater's failure
"""
if exit_code == 0 and not os.path.isfile(PKGUPDATE_ERROR_LOG):
......@@ -103,8 +103,7 @@ def approval():
def notifier():
"""This just calls notifier. It processes new notification and sends them
together.
"""This just calls notifier. It processes new notification and sends them together.
"""
if subprocess.call(['notifier']) != 0:
report('Notifier failed')
......@@ -11,7 +11,7 @@ from .const import PING_ADDRESS
from .utils import report
def random_sleep(max_seconds):
def random_sleep(max_seconds: int):
"Sleep random amount of seconds with maximum of max_seconds"
if max_seconds is None or max_seconds <= 0:
return # No sleep at all
......@@ -20,7 +20,7 @@ def random_sleep(max_seconds):
report("Suspending updater start for " + str(suspend) + " seconds")
time.sleep(suspend)
def ping(address=PING_ADDRESS, count=1, deadline=1):
def ping(address: str = PING_ADDRESS, count: int = 1, deadline: int = 1) -> bool:
"""Ping address with given amount of pings and deadline.
Returns True on success and False if ping fails.
"""
......@@ -30,9 +30,9 @@ def ping(address=PING_ADDRESS, count=1, deadline=1):
stdin=devnull,
stdout=devnull,
stderr=devnull
)
) == 0
def wait_for_network(max_stall):
def wait_for_network(max_stall: int) -> bool:
"""This tries to connect to repo.turris.cz to check if we can access it and
otherwise it stalls execution for given maximum number of seconds.
......@@ -41,9 +41,9 @@ def wait_for_network(max_stall):
def network_test():
"Run network test (expected to be run as subprocess)"
if ping():
report("Waiting for network connection")
while ping():
if not ping():
utils.report("Waiting for network connection")
while not ping():
pass
if max_stall is None:
......
......@@ -10,9 +10,10 @@ import resource
import signal
import traceback
import syslog
import typing
def report(msg):
def report(msg: str):
"""Report message to syslog and to terminal.
"""
if sys.stderr.isatty():
......@@ -22,7 +23,7 @@ def report(msg):
syslog.syslog(msg)
def setup_alarm(func, timeout):
def setup_alarm(func: typing.Callable, timeout: int):
"This is simple alarm setup function with possibility of None timeout"
if timeout is None:
return
......@@ -30,7 +31,7 @@ def setup_alarm(func, timeout):
signal.alarm(timeout)
def check_exclusive_lock(path, isflock=False):
def check_exclusive_lock(path: str, isflock: bool = False) -> bool:
"""This returns True if someone holds exclusive lock on given path.
Otherwise it returns False.
"""
......@@ -57,7 +58,7 @@ def check_exclusive_lock(path, isflock=False):
return False
def daemonize():
def daemonize() -> bool:
"""Fork to daemon. It returns True for parent process and False for child
process.
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment