Increase version to 1.7.3
[pyi2ncommon] / test / test_arnied_wrapper.py
CommitLineData
f49f6323
PD
1#!/usr/bin/env python
2
11cbb815
PD
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
f49f6323
PD
23import unittest
24import unittest.mock as mock
787caa73 25import time
f49f6323 26
2f3212fe 27from src import arnied_wrapper
f49f6323
PD
28
29
7f66ff3e 30class DummyCmdOutputMapping:
f4c8f40e
SA
31 """
32 Class to replace the :py:function:`arnied_wrapper.run_cmd` function.
f49f6323 33
f4c8f40e
SA
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
f49f6323 42 fail_switch = False
f4c8f40e 43 # mapping between expected commands and their mocked output + return code
f49f6323
PD
44 asserted_cmds = []
45
80d82f13 46 def __init__(self, cmd="", ignore_errors=False, vm=None, timeout=60):
f4c8f40e
SA
47 """
48 Class constructor to mimic the run function of the arnied wrapper.
49
50 Arguments are the same of the mocked function.
51 """
f49f6323 52 self.returncode = 0
7102665a 53 self.stdout = b""
f49f6323
PD
54 self._get_result(cmd)
55 if self.fail_switch:
56 self.returncode = 1
57
58 def __str__(self):
f4c8f40e 59 """String representation of this class."""
f49f6323
PD
60 return "status %i, stdout %s" % (self.returncode, self.stdout)
61
62 def _get_result(self, cmd):
f4c8f40e
SA
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 """
f49f6323
PD
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
f4c8f40e
SA
81# make sure that invoking `run_cmd` returns an instance of DummyCmdOutputMapping
82@mock.patch("src.arnied_wrapper.run_cmd", DummyCmdOutputMapping)
f49f6323 83class ArniedWrapperTest(unittest.TestCase):
f49f6323
PD
84 def setUp(self):
85 DummyCmdOutputMapping.fail_switch = False
86 DummyCmdOutputMapping.asserted_cmds = []
f49f6323
PD
87
88 def test_verify_running(self):
f4c8f40e 89 """Test checking for running programs."""
4965c436 90 DummyCmdOutputMapping.asserted_cmds = [{"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0}]
f49f6323
PD
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
f1ca3964 96 def test_wait_for_arnied(self):
f4c8f40e 97 """Test waiting for arnied to be ready."""
4965c436 98 DummyCmdOutputMapping.asserted_cmds = [{"cmd": "/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 10", "stdout": b"", "returncode": 0}]
f1ca3964
SA
99 arnied_wrapper.wait_for_arnied(timeout=10)
100
f49f6323 101 def test_go_online(self):
4965c436
PD
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}]
f49f6323 104 arnied_wrapper.go_online(1)
f49f6323 105
f49f6323 106 def test_email_transfer(self):
f4c8f40e 107 """Test e-mail transferring."""
4965c436 108 DummyCmdOutputMapping.asserted_cmds = [{"cmd": '/usr/intranator/bin/arnied_helper --transfer-mail', "stdout": b"", "returncode": 0}]
f49f6323 109 arnied_wrapper.email_transfer()
787caa73
CH
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)
fdf40ad2
CH
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)