1 #!/usr/bin/python 2 # Copyright 2013 The Chromium Authors. All rights reserved. 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 6 """Unittests for timeout_and_retry.py.""" 7 8 import logging 9 import time 10 import unittest 11 12 from devil.utils import reraiser_thread 13 from devil.utils import timeout_retry 14 15 16 _DEFAULT_TIMEOUT = .1 17 18 19 class TestException(Exception): 20 pass 21 22 23 def _CountTries(tries): 24 tries[0] += 1 25 raise TestException 26 27 28 class TestRun(unittest.TestCase): 29 """Tests for timeout_retry.Run.""" 30 31 def testRun(self): 32 self.assertTrue(timeout_retry.Run( 33 lambda x: x, 30, 3, [True], {})) 34 35 def testTimeout(self): 36 tries = [0] 37 38 def _sleep(): 39 tries[0] += 1 40 time.sleep(1) 41 42 self.assertRaises( 43 reraiser_thread.TimeoutError, timeout_retry.Run, _sleep, .01, 1, 44 error_log_func=logging.debug) 45 self.assertEqual(tries[0], 2) 46 47 def testRetries(self): 48 tries = [0] 49 self.assertRaises( 50 TestException, timeout_retry.Run, lambda: _CountTries(tries), 51 _DEFAULT_TIMEOUT, 3, error_log_func=logging.debug) 52 self.assertEqual(tries[0], 4) 53 54 def testNoRetries(self): 55 tries = [0] 56 self.assertRaises( 57 TestException, timeout_retry.Run, lambda: _CountTries(tries), 58 _DEFAULT_TIMEOUT, 0, error_log_func=logging.debug) 59 self.assertEqual(tries[0], 1) 60 61 def testReturnValue(self): 62 self.assertTrue(timeout_retry.Run(lambda: True, _DEFAULT_TIMEOUT, 3)) 63 64 def testCurrentTimeoutThreadGroup(self): 65 def InnerFunc(): 66 current_thread_group = timeout_retry.CurrentTimeoutThreadGroup() 67 self.assertIsNotNone(current_thread_group) 68 69 def InnerInnerFunc(): 70 self.assertEqual(current_thread_group, 71 timeout_retry.CurrentTimeoutThreadGroup()) 72 return True 73 return reraiser_thread.RunAsync((InnerInnerFunc,))[0] 74 75 self.assertTrue(timeout_retry.Run(InnerFunc, _DEFAULT_TIMEOUT, 3)) 76 77 78 if __name__ == '__main__': 79 unittest.main() 80
