#!/usr/bin/env python # The software in this package is distributed under the GNU General # Public License version 2 (with a special exception described below). # # A copy of GNU General Public License (GPL) is included in this distribution, # in the file COPYING.GPL. # # As a special exception, if other files instantiate templates or use macros # or inline functions from this file, or you compile this file and link it # with other works to produce a work based on this file, this file # does not by itself cause the resulting work to be covered # by the GNU General Public License. # # However the source code for this file must still be made available # in accordance with section (3) of the GNU General Public License. # # This exception does not invalidate any other reasons why a work based # on this file might be covered by the GNU General Public License. # # Copyright (c) 2016-2018 Intra2net AG import unittest import unittest.mock as mock import time from src import arnied_wrapper class DummyCmdOutputMapping: """ 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 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) if self.fail_switch: 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'] self.stdout = dummy_cmd['stdout'] return raise ValueError("Could not locate the command '%s' among the known answers " "for the universe" % cmd) # 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 = [] def test_verify_running(self): """Test checking for running programs.""" DummyCmdOutputMapping.asserted_cmds = [{"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0}] 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 = [{"cmd": "/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 10", "stdout": b"", "returncode": 0}] arnied_wrapper.wait_for_arnied(timeout=10) def test_go_online(self): DummyCmdOutputMapping.asserted_cmds = [{"cmd": 'tell-connd --online P1', "stdout": b"", "returncode": 0}, {"cmd": '/usr/intranator/bin/get_var ONLINE', "stdout": b"DEFAULT: 2", "returncode": 0}] arnied_wrapper.go_online(1) def test_email_transfer(self): """Test e-mail transferring.""" DummyCmdOutputMapping.asserted_cmds = [{"cmd": '/usr/intranator/bin/arnied_helper --transfer-mail', "stdout": b"", "returncode": 0}] arnied_wrapper.email_transfer() def test_wait_for_no_generate(self): """Test waiting for generate if there is no generate scheduled""" DummyCmdOutputMapping.asserted_cmds = [ {"cmd": f'/usr/intranator/bin/arnied_helper --is-scheduled-or-running {job}', "stdout": b"", "returncode": 1} for job in ("GENERATE", "GENERATE_OFFLINE")] start_time = time.time() self.assertEqual(arnied_wrapper.wait_for_generate(), True) self.assertLess(time.time() - start_time, 1) # no sleep, just call and return at once def test_wait_for_generate(self): """Test waiting for generate if there is generate scheduled""" DummyCmdOutputMapping.asserted_cmds = [ {"cmd": f'/usr/intranator/bin/arnied_helper --is-scheduled-or-running {job}', "stdout": b"", "returncode": 0} for job in ("GENERATE", "GENERATE_OFFLINE")] start_time = time.time() self.assertEqual(arnied_wrapper.wait_for_generate(timeout=2), False) end_time = time.time() self.assertGreater(end_time - start_time, 1.5) self.assertLess(end_time - start_time, 2.5) def test_wait_for_email_transfer_timeout(self): """Test waiting for email transfer that reaches timeout.""" DummyCmdOutputMapping.asserted_cmds = ( {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0}, {"cmd": 'postqueue -j', "stdout": b"not empty", "returncode": 0}, ) start_time = time.time() self.assertRaises(TimeoutError, arnied_wrapper.wait_for_email_transfer, 2) time_diff = time.time() - start_time self.assertGreater(time_diff, 1.5) self.assertLess(time_diff, 2.5) def test_wait_for_email_transfer_succeed(self): """Test waiting for email transfer that succeeds at once""" DummyCmdOutputMapping.asserted_cmds = ( {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0}, {"cmd": 'postqueue -j', "stdout": b"", "returncode": 0}, ) start_time = time.time() arnied_wrapper.wait_for_email_transfer() time_diff = time.time() - start_time self.assertLess(time_diff, 1)