import os
import time
import subprocess
+from subprocess import CompletedProcess
import shutil
import tempfile
from shlex import quote
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:
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
# 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`
"""
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`
"""
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.
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
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
"""
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
"""
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
"""
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
"""
git://developer.intra2net.com / pyi2ncommon / commitdiff