| Index: build/android/devil/utils/timeout_retry_unittest.py
|
| diff --git a/build/android/devil/utils/timeout_retry_unittest.py b/build/android/devil/utils/timeout_retry_unittest.py
|
| deleted file mode 100755
|
| index f2ceef7d6045c29f71818c394b35e3a0d1ef12c1..0000000000000000000000000000000000000000
|
| --- a/build/android/devil/utils/timeout_retry_unittest.py
|
| +++ /dev/null
|
| @@ -1,77 +0,0 @@
|
| -#!/usr/bin/python
|
| -# Copyright 2013 The Chromium Authors. All rights reserved.
|
| -# Use of this source code is governed by a BSD-style license that can be
|
| -# found in the LICENSE file.
|
| -
|
| -"""Unittests for timeout_and_retry.py."""
|
| -
|
| -import logging
|
| -import time
|
| -import unittest
|
| -
|
| -from devil.utils import reraiser_thread
|
| -from devil.utils import timeout_retry
|
| -
|
| -
|
| -_DEFAULT_TIMEOUT = .1
|
| -
|
| -
|
| -class TestException(Exception):
|
| - pass
|
| -
|
| -
|
| -def _CountTries(tries):
|
| - tries[0] += 1
|
| - raise TestException
|
| -
|
| -
|
| -class TestRun(unittest.TestCase):
|
| - """Tests for timeout_retry.Run."""
|
| -
|
| - def testRun(self):
|
| - self.assertTrue(timeout_retry.Run(
|
| - lambda x: x, 30, 3, [True], {}))
|
| -
|
| - def testTimeout(self):
|
| - tries = [0]
|
| - def _sleep():
|
| - tries[0] += 1
|
| - time.sleep(1)
|
| -
|
| - self.assertRaises(
|
| - reraiser_thread.TimeoutError, timeout_retry.Run, _sleep, .0001, 1,
|
| - error_log_func=logging.debug)
|
| - self.assertEqual(tries[0], 2)
|
| -
|
| - def testRetries(self):
|
| - tries = [0]
|
| - self.assertRaises(
|
| - TestException, timeout_retry.Run, lambda: _CountTries(tries),
|
| - _DEFAULT_TIMEOUT, 3, error_log_func=logging.debug)
|
| - self.assertEqual(tries[0], 4)
|
| -
|
| - def testNoRetries(self):
|
| - tries = [0]
|
| - self.assertRaises(
|
| - TestException, timeout_retry.Run, lambda: _CountTries(tries),
|
| - _DEFAULT_TIMEOUT, 0, error_log_func=logging.debug)
|
| - self.assertEqual(tries[0], 1)
|
| -
|
| - def testReturnValue(self):
|
| - self.assertTrue(timeout_retry.Run(lambda: True, _DEFAULT_TIMEOUT, 3))
|
| -
|
| - def testCurrentTimeoutThreadGroup(self):
|
| - def InnerFunc():
|
| - current_thread_group = timeout_retry.CurrentTimeoutThreadGroup()
|
| - self.assertIsNotNone(current_thread_group)
|
| - def InnerInnerFunc():
|
| - self.assertEqual(current_thread_group,
|
| - timeout_retry.CurrentTimeoutThreadGroup())
|
| - return True
|
| - return reraiser_thread.RunAsync((InnerInnerFunc,))[0]
|
| -
|
| - self.assertTrue(timeout_retry.Run(InnerFunc, _DEFAULT_TIMEOUT, 3))
|
| -
|
| -
|
| -if __name__ == '__main__':
|
| - unittest.main()
|
|
|