:param str cmd: command to run
:param bool 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: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:param int timeout: amount of seconds to wait for the program to run
:returns: command result output
:rtype: str
:param str process: process to verify if running
:param int timeout: run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:raises: :py:class:`RuntimeError` if process is not running
"""
platform_str = ""
Accept the Intra2net license.
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
This is mostly useful for simplified webpage access.
"""
:param wait_online: whether to wait until online
:type wait_online: bool
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
.. seealso:: :py:func:`go_offline`, :py:func:`wait_for_online`
"""
: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: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
.. seealso:: :py:func:`go_online`, :py:func:`wait_for_offline`
"""
:param int timeout: maximum timeout for waiting
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
_wait_for_online_status('offline', None, timeout, vm)
:type provider_id: int
:param int timeout: maximum timeout for waiting
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
_wait_for_online_status('online', provider_id, timeout, vm)
Disable virscan that could block GENERATE and thus all configurations.
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
log.info("Disabling virus database update")
unset_cnf("VIRSCAN_UPDATE_CRON", vm=vm)
Transfer all the emails using the guest tool arnied_helper.
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
cmd = f"{BIN_ARNIED_HELPER} --transfer-mail"
result = run_cmd(cmd=cmd, vm=vm)
:param int timeout: email transfer timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
for i in range(timeout):
if i % 10 == 0:
:param int exec_time: scheduled time of program's execution
:param str optional_args: optional command line arguments
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
log.info("Scheduling %s to be executed at %i", program, exec_time)
schedule_dir = "/var/intranator/schedule"
:param int timeout: program run timeout
:param int 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: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
log.info("Waiting for program %s to finish with timeout %i",
program, timeout)
:param int timeout: maximum number of seconds to wait
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
cmd = f"{BIN_ARNIED_HELPER} --wait-for-arnied-socket " \
f"--wait-for-arnied-socket-timeout {timeout}"
:param bool compact: whether to retrieve compact version of the matched cnf keys
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:returns: extracted information via the regex
:rtype: Match object
:param str value: cnf value of the cnf key
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:returns: the cnf id or -1 if no such cnf variable
:rtype: int
"""
:param str data: "data" field by which the resulting CNF_VAR list should be filtered
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:returns: the resulting "cnfvar" structure or None if the lookup fails or the result could not be parsed
:rtype: cnfvar option
"""
:param str data: "data" field by which the resulting CNF_VAR list should be filtered
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:returns: the cnf id or -1 if no such cnf variable
:rtype: int
"""
:param int instance: "instance" of that variable to unset
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
"""
wait_for_arnied(timeout=timeout, vm=vm)
:param str kind: "json" or "cnf"
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:raises: :py:class:`ConfigError` if cannot apply file
The config files must be provided and are always expected to be found on
:param str kind: "json" or "cnf"
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
The config files must be provided and are always expected to be found on
the host. If these are absolute paths, they will be kept as is or
:param str kind: "json", "cnf", or "raw"
:param int timeout: arnied run verification timeout
:param vm: vm to run on if running on a guest instead of the host
- :type vm: VM object or None
+ :type vm: :py:class:`virttest.qemu_vm.VM` or None
:raises: :py:class:`ValueError` if `kind` is not an acceptable value
:raises: :py:class:`ConfigError` if cannot apply file
class DummyCmdOutputMapping(object):
+ """
+ Class to replace the :py:function:`arnied_wrapper.run_cmd` function.
+ In the arnied_wrapper, when running a command, instead of calling the actual
+ function, this class' constructor will be invoked and an instance returned
+ instead. It stubs :py:class:`subprocess.CompletedProcess`, such that the
+ returncode and stdout attributes are set depending on the command or on the
+ test.
+ """
+
+ # whether to return 1 as a fail indicator
fail_switch = False
+ # mapping between expected commands and their mocked output + return code
cmds = [
{"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0},
asserted_cmds = []
def __init__(self, cmd="", ignore_errors=False, vm=None, timeout=60):
+ """
+ Class constructor to mimic the run function of the arnied wrapper.
+
+ Arguments are the same of the mocked function.
+ """
self.returncode = 0
self.stdout = b""
self._get_result(cmd)
self.returncode = 1
def __str__(self):
+ """String representation of this class."""
return "status %i, stdout %s" % (self.returncode, self.stdout)
def _get_result(self, cmd):
+ """
+ Return the first (or raise) values in the mapping matching the command.
+
+ :param str cmd: command to check the mapping against
+ :raises: :py:class:`ValueError` if the command has no corresponding
+ mapping
+ :returns: this instance with the return code and stdout attributes set
+ :rtype: :py:class:`DummyCmdOutputMapping`
+ """
for dummy_cmd in self.asserted_cmds:
if dummy_cmd['cmd'] == cmd:
self.returncode = dummy_cmd['returncode']
"for the universe" % cmd)
-@mock.patch('src.arnied_wrapper.run_cmd', DummyCmdOutputMapping)
+# make sure that invoking `run_cmd` returns an instance of DummyCmdOutputMapping
+@mock.patch("src.arnied_wrapper.run_cmd", DummyCmdOutputMapping)
class ArniedWrapperTest(unittest.TestCase):
-
def setUp(self):
DummyCmdOutputMapping.fail_switch = False
DummyCmdOutputMapping.asserted_cmds = []
self.cmd_db = DummyCmdOutputMapping.cmds
def test_verify_running(self):
+ """Test checking for running programs."""
DummyCmdOutputMapping.asserted_cmds = self.cmd_db[0:1]
arnied_wrapper.verify_running(timeout=1)
DummyCmdOutputMapping.fail_switch = True
arnied_wrapper.verify_running(timeout=1)
def test_wait_for_arnied(self):
+ """Test waiting for arnied to be ready."""
DummyCmdOutputMapping.asserted_cmds = self.cmd_db
arnied_wrapper.wait_for_arnied(timeout=10)
def test_accept_license(self):
+ """Test accepting license."""
DummyCmdOutputMapping.asserted_cmds = self.cmd_db[1:3]
arnied_wrapper.accept_licence()
# make sure an error is ignored since license might
arnied_wrapper.go_online(1)
def test_disable_virscan(self):
+ """Test disabling the virus scanner."""
DummyCmdOutputMapping.asserted_cmds = self.cmd_db[6:]
arnied_wrapper.disable_virscan()
def test_email_transfer(self):
+ """Test e-mail transferring."""
DummyCmdOutputMapping.asserted_cmds = self.cmd_db[14:15]
arnied_wrapper.email_transfer()