4 # Copyright (C) 2011 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 testing ganeti.utils.retry"""
26 from ganeti import constants
27 from ganeti import errors
28 from ganeti import utils
33 class TestRetry(testutils.GanetiTestCase):
35 testutils.GanetiTestCase.setUp(self)
39 def _RaiseRetryAgain():
40 raise utils.RetryAgain()
43 def _RaiseRetryAgainWithArg(args):
44 raise utils.RetryAgain(*args)
46 def _WrongNestedLoop(self):
47 return utils.Retry(self._RaiseRetryAgain, 0.01, 0.02)
49 def _RetryAndSucceed(self, retries):
50 if self.retries < retries:
52 raise utils.RetryAgain()
56 def testRaiseTimeout(self):
57 self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
58 self._RaiseRetryAgain, 0.01, 0.02)
59 self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
60 self._RetryAndSucceed, 0.01, 0, args=[1])
61 self.failUnlessEqual(self.retries, 1)
63 def testComplete(self):
64 self.failUnlessEqual(utils.Retry(lambda: True, 0, 1), True)
65 self.failUnlessEqual(utils.Retry(self._RetryAndSucceed, 0, 1, args=[2]),
67 self.failUnlessEqual(self.retries, 2)
69 def testNestedLoop(self):
71 self.failUnlessRaises(errors.ProgrammerError, utils.Retry,
72 self._WrongNestedLoop, 0, 1)
73 except utils.RetryTimeout:
74 self.fail("Didn't detect inner loop's exception")
76 def testTimeoutArgument(self):
77 retry_arg="my_important_debugging_message"
79 utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02, args=[[retry_arg]])
80 except utils.RetryTimeout, err:
81 self.failUnlessEqual(err.args, (retry_arg, ))
83 self.fail("Expected timeout didn't happen")
85 def testRaiseInnerWithExc(self):
86 retry_arg="my_important_debugging_message"
89 utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
90 args=[[errors.GenericError(retry_arg, retry_arg)]])
91 except utils.RetryTimeout, err:
94 self.fail("Expected timeout didn't happen")
95 except errors.GenericError, err:
96 self.failUnlessEqual(err.args, (retry_arg, retry_arg))
98 self.fail("Expected GenericError didn't happen")
100 def testRaiseInnerWithMsg(self):
101 retry_arg="my_important_debugging_message"
104 utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
105 args=[[retry_arg, retry_arg]])
106 except utils.RetryTimeout, err:
109 self.fail("Expected timeout didn't happen")
110 except utils.RetryTimeout, err:
111 self.failUnlessEqual(err.args, (retry_arg, retry_arg))
113 self.fail("Expected RetryTimeout didn't happen")
116 if __name__ == "__main__":
117 testutils.GanetiTestProgram()