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
42 class FakeLU(cmdlib.LogicalUnit):
44 def BuildHooksEnv(self):
45 return {}, ["localhost"], ["localhost"]
47 class TestHooksRunner(unittest.TestCase):
48 """Testing case for HooksRunner"""
51 self.tmpdir = tempfile.mkdtemp()
52 self.torm.append((self.tmpdir, True))
53 self.logdir = tempfile.mkdtemp()
54 self.torm.append((self.logdir, True))
57 for i in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
58 dname = "%s/%s-%s.d" % (self.tmpdir, self.hpath, i)
60 self.torm.append((dname, True))
61 self.ph_dirs[i] = dname
62 self.hr = backend.HooksRunner(hooks_base_dir=self.tmpdir)
66 for path, kind in self.torm:
72 def _rname(self, fname):
73 return "/".join(fname.split("/")[-2:])
77 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
78 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), [])
80 def testSkipNonExec(self):
81 """Test skip non-exec file"""
82 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
83 fname = "%s/test" % self.ph_dirs[phase]
86 self.torm.append((fname, False))
87 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
88 [(self._rname(fname), HKR_SKIP, "")])
90 def testSkipInvalidName(self):
91 """Test skip script with invalid name"""
92 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
93 fname = "%s/a.off" % self.ph_dirs[phase]
95 f.write("#!/bin/sh\nexit 0\n")
98 self.torm.append((fname, False))
99 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
100 [(self._rname(fname), HKR_SKIP, "")])
102 def testSkipDir(self):
103 """Test skip directory"""
104 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
105 fname = "%s/testdir" % self.ph_dirs[phase]
107 self.torm.append((fname, True))
108 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
109 [(self._rname(fname), HKR_SKIP, "")])
111 def testSuccess(self):
112 """Test success execution"""
113 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
114 fname = "%s/success" % self.ph_dirs[phase]
116 f.write("#!/bin/sh\nexit 0\n")
118 self.torm.append((fname, False))
119 os.chmod(fname, 0700)
120 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
121 [(self._rname(fname), HKR_SUCCESS, "")])
123 def testSymlink(self):
124 """Test running a symlink"""
125 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
126 fname = "%s/success" % self.ph_dirs[phase]
127 os.symlink("/bin/true", fname)
128 self.torm.append((fname, False))
129 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
130 [(self._rname(fname), HKR_SUCCESS, "")])
133 """Test success execution"""
134 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
135 fname = "%s/success" % self.ph_dirs[phase]
137 f.write("#!/bin/sh\nexit 1\n")
139 self.torm.append((fname, False))
140 os.chmod(fname, 0700)
141 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}),
142 [(self._rname(fname), HKR_FAIL, "")])
144 def testCombined(self):
145 """Test success, failure and skip all in one test"""
146 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
148 for fbase, ecode, rs in [("00succ", 0, HKR_SUCCESS),
149 ("10fail", 1, HKR_FAIL),
150 ("20inv.", 0, HKR_SKIP),
152 fname = "%s/%s" % (self.ph_dirs[phase], fbase)
154 f.write("#!/bin/sh\nexit %d\n" % ecode)
156 self.torm.append((fname, False))
157 os.chmod(fname, 0700)
158 expect.append((self._rname(fname), rs, ""))
159 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)
161 def testOrdering(self):
162 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
164 for fbase in ["10s1",
170 fname = "%s/%s" % (self.ph_dirs[phase], fbase)
171 os.symlink("/bin/true", fname)
172 self.torm.append((fname, False))
173 expect.append((self._rname(fname), HKR_SUCCESS, ""))
175 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, {}), expect)
178 """Test environment execution"""
179 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
181 fname = "%s/%s" % (self.ph_dirs[phase], fbase)
182 os.symlink("/usr/bin/env", fname)
183 self.torm.append((fname, False))
184 env_snt = {"PHASE": phase}
185 env_exp = "PHASE=%s" % phase
186 self.failUnlessEqual(self.hr.RunHooks(self.hpath, phase, env_snt),
187 [(self._rname(fname), HKR_SUCCESS, env_exp)])
190 class TestHooksMaster(unittest.TestCase):
191 """Testing case for HooksMaster"""
193 def _call_false(*args):
194 """Fake call_hooks_runner function which returns False."""
198 def _call_nodes_false(node_list, hpath, phase, env):
199 """Fake call_hooks_runner function.
201 @rtype: dict of node -> L{rpc.RpcResult} with an rpc error
202 @return: rpc failure from all nodes
205 return dict([(node, rpc.RpcResult('error', failed=True,
206 node=node, call='FakeError')) for node in node_list])
209 def _call_script_fail(node_list, hpath, phase, env):
210 """Fake call_hooks_runner function.
212 @rtype: dict of node -> L{rpc.RpcResult} with a failed script result
213 @return: script execution failure from all nodes
216 return dict([(node, rpc.RpcResult([("utest", constants.HKR_FAIL, "err")],
217 node=node, call='FakeScriptFail')) for node in node_list])
220 def _call_script_succeed(node_list, hpath, phase, env):
221 """Fake call_hooks_runner function.
223 @rtype: dict of node -> L{rpc.RpcResult} with a successful script result
224 @return: script execution from all nodes
227 return dict([(node, rpc.RpcResult([("utest", constants.HKR_SUCCESS, "ok")],
228 node=node, call='FakeScriptOk')) for node in node_list])
231 self.op = opcodes.OpCode()
232 self.context = FakeContext()
233 # WARNING: here we pass None as RpcRunner instance since we know
234 # our usage via HooksMaster will not use lu.rpc
235 self.lu = FakeLU(FakeProc(), self.op, self.context, None)
237 def testTotalFalse(self):
238 """Test complete rpc failure"""
239 hm = mcpu.HooksMaster(self._call_false, FakeProc(), self.lu)
240 self.failUnlessRaises(errors.HooksFailure,
241 hm.RunPhase, constants.HOOKS_PHASE_PRE)
242 hm.RunPhase(constants.HOOKS_PHASE_POST)
244 def testIndividualFalse(self):
245 """Test individual node failure"""
246 hm = mcpu.HooksMaster(self._call_nodes_false, FakeProc(), self.lu)
247 hm.RunPhase(constants.HOOKS_PHASE_PRE)
248 #self.failUnlessRaises(errors.HooksFailure,
249 # hm.RunPhase, constants.HOOKS_PHASE_PRE)
250 hm.RunPhase(constants.HOOKS_PHASE_POST)
252 def testScriptFalse(self):
253 """Test individual rpc failure"""
254 hm = mcpu.HooksMaster(self._call_script_fail, FakeProc(), self.lu)
255 self.failUnlessRaises(errors.HooksAbort,
256 hm.RunPhase, constants.HOOKS_PHASE_PRE)
257 hm.RunPhase(constants.HOOKS_PHASE_POST)
259 def testScriptSucceed(self):
260 """Test individual rpc failure"""
261 hm = mcpu.HooksMaster(self._call_script_succeed, FakeProc(), self.lu)
262 for phase in (constants.HOOKS_PHASE_PRE, constants.HOOKS_PHASE_POST):
265 if __name__ == '__main__':