#: default set_cnf binary
BIN_SET_CNF = "/usr/intranator/bin/set_cnf"
+#: default arnied_helper binary
+BIN_ARNIED_HELPER = "/usr/intranator/bin/arnied_helper"
#: default location for template configuration files
SRC_CONFIG_DIR = "."
#: default location for dumped configuration files
: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
def verify_running(process='arnied', timeout=60, vm=None):
"""
Verify if a given process is running via 'pgrep'.
- Normally this is used to check if arnied is running.
: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.
"""
cmd = 'echo "LICENSE_ACCEPTED,0: \\"1\\"" | set_cnf'
result = run_cmd(cmd=cmd, ignore_errors=True, vm=vm)
log.debug(result)
- cmd = "/usr/intranator/bin/arnied_helper --wait-for-program-end GENERATE"
- run_cmd(cmd=cmd, vm=vm)
+ wait_for_generate(vm=vm)
def go_online(provider_id, wait_online=True, timeout=60, vm=None):
: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 = "/usr/intranator/bin/arnied_helper --transfer-mail"
+ cmd = f"{BIN_ARNIED_HELPER} --transfer-mail"
result = run_cmd(cmd=cmd, vm=vm)
log.debug(result)
: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)
for i in range(retries):
- cmd = "/usr/intranator/bin/arnied_helper --is-scheduled-or-running " \
+ cmd = f"{BIN_ARNIED_HELPER} --is-scheduled-or-running " \
+ program.upper()
check_scheduled = run_cmd(cmd=cmd, ignore_errors=True, vm=vm)
if check_scheduled.returncode == 0:
else:
log.warning("The program %s was not scheduled and is not running", program)
return
- cmd = "/usr/intranator/bin/arnied_helper --wait-for-program-end " \
- + program.upper() + " --wait-for-program-timeout " + str(timeout)
+ cmd = f"{BIN_ARNIED_HELPER} --wait-for-program-end " \
+ f"{program.upper()} --wait-for-program-timeout {timeout}"
+ # add one second to make sure arnied_helper is finished when we expire
+ result = run_cmd(cmd=cmd, vm=vm, timeout=timeout+1)
+ log.debug(result.stdout)
+
+
+def wait_for_arnied(timeout=60, vm=None):
+ """
+ Wait for arnied socket to be ready.
+
+ :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: :py:class:`virttest.qemu_vm.VM` or None
+ """
+ cmd = f"{BIN_ARNIED_HELPER} --wait-for-arnied-socket " \
+ f"--wait-for-arnied-socket-timeout {timeout}"
# add one second to make sure arnied_helper is finished when we expire
result = run_cmd(cmd=cmd, vm=vm, timeout=timeout+1)
log.debug(result.stdout)
: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
If `cnf_index` is set to -1, retrieve and perform regex matching on all instances.
"""
- verify_running(timeout=timeout, vm=vm)
+ wait_for_arnied(timeout=timeout, vm=vm)
platform_str = ""
if vm is not None:
platform_str = " from %s" % vm.name
: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
"""
- verify_running(timeout=timeout, vm=vm)
+ wait_for_arnied(timeout=timeout, vm=vm)
regex = "%s,(\d+): \"%s\"" % (cnf_key, value)
cnf_id = get_cnf(cnf_key, cnf_index=-1, regex=regex, compact=True, vm=vm)
if cnf_id is None:
: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
"""
- verify_running(timeout=timeout, vm=vm)
+ wait_for_arnied(timeout=timeout, vm=vm)
# firstly, build argv for get_cnf
cmd = ["get_cnf", "-j"]
if varname is not None:
: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
"""
- verify_running(timeout=timeout, vm=vm)
+ wait_for_arnied(timeout=timeout, vm=vm)
log.info("Extracting from arnied CNF_VAR %s with data %s",
varname, data)
cnf = get_cnfvar(varname=varname, data=data, vm=vm)
: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
"""
- verify_running(timeout=timeout, vm=vm)
+ wait_for_arnied(timeout=timeout, vm=vm)
cmd = "get_cnf %s %s | set_cnf -x" % (varname, instance)
run_cmd(cmd=cmd, 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
the config files will be copied there as temporary files before applying.
"""
log.info("Setting arnied configuration")
- verify_running(timeout=timeout, vm=vm)
+ wait_for_arnied(timeout=timeout, vm=vm)
config_paths = prep_config_paths(config_files)
for config_path in config_paths:
: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
This is obviously not generic but supposed to be run on the guest.
"""
log.info("Setting arnied configuration through local pipe")
- verify_running(timeout=timeout)
+ wait_for_arnied(timeout=timeout)
st, out, exit = sysmisc.run_cmd_with_pipe([BIN_SET_CNF, "-j"], inp=str(cnf))
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},
{"cmd": 'echo "LICENSE_ACCEPTED,0: \\"1\\"" | set_cnf', "stdout": b"", "returncode": 0},
{"cmd": '/usr/intranator/bin/arnied_helper --wait-for-program-end GENERATE', "stdout": b"", "returncode": 0},
- {"cmd": 'get_cnf PROVIDER 1', "stdout": b"PRODIVER 1, 'sample-provider'", "returncode": 0},
+ {"cmd": 'get_cnf PROVIDER 1', "stdout": b"1 PROVIDER,1: \"sample-provider\"", "returncode": 0},
{"cmd": 'tell-connd --online P1', "stdout": b"", "returncode": 0},
+ {"cmd": '/usr/intranator/bin/get_var ONLINE', "stdout": b"DEFAULT: 2", "returncode": 0},
{"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0},
{"cmd": 'get_cnf VIRSCAN_UPDATE_CRON | set_cnf -x', "stdout": b"", "returncode": 0},
{"cmd": 'rm -f /var/intranator/schedule/UPDATE_VIRSCAN_NODIAL*', "stdout": b"", "returncode": 0},
{"cmd": '/usr/intranator/bin/arnied_helper --transfer-mail', "stdout": b"", "returncode": 0},
+
+ {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 10', "stdout": b"", "returncode": 0},
+ {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 30', "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
with self.assertRaises(RuntimeError):
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):
- DummyCmdOutputMapping.asserted_cmds = self.cmd_db[1:3]
+ """Test accepting license."""
+ DummyCmdOutputMapping.asserted_cmds = self.cmd_db[1:3] + self.cmd_db[8:11]
arnied_wrapper.accept_licence()
# make sure an error is ignored since license might
# already be accepted
arnied_wrapper.accept_licence()
def test_go_online(self):
- DummyCmdOutputMapping.asserted_cmds = self.cmd_db[3:5]
+ DummyCmdOutputMapping.asserted_cmds = self.cmd_db[3:6]
arnied_wrapper.go_online(1)
- # TODO: for some reason failing to go online doesn't raise
- # an error which could be very misleading during debugging
- # DummyCmdOutputMapping.fail_switch = True
- # with self.assertRaises(Exception):
- # arnied_wrapper.go_online(1)
def test_disable_virscan(self):
- DummyCmdOutputMapping.asserted_cmds = self.cmd_db[5:13]
+ """Test disabling the virus scanner."""
+ DummyCmdOutputMapping.asserted_cmds = self.cmd_db[6:]
arnied_wrapper.disable_virscan()
def test_email_transfer(self):
- DummyCmdOutputMapping.asserted_cmds = self.cmd_db[13:14]
+ """Test e-mail transferring."""
+ DummyCmdOutputMapping.asserted_cmds = self.cmd_db[14:15]
arnied_wrapper.email_transfer()
git://developer.intra2net.com / pyi2ncommon / commitdiff