a2f3fca15330e3baf1c296814624bae65e244054
[ganeti-github.git] / test / py / ganeti.rapi.testutils_unittest.py
1 #!/usr/bin/python
2 #
3
4 # Copyright (C) 2012 Google Inc.
5 #
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
10 #
11 # This program is distributed in the hope that it will be useful, but
12 # WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 # General Public License for more details.
15 #
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19 # 02110-1301, USA.
20
21
22 """Script for testing ganeti.rapi.testutils"""
23
24 import unittest
25
26 from ganeti import compat
27 from ganeti import constants
28 from ganeti import errors
29 from ganeti import opcodes
30 from ganeti import luxi
31 from ganeti import rapi
32 from ganeti import utils
33
34 import ganeti.rapi.testutils
35 import ganeti.rapi.client
36
37 import testutils
38
39
40 KNOWN_UNUSED_LUXI = compat.UniqueFrozenset([
41 luxi.REQ_SUBMIT_MANY_JOBS,
42 luxi.REQ_ARCHIVE_JOB,
43 luxi.REQ_AUTO_ARCHIVE_JOBS,
44 luxi.REQ_CHANGE_JOB_PRIORITY,
45 luxi.REQ_QUERY_EXPORTS,
46 luxi.REQ_QUERY_CONFIG_VALUES,
47 luxi.REQ_QUERY_TAGS,
48 luxi.REQ_SET_DRAIN_FLAG,
49 luxi.REQ_SET_WATCHER_PAUSE,
50 ])
51
52
53 # Global variable for storing used LUXI calls
54 _used_luxi_calls = None
55
56
57 class TestHideInternalErrors(unittest.TestCase):
58 def test(self):
59 def inner():
60 raise errors.GenericError("error")
61
62 fn = rapi.testutils._HideInternalErrors(inner)
63
64 self.assertRaises(rapi.testutils.VerificationError, fn)
65
66
67 class TestVerifyOpInput(unittest.TestCase):
68 def testUnknownOpId(self):
69 voi = rapi.testutils.VerifyOpInput
70
71 self.assertRaises(rapi.testutils.VerificationError, voi, "UNK_OP_ID", None)
72
73 def testUnknownParameter(self):
74 voi = rapi.testutils.VerifyOpInput
75
76 self.assertRaises(rapi.testutils.VerificationError, voi,
77 opcodes.OpClusterRename.OP_ID, {
78 "unk": "unk",
79 })
80
81 def testWrongParameterValue(self):
82 voi = rapi.testutils.VerifyOpInput
83 self.assertRaises(rapi.testutils.VerificationError, voi,
84 opcodes.OpClusterRename.OP_ID, {
85 "name": object(),
86 })
87
88 def testSuccess(self):
89 voi = rapi.testutils.VerifyOpInput
90 voi(opcodes.OpClusterRename.OP_ID, {
91 "name": "new-name.example.com",
92 })
93
94
95 class TestVerifyOpResult(unittest.TestCase):
96 def testSuccess(self):
97 vor = rapi.testutils.VerifyOpResult
98
99 vor(opcodes.OpClusterVerify.OP_ID, {
100 constants.JOB_IDS_KEY: [
101 (False, "error message"),
102 ],
103 })
104
105 def testWrongResult(self):
106 vor = rapi.testutils.VerifyOpResult
107
108 self.assertRaises(rapi.testutils.VerificationError, vor,
109 opcodes.OpClusterVerify.OP_ID, [])
110
111 def testNoResultCheck(self):
112 vor = rapi.testutils.VerifyOpResult
113
114 assert opcodes.OpTestDummy.OP_RESULT is None
115
116 vor(opcodes.OpTestDummy.OP_ID, None)
117
118
119 class TestInputTestClient(unittest.TestCase):
120 def setUp(self):
121 self.cl = rapi.testutils.InputTestClient()
122
123 def tearDown(self):
124 _used_luxi_calls.update(self.cl._GetLuxiCalls())
125
126 def testGetInfo(self):
127 self.assertTrue(self.cl.GetInfo() is NotImplemented)
128
129 def testPrepareExport(self):
130 result = self.cl.PrepareExport("inst1.example.com",
131 constants.EXPORT_MODE_LOCAL)
132 self.assertTrue(result is NotImplemented)
133 self.assertRaises(rapi.testutils.VerificationError, self.cl.PrepareExport,
134 "inst1.example.com", "###invalid###")
135
136 def testGetJobs(self):
137 self.assertTrue(self.cl.GetJobs() is NotImplemented)
138
139 def testQuery(self):
140 result = self.cl.Query(constants.QR_NODE, ["name"])
141 self.assertTrue(result is NotImplemented)
142
143 def testQueryFields(self):
144 result = self.cl.QueryFields(constants.QR_INSTANCE)
145 self.assertTrue(result is NotImplemented)
146
147 def testCancelJob(self):
148 self.assertTrue(self.cl.CancelJob("1") is NotImplemented)
149
150 def testGetNodes(self):
151 self.assertTrue(self.cl.GetNodes() is NotImplemented)
152
153 def testGetInstances(self):
154 self.assertTrue(self.cl.GetInstances() is NotImplemented)
155
156 def testGetGroups(self):
157 self.assertTrue(self.cl.GetGroups() is NotImplemented)
158
159 def testWaitForJobChange(self):
160 result = self.cl.WaitForJobChange("1", ["id"], None, None)
161 self.assertTrue(result is NotImplemented)
162
163
164 class CustomTestRunner(unittest.TextTestRunner):
165 def run(self, *args):
166 global _used_luxi_calls
167 assert _used_luxi_calls is None
168
169 diff = (KNOWN_UNUSED_LUXI - luxi.REQ_ALL)
170 assert not diff, "Non-existing LUXI calls listed as unused: %s" % diff
171
172 _used_luxi_calls = set()
173 try:
174 # Run actual tests
175 result = unittest.TextTestRunner.run(self, *args)
176
177 diff = _used_luxi_calls & KNOWN_UNUSED_LUXI
178 if diff:
179 raise AssertionError("LUXI methods marked as unused were called: %s" %
180 utils.CommaJoin(diff))
181
182 diff = (luxi.REQ_ALL - KNOWN_UNUSED_LUXI - _used_luxi_calls)
183 if diff:
184 raise AssertionError("The following LUXI methods were not used: %s" %
185 utils.CommaJoin(diff))
186 finally:
187 # Reset global variable
188 _used_luxi_calls = None
189
190 return result
191
192
193 if __name__ == "__main__":
194 testutils.GanetiTestProgram(testRunner=CustomTestRunner)