Document the rapi client not to have a QueryNetworks method
[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_NETWORKS,
48 luxi.REQ_QUERY_TAGS,
49 luxi.REQ_SET_DRAIN_FLAG,
50 luxi.REQ_SET_WATCHER_PAUSE,
51 ])
52
53
54 # Global variable for storing used LUXI calls
55 _used_luxi_calls = None
56
57
58 class TestHideInternalErrors(unittest.TestCase):
59 def test(self):
60 def inner():
61 raise errors.GenericError("error")
62
63 fn = rapi.testutils._HideInternalErrors(inner)
64
65 self.assertRaises(rapi.testutils.VerificationError, fn)
66
67
68 class TestVerifyOpInput(unittest.TestCase):
69 def testUnknownOpId(self):
70 voi = rapi.testutils.VerifyOpInput
71
72 self.assertRaises(rapi.testutils.VerificationError, voi, "UNK_OP_ID", None)
73
74 def testUnknownParameter(self):
75 voi = rapi.testutils.VerifyOpInput
76
77 self.assertRaises(rapi.testutils.VerificationError, voi,
78 opcodes.OpClusterRename.OP_ID, {
79 "unk": "unk",
80 })
81
82 def testWrongParameterValue(self):
83 voi = rapi.testutils.VerifyOpInput
84 self.assertRaises(rapi.testutils.VerificationError, voi,
85 opcodes.OpClusterRename.OP_ID, {
86 "name": object(),
87 })
88
89 def testSuccess(self):
90 voi = rapi.testutils.VerifyOpInput
91 voi(opcodes.OpClusterRename.OP_ID, {
92 "name": "new-name.example.com",
93 })
94
95
96 class TestVerifyOpResult(unittest.TestCase):
97 def testSuccess(self):
98 vor = rapi.testutils.VerifyOpResult
99
100 vor(opcodes.OpClusterVerify.OP_ID, {
101 constants.JOB_IDS_KEY: [
102 (False, "error message"),
103 ],
104 })
105
106 def testWrongResult(self):
107 vor = rapi.testutils.VerifyOpResult
108
109 self.assertRaises(rapi.testutils.VerificationError, vor,
110 opcodes.OpClusterVerify.OP_ID, [])
111
112 def testNoResultCheck(self):
113 vor = rapi.testutils.VerifyOpResult
114
115 assert opcodes.OpTestDummy.OP_RESULT is None
116
117 vor(opcodes.OpTestDummy.OP_ID, None)
118
119
120 class TestInputTestClient(unittest.TestCase):
121 def setUp(self):
122 self.cl = rapi.testutils.InputTestClient()
123
124 def tearDown(self):
125 _used_luxi_calls.update(self.cl._GetLuxiCalls())
126
127 def testGetInfo(self):
128 self.assertTrue(self.cl.GetInfo() is NotImplemented)
129
130 def testPrepareExport(self):
131 result = self.cl.PrepareExport("inst1.example.com",
132 constants.EXPORT_MODE_LOCAL)
133 self.assertTrue(result is NotImplemented)
134 self.assertRaises(rapi.testutils.VerificationError, self.cl.PrepareExport,
135 "inst1.example.com", "###invalid###")
136
137 def testGetJobs(self):
138 self.assertTrue(self.cl.GetJobs() is NotImplemented)
139
140 def testQuery(self):
141 result = self.cl.Query(constants.QR_NODE, ["name"])
142 self.assertTrue(result is NotImplemented)
143
144 def testQueryFields(self):
145 result = self.cl.QueryFields(constants.QR_INSTANCE)
146 self.assertTrue(result is NotImplemented)
147
148 def testCancelJob(self):
149 self.assertTrue(self.cl.CancelJob("1") is NotImplemented)
150
151 def testGetNodes(self):
152 self.assertTrue(self.cl.GetNodes() is NotImplemented)
153
154 def testGetInstances(self):
155 self.assertTrue(self.cl.GetInstances() is NotImplemented)
156
157 def testGetGroups(self):
158 self.assertTrue(self.cl.GetGroups() is NotImplemented)
159
160 def testWaitForJobChange(self):
161 result = self.cl.WaitForJobChange("1", ["id"], None, None)
162 self.assertTrue(result is NotImplemented)
163
164
165 class CustomTestRunner(unittest.TextTestRunner):
166 def run(self, *args):
167 global _used_luxi_calls
168 assert _used_luxi_calls is None
169
170 diff = (KNOWN_UNUSED_LUXI - luxi.REQ_ALL)
171 assert not diff, "Non-existing LUXI calls listed as unused: %s" % diff
172
173 _used_luxi_calls = set()
174 try:
175 # Run actual tests
176 result = unittest.TextTestRunner.run(self, *args)
177
178 diff = _used_luxi_calls & KNOWN_UNUSED_LUXI
179 if diff:
180 raise AssertionError("LUXI methods marked as unused were called: %s" %
181 utils.CommaJoin(diff))
182
183 diff = (luxi.REQ_ALL - KNOWN_UNUSED_LUXI - _used_luxi_calls)
184 if diff:
185 raise AssertionError("The following LUXI methods were not used: %s" %
186 utils.CommaJoin(diff))
187 finally:
188 # Reset global variable
189 _used_luxi_calls = None
190
191 return result
192
193
194 if __name__ == "__main__":
195 testutils.GanetiTestProgram(testRunner=CustomTestRunner)