OLD | NEW |
| (Empty) |
1 # Copyright 2014 The Chromium Authors. All rights reserved. | |
2 # Use of this source code is governed by a BSD-style license that can be | |
3 # found in the LICENSE file. | |
4 | |
5 """ | |
6 Unit tests for decorators.py. | |
7 """ | |
8 | |
9 # pylint: disable=W0613 | |
10 | |
11 import os | |
12 import sys | |
13 import time | |
14 import traceback | |
15 import unittest | |
16 | |
17 from pylib import constants | |
18 from pylib.device import decorators | |
19 from pylib.device import device_errors | |
20 from pylib.utils import reraiser_thread | |
21 | |
22 _DEFAULT_TIMEOUT = 30 | |
23 _DEFAULT_RETRIES = 3 | |
24 | |
25 class DecoratorsTest(unittest.TestCase): | |
26 _decorated_function_called_count = 0 | |
27 | |
28 def testFunctionDecoratorDoesTimeouts(self): | |
29 """Tests that the base decorator handles the timeout logic.""" | |
30 DecoratorsTest._decorated_function_called_count = 0 | |
31 @decorators.WithTimeoutAndRetries | |
32 def alwaysTimesOut(timeout=None, retries=None): | |
33 DecoratorsTest._decorated_function_called_count += 1 | |
34 time.sleep(100) | |
35 | |
36 start_time = time.time() | |
37 with self.assertRaises(device_errors.CommandTimeoutError): | |
38 alwaysTimesOut(timeout=1, retries=0) | |
39 elapsed_time = time.time() - start_time | |
40 self.assertTrue(elapsed_time >= 1) | |
41 self.assertEquals(1, DecoratorsTest._decorated_function_called_count) | |
42 | |
43 def testFunctionDecoratorDoesRetries(self): | |
44 """Tests that the base decorator handles the retries logic.""" | |
45 DecoratorsTest._decorated_function_called_count = 0 | |
46 @decorators.WithTimeoutAndRetries | |
47 def alwaysRaisesCommandFailedError(timeout=None, retries=None): | |
48 DecoratorsTest._decorated_function_called_count += 1 | |
49 raise device_errors.CommandFailedError('testCommand failed') | |
50 | |
51 with self.assertRaises(device_errors.CommandFailedError): | |
52 alwaysRaisesCommandFailedError(timeout=30, retries=10) | |
53 self.assertEquals(11, DecoratorsTest._decorated_function_called_count) | |
54 | |
55 def testFunctionDecoratorRequiresParams(self): | |
56 """Tests that the base decorator requires timeout and retries params.""" | |
57 @decorators.WithTimeoutAndRetries | |
58 def requiresExplicitTimeoutAndRetries(timeout=None, retries=None): | |
59 return (timeout, retries) | |
60 | |
61 with self.assertRaises(KeyError): | |
62 requiresExplicitTimeoutAndRetries() | |
63 with self.assertRaises(KeyError): | |
64 requiresExplicitTimeoutAndRetries(timeout=10) | |
65 with self.assertRaises(KeyError): | |
66 requiresExplicitTimeoutAndRetries(retries=0) | |
67 expected_timeout = 10 | |
68 expected_retries = 1 | |
69 (actual_timeout, actual_retries) = ( | |
70 requiresExplicitTimeoutAndRetries(timeout=expected_timeout, | |
71 retries=expected_retries)) | |
72 self.assertEquals(expected_timeout, actual_timeout) | |
73 self.assertEquals(expected_retries, actual_retries) | |
74 | |
75 def testFunctionDecoratorTranslatesReraiserExceptions(self): | |
76 """Tests that the explicit decorator translates reraiser exceptions.""" | |
77 @decorators.WithTimeoutAndRetries | |
78 def alwaysRaisesProvidedException(exception, timeout=None, retries=None): | |
79 raise exception | |
80 | |
81 exception_desc = 'Reraiser thread timeout error' | |
82 with self.assertRaises(device_errors.CommandTimeoutError) as e: | |
83 alwaysRaisesProvidedException( | |
84 reraiser_thread.TimeoutError(exception_desc), | |
85 timeout=10, retries=1) | |
86 self.assertEquals(exception_desc, str(e.exception)) | |
87 | |
88 def testDefaultsFunctionDecoratorDoesTimeouts(self): | |
89 """Tests that the defaults decorator handles timeout logic.""" | |
90 DecoratorsTest._decorated_function_called_count = 0 | |
91 @decorators.WithTimeoutAndRetriesDefaults(1, 0) | |
92 def alwaysTimesOut(timeout=None, retries=None): | |
93 DecoratorsTest._decorated_function_called_count += 1 | |
94 time.sleep(100) | |
95 | |
96 start_time = time.time() | |
97 with self.assertRaises(device_errors.CommandTimeoutError): | |
98 alwaysTimesOut() | |
99 elapsed_time = time.time() - start_time | |
100 self.assertTrue(elapsed_time >= 1) | |
101 self.assertEquals(1, DecoratorsTest._decorated_function_called_count) | |
102 | |
103 DecoratorsTest._decorated_function_called_count = 0 | |
104 with self.assertRaises(device_errors.CommandTimeoutError): | |
105 alwaysTimesOut(timeout=2) | |
106 elapsed_time = time.time() - start_time | |
107 self.assertTrue(elapsed_time >= 2) | |
108 self.assertEquals(1, DecoratorsTest._decorated_function_called_count) | |
109 | |
110 def testDefaultsFunctionDecoratorDoesRetries(self): | |
111 """Tests that the defaults decorator handles retries logic.""" | |
112 DecoratorsTest._decorated_function_called_count = 0 | |
113 @decorators.WithTimeoutAndRetriesDefaults(30, 10) | |
114 def alwaysRaisesCommandFailedError(timeout=None, retries=None): | |
115 DecoratorsTest._decorated_function_called_count += 1 | |
116 raise device_errors.CommandFailedError('testCommand failed') | |
117 | |
118 with self.assertRaises(device_errors.CommandFailedError): | |
119 alwaysRaisesCommandFailedError() | |
120 self.assertEquals(11, DecoratorsTest._decorated_function_called_count) | |
121 | |
122 DecoratorsTest._decorated_function_called_count = 0 | |
123 with self.assertRaises(device_errors.CommandFailedError): | |
124 alwaysRaisesCommandFailedError(retries=5) | |
125 self.assertEquals(6, DecoratorsTest._decorated_function_called_count) | |
126 | |
127 def testDefaultsFunctionDecoratorPassesValues(self): | |
128 """Tests that the defaults decorator passes timeout and retries kwargs.""" | |
129 @decorators.WithTimeoutAndRetriesDefaults(30, 10) | |
130 def alwaysReturnsTimeouts(timeout=None, retries=None): | |
131 return timeout | |
132 | |
133 self.assertEquals(30, alwaysReturnsTimeouts()) | |
134 self.assertEquals(120, alwaysReturnsTimeouts(timeout=120)) | |
135 | |
136 @decorators.WithTimeoutAndRetriesDefaults(30, 10) | |
137 def alwaysReturnsRetries(timeout=None, retries=None): | |
138 return retries | |
139 | |
140 self.assertEquals(10, alwaysReturnsRetries()) | |
141 self.assertEquals(1, alwaysReturnsRetries(retries=1)) | |
142 | |
143 def testDefaultsFunctionDecoratorTranslatesReraiserExceptions(self): | |
144 """Tests that the explicit decorator translates reraiser exceptions.""" | |
145 @decorators.WithTimeoutAndRetriesDefaults(30, 10) | |
146 def alwaysRaisesProvidedException(exception, timeout=None, retries=None): | |
147 raise exception | |
148 | |
149 exception_desc = 'Reraiser thread timeout error' | |
150 with self.assertRaises(device_errors.CommandTimeoutError) as e: | |
151 alwaysRaisesProvidedException( | |
152 reraiser_thread.TimeoutError(exception_desc)) | |
153 self.assertEquals(exception_desc, str(e.exception)) | |
154 | |
155 def testExplicitFunctionDecoratorDoesTimeouts(self): | |
156 """Tests that the explicit decorator handles timeout logic.""" | |
157 DecoratorsTest._decorated_function_called_count = 0 | |
158 @decorators.WithExplicitTimeoutAndRetries(1, 0) | |
159 def alwaysTimesOut(): | |
160 DecoratorsTest._decorated_function_called_count += 1 | |
161 time.sleep(100) | |
162 | |
163 start_time = time.time() | |
164 with self.assertRaises(device_errors.CommandTimeoutError): | |
165 alwaysTimesOut() | |
166 elapsed_time = time.time() - start_time | |
167 self.assertTrue(elapsed_time >= 1) | |
168 self.assertEquals(1, DecoratorsTest._decorated_function_called_count) | |
169 | |
170 def testExplicitFunctionDecoratorDoesRetries(self): | |
171 """Tests that the explicit decorator handles retries logic.""" | |
172 DecoratorsTest._decorated_function_called_count = 0 | |
173 @decorators.WithExplicitTimeoutAndRetries(30, 10) | |
174 def alwaysRaisesCommandFailedError(): | |
175 DecoratorsTest._decorated_function_called_count += 1 | |
176 raise device_errors.CommandFailedError('testCommand failed') | |
177 | |
178 with self.assertRaises(device_errors.CommandFailedError): | |
179 alwaysRaisesCommandFailedError() | |
180 self.assertEquals(11, DecoratorsTest._decorated_function_called_count) | |
181 | |
182 def testExplicitDecoratorTranslatesReraiserExceptions(self): | |
183 """Tests that the explicit decorator translates reraiser exceptions.""" | |
184 @decorators.WithExplicitTimeoutAndRetries(30, 10) | |
185 def alwaysRaisesProvidedException(exception): | |
186 raise exception | |
187 | |
188 exception_desc = 'Reraiser thread timeout error' | |
189 with self.assertRaises(device_errors.CommandTimeoutError) as e: | |
190 alwaysRaisesProvidedException( | |
191 reraiser_thread.TimeoutError(exception_desc)) | |
192 self.assertEquals(exception_desc, str(e.exception)) | |
193 | |
194 class _MethodDecoratorTestObject(object): | |
195 """An object suitable for testing the method decorator.""" | |
196 | |
197 def __init__(self, test_case, default_timeout=_DEFAULT_TIMEOUT, | |
198 default_retries=_DEFAULT_RETRIES): | |
199 self._test_case = test_case | |
200 self.default_timeout = default_timeout | |
201 self.default_retries = default_retries | |
202 self.function_call_counters = { | |
203 'alwaysRaisesCommandFailedError': 0, | |
204 'alwaysTimesOut': 0, | |
205 'requiresExplicitTimeoutAndRetries': 0, | |
206 } | |
207 | |
208 @decorators.WithTimeoutAndRetriesFromInstance( | |
209 'default_timeout', 'default_retries') | |
210 def alwaysTimesOut(self, timeout=None, retries=None): | |
211 self.function_call_counters['alwaysTimesOut'] += 1 | |
212 time.sleep(100) | |
213 self._test_case.assertFalse(True, msg='Failed to time out?') | |
214 | |
215 @decorators.WithTimeoutAndRetriesFromInstance( | |
216 'default_timeout', 'default_retries') | |
217 def alwaysRaisesCommandFailedError(self, timeout=None, retries=None): | |
218 self.function_call_counters['alwaysRaisesCommandFailedError'] += 1 | |
219 raise device_errors.CommandFailedError('testCommand failed') | |
220 | |
221 # pylint: disable=no-self-use | |
222 | |
223 @decorators.WithTimeoutAndRetriesFromInstance( | |
224 'default_timeout', 'default_retries') | |
225 def alwaysReturnsTimeout(self, timeout=None, retries=None): | |
226 return timeout | |
227 | |
228 @decorators.WithTimeoutAndRetriesFromInstance( | |
229 'default_timeout', 'default_retries') | |
230 def alwaysReturnsRetries(self, timeout=None, retries=None): | |
231 return retries | |
232 | |
233 @decorators.WithTimeoutAndRetriesFromInstance( | |
234 'default_timeout', 'default_retries') | |
235 def alwaysRaisesProvidedException(self, exception, timeout=None, | |
236 retries=None): | |
237 raise exception | |
238 | |
239 # pylint: enable=no-self-use | |
240 | |
241 | |
242 def testMethodDecoratorDoesTimeout(self): | |
243 """Tests that the method decorator handles timeout logic.""" | |
244 test_obj = self._MethodDecoratorTestObject(self) | |
245 start_time = time.time() | |
246 with self.assertRaises(device_errors.CommandTimeoutError): | |
247 try: | |
248 test_obj.alwaysTimesOut(timeout=1, retries=0) | |
249 except: | |
250 traceback.print_exc() | |
251 raise | |
252 elapsed_time = time.time() - start_time | |
253 self.assertTrue(elapsed_time >= 1) | |
254 self.assertEquals(1, test_obj.function_call_counters['alwaysTimesOut']) | |
255 | |
256 def testMethodDecoratorDoesRetries(self): | |
257 """Tests that the method decorator handles retries logic.""" | |
258 test_obj = self._MethodDecoratorTestObject(self) | |
259 with self.assertRaises(device_errors.CommandFailedError): | |
260 try: | |
261 test_obj.alwaysRaisesCommandFailedError(retries=10) | |
262 except: | |
263 traceback.print_exc() | |
264 raise | |
265 self.assertEquals( | |
266 11, test_obj.function_call_counters['alwaysRaisesCommandFailedError']) | |
267 | |
268 def testMethodDecoratorPassesValues(self): | |
269 """Tests that the method decorator passes timeout and retries kwargs.""" | |
270 test_obj = self._MethodDecoratorTestObject( | |
271 self, default_timeout=42, default_retries=31) | |
272 self.assertEquals(42, test_obj.alwaysReturnsTimeout()) | |
273 self.assertEquals(41, test_obj.alwaysReturnsTimeout(timeout=41)) | |
274 self.assertEquals(31, test_obj.alwaysReturnsRetries()) | |
275 self.assertEquals(32, test_obj.alwaysReturnsRetries(retries=32)) | |
276 | |
277 def testMethodDecoratorTranslatesReraiserExceptions(self): | |
278 test_obj = self._MethodDecoratorTestObject(self) | |
279 | |
280 exception_desc = 'Reraiser thread timeout error' | |
281 with self.assertRaises(device_errors.CommandTimeoutError) as e: | |
282 test_obj.alwaysRaisesProvidedException( | |
283 reraiser_thread.TimeoutError(exception_desc)) | |
284 self.assertEquals(exception_desc, str(e.exception)) | |
285 | |
286 if __name__ == '__main__': | |
287 unittest.main(verbosity=2) | |
288 | |
OLD | NEW |