developer.intra2net.com Git - pyi2ncommon/blob - test/test_arnied_wrapper.py

   1 #!/usr/bin/env python
   2
   3 # The software in this package is distributed under the GNU General
   4 # Public License version 2 (with a special exception described below).
   5 #
   6 # A copy of GNU General Public License (GPL) is included in this distribution,
   7 # in the file COPYING.GPL.
   8 #
   9 # As a special exception, if other files instantiate templates or use macros
  10 # or inline functions from this file, or you compile this file and link it
  11 # with other works to produce a work based on this file, this file
  12 # does not by itself cause the resulting work to be covered
  13 # by the GNU General Public License.
  14 #
  15 # However the source code for this file must still be made available
  16 # in accordance with section (3) of the GNU General Public License.
  17 #
  18 # This exception does not invalidate any other reasons why a work based
  19 # on this file might be covered by the GNU General Public License.
  20 #
  21 # Copyright (c) 2016-2018 Intra2net AG <info@intra2net.com>
  22
  23 import unittest
  24 import unittest.mock as mock
  25 import time
  26
  27 from src import arnied_wrapper
  28
  29
  30 class DummyCmdOutputMapping:
  31     """
  32     Class to replace the :py:function:`arnied_wrapper.run_cmd` function.
  33
  34     In the arnied_wrapper, when running a command, instead of calling the actual
  35     function, this class' constructor will be invoked and an instance returned
  36     instead. It stubs :py:class:`subprocess.CompletedProcess`, such that the
  37     returncode and stdout attributes are set depending on the command or on the
  38     test.
  39     """
  40
  41     # whether to return 1 as a fail indicator
  42     fail_switch = False
  43     # mapping between expected commands and their mocked output + return code
  44     asserted_cmds = []
  45
  46     def __init__(self, cmd="", ignore_errors=False, vm=None, timeout=60):
  47         """
  48         Class constructor to mimic the run function of the arnied wrapper.
  49
  50         Arguments are the same of the mocked function.
  51         """
  52         self.returncode = 0
  53         self.stdout = b""
  54         self._get_result(cmd)
  55         if self.fail_switch:
  56             self.returncode = 1
  57
  58     def __str__(self):
  59         """String representation of this class."""
  60         return "status %i, stdout %s" % (self.returncode, self.stdout)
  61
  62     def _get_result(self, cmd):
  63         """
  64         Return the first (or raise) values in the mapping matching the command.
  65
  66         :param str cmd: command to check the mapping against
  67         :raises: :py:class:`ValueError` if the command has no corresponding
  68                  mapping
  69         :returns: this instance with the return code and stdout attributes set
  70         :rtype: :py:class:`DummyCmdOutputMapping`
  71         """
  72         for dummy_cmd in self.asserted_cmds:
  73             if dummy_cmd['cmd'] == cmd:
  74                 self.returncode = dummy_cmd['returncode']
  75                 self.stdout = dummy_cmd['stdout']
  76                 return
  77         raise ValueError("Could not locate the command '%s' among the known answers "
  78                          "for the universe" % cmd)
  79
  80
  81 # make sure that invoking `run_cmd` returns an instance of DummyCmdOutputMapping
  82 @mock.patch("src.arnied_wrapper.run_cmd", DummyCmdOutputMapping)
  83 class ArniedWrapperTest(unittest.TestCase):
  84     def setUp(self):
  85         DummyCmdOutputMapping.fail_switch = False
  86         DummyCmdOutputMapping.asserted_cmds = []
  87
  88     def test_verify_running(self):
  89         """Test checking for running programs."""
  90         DummyCmdOutputMapping.asserted_cmds = [{"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0}]
  91         arnied_wrapper.verify_running(timeout=1)
  92         DummyCmdOutputMapping.fail_switch = True
  93         with self.assertRaises(RuntimeError):
  94             arnied_wrapper.verify_running(timeout=1)
  95
  96     def test_wait_for_arnied(self):
  97         """Test waiting for arnied to be ready."""
  98         DummyCmdOutputMapping.asserted_cmds = [{"cmd": "/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 10", "stdout": b"", "returncode": 0}]
  99         arnied_wrapper.wait_for_arnied(timeout=10)
 100
 101     def test_go_online(self):
 102         DummyCmdOutputMapping.asserted_cmds = [{"cmd": 'tell-connd --online P1', "stdout": b"", "returncode": 0},
 103                                                {"cmd": '/usr/intranator/bin/get_var ONLINE', "stdout": b"DEFAULT: 2", "returncode": 0}]
 104         arnied_wrapper.go_online(1)
 105
 106     def test_email_transfer(self):
 107         """Test e-mail transferring."""
 108         DummyCmdOutputMapping.asserted_cmds = [{"cmd": '/usr/intranator/bin/arnied_helper --transfer-mail', "stdout": b"", "returncode": 0}]
 109         arnied_wrapper.email_transfer()
 110
 111     def test_wait_for_no_generate(self):
 112         """Test waiting for generate if there is no generate scheduled"""
 113         DummyCmdOutputMapping.asserted_cmds = [
 114             {"cmd": f'/usr/intranator/bin/arnied_helper --is-scheduled-or-running {job}', "stdout": b"", "returncode": 1}
 115                 for job in ("GENERATE", "GENERATE_OFFLINE")]
 116         start_time = time.time()
 117         self.assertEqual(arnied_wrapper.wait_for_generate(),  True)
 118         self.assertLess(time.time() - start_time, 1)    # no sleep, just call and return at once
 119
 120     def test_wait_for_generate(self):
 121         """Test waiting for generate if there is generate scheduled"""
 122         DummyCmdOutputMapping.asserted_cmds = [
 123             {"cmd": f'/usr/intranator/bin/arnied_helper --is-scheduled-or-running {job}', "stdout": b"", "returncode": 0}
 124             for job in ("GENERATE", "GENERATE_OFFLINE")]
 125         start_time = time.time()
 126         self.assertEqual(arnied_wrapper.wait_for_generate(timeout=2),  False)
 127         end_time = time.time()
 128         self.assertGreater(end_time - start_time, 1.5)
 129         self.assertLess(end_time - start_time, 2.5)
 130
 131     def test_wait_for_email_transfer_timeout(self):
 132         """Test waiting for email transfer that reaches timeout."""
 133         DummyCmdOutputMapping.asserted_cmds = (
 134             {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0},
 135             {"cmd": 'postqueue -j', "stdout": b"not empty", "returncode": 0},
 136         )
 137         start_time = time.time()
 138         self.assertRaises(TimeoutError, arnied_wrapper.wait_for_email_transfer, 2)
 139         time_diff = time.time() - start_time
 140         self.assertGreater(time_diff, 1.5)
 141         self.assertLess(time_diff, 2.5)
 142
 143     def test_wait_for_email_transfer_succeed(self):
 144         """Test waiting for email transfer that succeeds at once"""
 145         DummyCmdOutputMapping.asserted_cmds = (
 146             {"cmd": 'postqueue -f', "stdout": b"", "returncode": 0},
 147             {"cmd": 'postqueue -j', "stdout": b"", "returncode": 0},
 148         )
 149         start_time = time.time()
 150         arnied_wrapper.wait_for_email_transfer()
 151         time_diff = time.time() - start_time
 152         self.assertLess(time_diff, 1)