Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
#!/usr/bin/python
#
# Copyright (C) 2011 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
"""Script for testing ganeti.utils.retry"""
import unittest
from ganeti import constants
from ganeti import errors
from ganeti import utils
import testutils
class TestRetry(testutils.GanetiTestCase):
def setUp(self):
testutils.GanetiTestCase.setUp(self)
self.retries = 0
@staticmethod
def _RaiseRetryAgain():
raise utils.RetryAgain()
@staticmethod
def _RaiseRetryAgainWithArg(args):
raise utils.RetryAgain(*args)
def _WrongNestedLoop(self):
return utils.Retry(self._RaiseRetryAgain, 0.01, 0.02)
def _RetryAndSucceed(self, retries):
if self.retries < retries:
self.retries += 1
raise utils.RetryAgain()
else:
return True
def testRaiseTimeout(self):
self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
self._RaiseRetryAgain, 0.01, 0.02)
self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
self._RetryAndSucceed, 0.01, 0, args=[1])
self.failUnlessEqual(self.retries, 1)
def testComplete(self):
self.failUnlessEqual(utils.Retry(lambda: True, 0, 1), True)
self.failUnlessEqual(utils.Retry(self._RetryAndSucceed, 0, 1, args=[2]),
True)
self.failUnlessEqual(self.retries, 2)
def testNestedLoop(self):
try:
self.failUnlessRaises(errors.ProgrammerError, utils.Retry,
self._WrongNestedLoop, 0, 1)
except utils.RetryTimeout:
self.fail("Didn't detect inner loop's exception")
def testTimeoutArgument(self):
retry_arg="my_important_debugging_message"
try:
utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02, args=[[retry_arg]])
except utils.RetryTimeout, err:
self.failUnlessEqual(err.args, (retry_arg, ))
else:
self.fail("Expected timeout didn't happen")
def testRaiseInnerWithExc(self):
retry_arg="my_important_debugging_message"
try:
try:
utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
args=[[errors.GenericError(retry_arg, retry_arg)]])
except utils.RetryTimeout, err:
err.RaiseInner()
else:
self.fail("Expected timeout didn't happen")
except errors.GenericError, err:
self.failUnlessEqual(err.args, (retry_arg, retry_arg))
else:
self.fail("Expected GenericError didn't happen")
def testRaiseInnerWithMsg(self):
retry_arg="my_important_debugging_message"
try:
try:
utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
args=[[retry_arg, retry_arg]])
except utils.RetryTimeout, err:
err.RaiseInner()
else:
self.fail("Expected timeout didn't happen")
except utils.RetryTimeout, err:
self.failUnlessEqual(err.args, (retry_arg, retry_arg))
else:
self.fail("Expected RetryTimeout didn't happen")
if __name__ == "__main__":
testutils.GanetiTestProgram()