1 files changed, 79 insertions, 0 deletions
diff --git a/systrace/catapult/devil/devil/utils/timeout_retry_unittest.py b/systrace/catapult/devil/devil/utils/timeout_retry_unittest.py
new file mode 100755
index 0000000..0eeb31a
--- /dev/null
+++ b/systrace/catapult/devil/devil/utils/timeout_retry_unittest.py
@@ -0,0 +1,79 @@
+#!/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, .01, 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()