Add a new function to wait for arnied to be ready
[pyi2ncommon] / 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 subprocess
26
27 from src import arnied_wrapper
28
29
30 class DummyCmdOutputMapping(object):
31
32     fail_switch = False
33     cmds = [
34         {"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0},
35
36         {"cmd": 'echo "LICENSE_ACCEPTED,0: \\"1\\"" | set_cnf', "stdout": b"", "returncode": 0},
37         {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-program-end GENERATE', "stdout": b"", "returncode": 0},
38
39         {"cmd": 'get_cnf PROVIDER 1', "stdout": b"1  PROVIDER,1: \"sample-provider\"", "returncode": 0},
40         {"cmd": 'tell-connd --online P1', "stdout": b"", "returncode": 0},
41         {"cmd": '/usr/intranator/bin/get_var ONLINE', "stdout": b"DEFAULT: 2", "returncode": 0},
42
43         {"cmd": "pgrep -l -x arnied", "stdout": b"", "returncode": 0},
44         {"cmd": 'get_cnf VIRSCAN_UPDATE_CRON  | set_cnf -x', "stdout": b"", "returncode": 0},
45         {"cmd": '/usr/intranator/bin/arnied_helper --is-scheduled-or-running GENERATE', "stdout": b"", "returncode": 1},
46         {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-program-end GENERATE --wait-for-program-timeout 300', "stdout": b"", "returncode": 1},
47         {"cmd": '/usr/intranator/bin/arnied_helper --is-scheduled-or-running GENERATE_OFFLINE', "stdout": b"", "returncode": 1},
48         {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-program-end GENERATE_OFFLINE --wait-for-program-timeout 300', "stdout": b"", "returncode": 1},
49         {"cmd": 'echo \'VIRSCAN_UPDATE_DNS_PUSH,0:"0"\' |set_cnf', "stdout": b"", "returncode": 0},
50         {"cmd": 'rm -f /var/intranator/schedule/UPDATE_VIRSCAN_NODIAL*', "stdout": b"", "returncode": 0},
51
52         {"cmd": '/usr/intranator/bin/arnied_helper --transfer-mail', "stdout": b"", "returncode": 0},
53
54         {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 10', "stdout": b"", "returncode": 0},
55         {"cmd": '/usr/intranator/bin/arnied_helper --wait-for-arnied-socket --wait-for-arnied-socket-timeout 30', "stdout": b"", "returncode": 0},
56     ]
57     asserted_cmds = []
58
59     def __init__(self, cmd="", ignore_errors=False, vm=None, timeout=60):
60         self.returncode = 0
61         self.stdout = b""
62         self._get_result(cmd)
63         if self.fail_switch:
64             self.returncode = 1
65
66     def __str__(self):
67         return "status %i, stdout %s" % (self.returncode, self.stdout)
68
69     def _get_result(self, cmd):
70         for dummy_cmd in self.asserted_cmds:
71             if dummy_cmd['cmd'] == cmd:
72                 self.returncode = dummy_cmd['returncode']
73                 self.stdout = dummy_cmd['stdout']
74                 return
75         raise ValueError("Could not locate the command '%s' among the known answers "
76                          "for the universe" % cmd)
77
78
79 @mock.patch('src.arnied_wrapper.run_cmd', DummyCmdOutputMapping)
80 class ArniedWrapperTest(unittest.TestCase):
81
82     def setUp(self):
83         DummyCmdOutputMapping.fail_switch = False
84         DummyCmdOutputMapping.asserted_cmds = []
85         self.cmd_db = DummyCmdOutputMapping.cmds
86
87     def test_verify_running(self):
88         DummyCmdOutputMapping.asserted_cmds = self.cmd_db[0:1]
89         arnied_wrapper.verify_running(timeout=1)
90         DummyCmdOutputMapping.fail_switch = True
91         with self.assertRaises(RuntimeError):
92             arnied_wrapper.verify_running(timeout=1)
93
94     def test_wait_for_arnied(self):
95         DummyCmdOutputMapping.asserted_cmds = self.cmd_db
96         arnied_wrapper.wait_for_arnied(timeout=10)
97
98     def test_accept_license(self):
99         DummyCmdOutputMapping.asserted_cmds = self.cmd_db[1:3]
100         arnied_wrapper.accept_licence()
101         # make sure an error is ignored since license might
102         # already be accepted
103         DummyCmdOutputMapping.fail_switch = True
104         arnied_wrapper.accept_licence()
105
106     def test_go_online(self):
107         DummyCmdOutputMapping.asserted_cmds = self.cmd_db[3:6]
108         arnied_wrapper.go_online(1)
109
110     def test_disable_virscan(self):
111         DummyCmdOutputMapping.asserted_cmds = self.cmd_db[6:]
112         arnied_wrapper.disable_virscan()
113
114     def test_email_transfer(self):
115         DummyCmdOutputMapping.asserted_cmds = self.cmd_db[14:15]
116         arnied_wrapper.email_transfer()