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)
40 def _RaiseRetryAgain():
41 raise utils.RetryAgain()
44 def _RaiseRetryAgainWithArg(args):
45 raise utils.RetryAgain(*args)
47 def _WrongNestedLoop(self):
48 return utils.Retry(self._RaiseRetryAgain, 0.01, 0.02)
50 def _RetryAndSucceed(self, retries):
51 if self.retries < retries:
53 raise utils.RetryAgain()
57 def _SimpleRetryAndSucceed(self, retries):
59 if self.retries < retries:
65 def testRaiseTimeout(self):
66 self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
67 self._RaiseRetryAgain, 0.01, 0.02)
68 self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
69 self._RetryAndSucceed, 0.01, 0, args=[1])
70 self.failUnlessEqual(self.retries, 1)
72 def testComplete(self):
73 self.failUnlessEqual(utils.Retry(lambda: True, 0, 1), True)
74 self.failUnlessEqual(utils.Retry(self._RetryAndSucceed, 0, 1, args=[2]),
76 self.failUnlessEqual(self.retries, 2)
78 def testNestedLoop(self):
80 self.failUnlessRaises(errors.ProgrammerError, utils.Retry,
81 self._WrongNestedLoop, 0, 1)
82 except utils.RetryTimeout:
83 self.fail("Didn't detect inner loop's exception")
85 def testTimeoutArgument(self):
86 retry_arg="my_important_debugging_message"
88 utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02, args=[[retry_arg]])
89 except utils.RetryTimeout, err:
90 self.failUnlessEqual(err.args, (retry_arg, ))
92 self.fail("Expected timeout didn't happen")
94 def testRaiseInnerWithExc(self):
95 retry_arg="my_important_debugging_message"
98 utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
99 args=[[errors.GenericError(retry_arg, retry_arg)]])
100 except utils.RetryTimeout, err:
103 self.fail("Expected timeout didn't happen")
104 except errors.GenericError, err:
105 self.failUnlessEqual(err.args, (retry_arg, retry_arg))
107 self.fail("Expected GenericError didn't happen")
109 def testRaiseInnerWithMsg(self):
110 retry_arg="my_important_debugging_message"
113 utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
114 args=[[retry_arg, retry_arg]])
115 except utils.RetryTimeout, err:
118 self.fail("Expected timeout didn't happen")
119 except utils.RetryTimeout, err:
120 self.failUnlessEqual(err.args, (retry_arg, retry_arg))
122 self.fail("Expected RetryTimeout didn't happen")
124 def testSimpleRetry(self):
125 self.assertFalse(utils.SimpleRetry(True, lambda: False, 0.01, 0.02))
126 self.assertFalse(utils.SimpleRetry(lambda x: x, lambda: False, 0.01, 0.02))
127 self.assertTrue(utils.SimpleRetry(True, lambda: True, 0, 1))
128 self.assertTrue(utils.SimpleRetry(lambda x: x, lambda: True, 0, 1))
129 self.assertTrue(utils.SimpleRetry(True, self._SimpleRetryAndSucceed,
131 self.assertEqual(self.retries, 1)
132 self.assertEqual(self.called, 2)
133 self.called = self.retries = 0
134 self.assertTrue(utils.SimpleRetry(True, self._SimpleRetryAndSucceed,
136 self.assertEqual(self.called, 3)
139 if __name__ == "__main__":
140 testutils.GanetiTestProgram()