From 103cd3143af9386e5fbca009c86a27aac3d7923c Mon Sep 17 00:00:00 2001 From: Christian Herdtweck Date: Tue, 24 Jun 2025 12:07:24 +0200 Subject: [PATCH] Add type hints in arnied_wrapper This is one of the most used modules (from QA), so raise coding standard toward QA standards. Avoid importing VM class, so leave type spec for this param in function doc string. Mention possibility of RuntimeError in a few function doc strings --- src/arnied_wrapper.py | 66 +++++++++++++++++++++++++----------------------- 1 files changed, 34 insertions(+), 32 deletions(-) diff --git a/src/arnied_wrapper.py b/src/arnied_wrapper.py index 66d5028..c9f3683 100644 --- a/src/arnied_wrapper.py +++ b/src/arnied_wrapper.py @@ -32,6 +32,7 @@ Copyright: Intra2net AG import os import time import subprocess +from subprocess import CompletedProcess import shutil import tempfile from shlex import quote @@ -44,18 +45,17 @@ log = logging.getLogger('pyi2ncommon.arnied_wrapper') BIN_ARNIED_HELPER = "/usr/intranator/bin/arnied_helper" -def run_cmd(cmd="", ignore_errors=False, vm=None, timeout=60): +def run_cmd(cmd: str = "", ignore_errors: bool = False, vm=None, timeout: int = 60) -> CompletedProcess: """ Universal command run wrapper. - :param str cmd: command to run - :param bool ignore_errors: whether not to raise error on command failure + :param cmd: command to run + :param ignore_errors: whether not to raise error on command failure :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None - :param int timeout: amount of seconds to wait for the program to run + :param timeout: amount of seconds to wait for the program to run :returns: command result output where output (stdout/stderr) is bytes (encoding dependent on environment and command given) - :rtype: :py:class:`subprocess.CompletedProcess` :raises: :py:class:`OSError` if command failed and cannot be ignored """ if vm is not None: @@ -74,12 +74,12 @@ def run_cmd(cmd="", ignore_errors=False, vm=None, timeout=60): capture_output=True) -def verify_running(process='arnied', timeout=60, vm=None): +def verify_running(process: str = "arnied", timeout: int = 60, vm=None): """ Verify if a given process is running via 'pgrep'. - :param str process: process to verify if running - :param int timeout: run verification timeout + :param process: process to verify if running + :param timeout: run verification timeout :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None :raises: :py:class:`RuntimeError` if process is not running @@ -103,17 +103,16 @@ def verify_running(process='arnied', timeout=60, vm=None): # Basic functionality -def go_online(provider_id, wait_online=True, timeout=60, vm=None): +def go_online(provider_id: int, wait_online: bool = True, timeout: int = 60, vm=None): """ Go online with the given provider id. :param provider_id: provider to go online with - :type provider_id: int :param wait_online: whether to wait until online - :type wait_online: bool - :param int timeout: Seconds to wait in :py:func:`wait_for_online` + :param timeout: Seconds to wait in :py:func:`wait_for_online` :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None + :raises: :py:class:`RuntimeError` if waiting requested but failed within timeout .. seealso:: :py:func:`go_offline`, :py:func:`wait_for_online` """ @@ -127,14 +126,14 @@ def go_online(provider_id, wait_online=True, timeout=60, vm=None): wait_for_online(provider_id, timeout=timeout, vm=vm) -def go_offline(wait_offline=True, vm=None): +def go_offline(wait_offline: bool = True, vm=None): """ Go offline. :param wait_offline: whether to wait until offline - :type wait_offline: bool :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None + :raises: :py:class:`RuntimeError` if waiting requested but failed within timeout .. seealso:: :py:func:`go_online`, :py:func:`wait_for_offline` """ @@ -149,31 +148,32 @@ def go_offline(wait_offline=True, vm=None): wait_for_offline(wait_offline, vm=vm) -def wait_for_offline(timeout=60, vm=None): +def wait_for_offline(timeout: int = 60, vm=None): """ Wait for arnied to signal we are offline. - :param int timeout: maximum timeout for waiting + :param timeout: maximum timeout for waiting :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None + :raises: :py:class:`RuntimeError` if did not go online within timeout """ _wait_for_online_status('offline', None, timeout, vm) -def wait_for_online(provider_id, timeout=60, vm=None): +def wait_for_online(provider_id: int, timeout: int = 60, vm=None): """ Wait for arnied to signal we are online. :param provider_id: provider to go online with - :type provider_id: int - :param int timeout: maximum timeout for waiting + :param timeout: maximum timeout for waiting :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None + :raises: :py:class:`RuntimeError` if did not go online within timeout """ _wait_for_online_status('online', provider_id, timeout, vm) -def _wait_for_online_status(status, provider_id, timeout, vm): +def _wait_for_online_status(status: str, provider_id: int, timeout: int, vm): # Don't use tell-connd --status here since the actual # ONLINE signal to arnied is transmitted # asynchronously via arnieclient_muxer. @@ -230,14 +230,14 @@ def email_transfer(vm=None): log.debug(result) -def wait_for_email_transfer(timeout=300, vm=None): +def wait_for_email_transfer(timeout: int = 300, vm=None): """ Wait until the mail queue is empty and all emails are sent. If the mail queue is still not empty after timeout is reached, a warning is logged and a :py:class:`TimeoutError` is raised. - :param int timeout: email transfer timeout + :param timeout: email transfer timeout :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None :raises TimeoutError: if mail transfer was not complete when timeout was reached @@ -300,13 +300,13 @@ def wait_for_quarantine_processing(vm_session: Any = None, max_wait: int = 30) - return True -def schedule(program, exec_time=0, optional_args="", vm=None): +def schedule(program: str, exec_time: int = 0, optional_args: str = "", vm=None): """ Schedule a program to be executed at a given unix time stamp. - :param str program: program whose execution is scheduled - :param int exec_time: scheduled time of program's execution - :param str optional_args: optional command line arguments + :param program: program whose execution is scheduled + :param exec_time: scheduled time of program's execution + :param optional_args: optional command line arguments :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None """ @@ -341,13 +341,13 @@ def schedule(program, exec_time=0, optional_args="", vm=None): log.debug("Moved temporary file to %s", moved_tmp_file) -def wait_for_run(program, timeout=300, retries=10, vm=None): +def wait_for_run(program: str, timeout: int = 300, retries: int = 10, vm=None): """ Wait for a program using the guest arnied_helper tool. - :param str program: scheduled or running program to wait for - :param int timeout: program run timeout - :param int retries: number of tries to verify that the program is scheduled or running + :param program: scheduled or running program to wait for + :param timeout: program run timeout + :param retries: number of tries to verify that the program is scheduled or running :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None """ @@ -372,11 +372,11 @@ def wait_for_run(program, timeout=300, retries=10, vm=None): log.debug(result.stdout) -def wait_for_arnied(timeout=60, vm=None): +def wait_for_arnied(timeout: int = 60, vm=None): """ Wait for arnied socket to be ready. - :param int timeout: maximum number of seconds to wait + :param timeout: maximum number of seconds to wait :param vm: vm to run on if running on a guest instead of the host :type vm: :py:class:`virttest.qemu_vm.VM` or None """ @@ -399,6 +399,8 @@ def wait_for_generate(timeout: int = 300, vm=None) -> bool: in those cases, `True` otherwise :param timeout: max time to wait for this function to finish + :param vm: vm to run on if running on a guest instead of the host + :type vm: :py:class:`virttest.qemu_vm.VM` or None :returns: True if no runs of generate are underway or scheduled, False if `timeout` was not enough """ -- 1.7.1