4 # Copyright (C) 2006, 2007 Google Inc.
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.
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.
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
22 """Script for unittesting the hooks module"""
31 from ganeti import errors
32 from ganeti import opcodes
33 from ganeti import mcpu
34 from ganeti import backend
35 from ganeti import constants
36 from ganeti import cmdlib
37 from ganeti import rpc
38 from ganeti.constants import HKR_SUCCESS, HKR_FAIL, HKR_SKIP
40 from mocks import FakeConfig, FakeProc, FakeContext
45 class FakeLU(cmdlib.LogicalUnit):
47 def BuildHooksEnv(self):
48 return {}, ["localhost"], ["localhost"]
51 class TestHooksRunner(unittest.TestCase):
52 """Testing case for HooksRunner"""
55 self.tmpdir = tempfile.mkdtemp()
56 self.torm.append((self.tmpdir, True))
57 self.logdir = tempfile.mkdtemp()
58 self.torm.append((self.logdir, True))
61 for i in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
62 dname = "%s/%s-%s.d" % (self.tmpdir, self.hpath, i)
64 self.torm.append((dname, True))
65 self.ph_dirs[i] = dname
66 self.hr = backend.HooksRunner(hooks_base_dir=self.tmpdir)
70 for path, kind in self.torm:
76 def _rname(self, fname):
77 return "/".join(fname.split("/")[-2:])
81 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
82 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), [])
84 def testSkipNonExec(self):
85 """Test skip non-exec file"""
86 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
87 fname = "%s/test" % self.ph_dirs[phase]
90 self.torm.append((fname, False))
91 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
92 [(self._rname(fname), HKR_SKIP, "")])
94 def testSkipInvalidName(self):
95 """Test skip script with invalid name"""
96 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
97 fname = "%s/a.off" % self.ph_dirs[phase]
99 f.write("#!/bin/sh\nexit 0\n")
101 os.chmod(fname, 0700)
102 self.torm.append((fname, False))
103 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
104 [(self._rname(fname), HKR_SKIP, "")])
106 def testSkipDir(self):
107 """Test skip directory"""
108 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
109 fname = "%s/testdir" % self.ph_dirs[phase]
111 self.torm.append((fname, True))
112 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
113 [(self._rname(fname), HKR_SKIP, "")])
115 def testSuccess(self):
116 """Test success execution"""
117 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
118 fname = "%s/success" % self.ph_dirs[phase]
120 f.write("#!/bin/sh\nexit 0\n")
122 self.torm.append((fname, False))
123 os.chmod(fname, 0700)
124 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
125 [(self._rname(fname), HKR_SUCCESS, "")])
127 def testSymlink(self):
128 """Test running a symlink"""
129 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
130 fname = "%s/success" % self.ph_dirs[phase]
131 os.symlink("/bin/true", fname)
132 self.torm.append((fname, False))
133 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
134 [(self._rname(fname), HKR_SUCCESS, "")])
137 """Test success execution"""
138 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
139 fname = "%s/success" % self.ph_dirs[phase]
141 f.write("#!/bin/sh\nexit 1\n")
143 self.torm.append((fname, False))
144 os.chmod(fname, 0700)
145 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
146 [(self._rname(fname), HKR_FAIL, "")])
148 def testCombined(self):
149 """Test success, failure and skip all in one test"""
150 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
152 for fbase, ecode, rs in [("00succ", 0, HKR_SUCCESS),
153 ("10fail", 1, HKR_FAIL),
154 ("20inv.", 0, HKR_SKIP),
156 fname = "%s/%s" % (self.ph_dirs[phase], fbase)
158 f.write("#!/bin/sh\nexit %d\n" % ecode)
160 self.torm.append((fname, False))
161 os.chmod(fname, 0700)
162 expect.append((self._rname(fname), rs, ""))
163 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)
165 def testOrdering(self):
166 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
168 for fbase in ["10s1",
174 fname = "%s/%s" % (self.ph_dirs[phase], fbase)
175 os.symlink("/bin/true", fname)
176 self.torm.append((fname, False))
177 expect.append((self._rname(fname), HKR_SUCCESS, ""))
179 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)
182 """Test environment execution"""
183 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
185 fname = "%s/%s" % (self.ph_dirs[phase], fbase)
186 os.symlink("/usr/bin/env", fname)
187 self.torm.append((fname, False))
188 env_snt = {"PHASE": phase}
189 env_exp = "PHASE=%s" % phase
190 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, env_snt),
191 [(self._rname(fname), HKR_SUCCESS, env_exp)])
194 class TestHooksMaster(unittest.TestCase):
195 """Testing case for HooksMaster"""
197 def _call_false(*args):
198 """Fake call_hooks_runner function which returns False."""
202 def _call_nodes_false(node_list, hpath, phase, env):
203 """Fake call_hooks_runner function.
205 @rtype: dict of node -> L{rpc.RpcResult} with an rpc error
206 @return: rpc failure from all nodes
209 return dict([(node, rpc.RpcResult('error', failed=True,
210 node=node, call='FakeError')) for node in node_list])
213 def _call_script_fail(node_list, hpath, phase, env):
214 """Fake call_hooks_runner function.
216 @rtype: dict of node -> L{rpc.RpcResult} with a failed script result
217 @return: script execution failure from all nodes
221 return dict([(node, rr((True, [("utest", constants.HKR_FAIL, "err")]),
222 node=node, call='FakeScriptFail'))
223 for node in node_list])
226 def _call_script_succeed(node_list, hpath, phase, env):
227 """Fake call_hooks_runner function.
229 @rtype: dict of node -> L{rpc.RpcResult} with a successful script result
230 @return: script execution from all nodes
234 return dict([(node, rr(True, [("utest", constants.HKR_SUCCESS, "ok")],
235 node=node, call='FakeScriptOk'))
236 for node in node_list])
239 self.op = opcodes.OpCode()
240 self.context = FakeContext()
241 # WARNING: here we pass None as RpcRunner instance since we know
242 # our usage via HooksMaster will not use lu.rpc
243 self.lu = FakeLU(FakeProc(), self.op, self.context, None)
245 def testTotalFalse(self):
246 """Test complete rpc failure"""
247 hm = mcpu.HooksMaster(self._call_false, self.lu)
248 self.failUnlessRaises(errors.HooksFailure,
249 hm.RunPhase, constants.HOOKS_PHASE_PRE)
250 hm.RunPhase(constants.HOOKS_PHASE_POST)
252 def testIndividualFalse(self):
253 """Test individual node failure"""
254 hm = mcpu.HooksMaster(self._call_nodes_false, self.lu)
255 hm.RunPhase(constants.HOOKS_PHASE_PRE)
256 #self.failUnlessRaises(errors.HooksFailure,
257 # hm.RunPhase, constants.HOOKS_PHASE_PRE)
258 hm.RunPhase(constants.HOOKS_PHASE_POST)
260 def testScriptFalse(self):
261 """Test individual rpc failure"""
262 hm = mcpu.HooksMaster(self._call_script_fail, self.lu)
263 self.failUnlessRaises(errors.HooksAbort,
264 hm.RunPhase, constants.HOOKS_PHASE_PRE)
265 hm.RunPhase(constants.HOOKS_PHASE_POST)
267 def testScriptSucceed(self):
268 """Test individual rpc failure"""
269 hm = mcpu.HooksMaster(self._call_script_succeed, self.lu)
270 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
274 if __name__ == '__main__':
275 testutils.GanetiTestProgram()