[Android] Switch to new interfaces of GetAVDs and RestartAdbServer.

build/android/pylib/device/device_utils_test.py

+# Copyright 2014 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.
+
+"""
+Unit tests for the contents of device_utils.py (mostly DeviceUtils).
+"""
+# pylint: disable=W0212
+# pylint: disable=W0613
+
+import time
+import unittest
+from pylib.device import adb_wrapper
+from pylib.device import device_utils
+
+
+class DeviceUtilsTest(unittest.TestCase):
+
+  _decorated_function_called_count = 0
+
+  def testFunctionTimesOut(self):
+    """ Tests that RunWithTimeoutAndRetries handles the timeout logic for
+        functions.
+    """
+    def alwaysTimesOut(timeout=None, retries=None):
+      DeviceUtilsTest._decorated_function_called_count += 1
+      time.sleep(100 * timeout)
+
+    start_time = time.time()
+    with self.assertRaises(adb_wrapper.CommandTimeoutError):
+      device_utils.RunWithTimeoutAndRetries(alwaysTimesOut, timeout=1,
+                                            retries=0)
+    elapsed_time = time.time() - start_time
+    self.assertTrue(elapsed_time >= 1)
+
+  def testFunctionRetries(self):
+    """ Tests that RunWithTimeoutAndRetries handles the retries logic for
+        functions.
+    """
+
+    DeviceUtilsTest._decorated_function_called_count = 0
+    def alwaysRaisesCommandFailedError(timeout=None, retries=None):
+      DeviceUtilsTest._decorated_function_called_count += 1
+      raise adb_wrapper.CommandFailedError(['testCommand'],
+                                           'testCommand failed')
+    with self.assertRaises(adb_wrapper.CommandFailedError):
+      device_utils.RunWithTimeoutAndRetries(alwaysRaisesCommandFailedError,
+                                            timeout=100, retries=10)
+    self.assertEquals(11, DeviceUtilsTest._decorated_function_called_count)
+
+  def testMethodTimesOut(self):
+    """ Tests that RunWithTimeoutAndRetries handles the timeout logic for
+        methods.
+    """
+    class TestObject(object):
+      def __init__(self, test_case):
+        self._test_case = test_case
+        self._default_timeout = device_utils._DEFAULT_TIMEOUT
+        self._default_retries = device_utils._DEFAULT_RETRIES
+        self.called = 0
+
+      def alwaysTimesOut(self, timeout=None, retries=None):
+        timeout = timeout if timeout is not None else self._default_timeout
+        retries = retries if retries is not None else self._default_retries
+        device_utils.RunWithTimeoutAndRetries(
+            type(self)._alwaysTimesOutImpl, self, timeout=timeout,
+            retries=retries)
+
+      def _alwaysTimesOutImpl(self, timeout=None, retries=None):
+        self.called += 1
+        time.sleep(100 * timeout)
+        self._test_case.assertFalse(True, msg='Failed to time out?')
+
+    test_obj = TestObject(self)
+    start_time = time.time()
+    with self.assertRaises(adb_wrapper.CommandTimeoutError):
+      test_obj.alwaysTimesOut(timeout=1, retries=0)
+    elapsed_time = time.time() - start_time
+    self.assertTrue(elapsed_time >= 1)
+    self.assertEquals(1, test_obj.called)
+
+  def testMethodRetries(self):
+    """ Tests that RunWithTimeoutAndRetries handles the retries logic for
+        methods.
+    """
+    class TestObject(object):
+      def __init__(self, test_case):
+        self._test_case = test_case
+        self._default_timeout = device_utils._DEFAULT_TIMEOUT
+        self._default_retries = device_utils._DEFAULT_RETRIES
+        self.called = 0
+
+      def alwaysRaisesCommandFailedError(self, timeout=None, retries=None):
+        timeout = timeout if timeout is not None else self._default_timeout
+        retries = retries if retries is not None else self._default_retries
+        device_utils.RunWithTimeoutAndRetries(
+            type(self)._alwaysRaisesCommandFailedErrorImpl,
+            self, timeout=timeout, retries=retries)
+
+      def _alwaysRaisesCommandFailedErrorImpl(self, timeout=None, retries=None):
+        self.called += 1
+        raise adb_wrapper.CommandFailedError(['testCommand'],
+                                             'testCommand failed')
+
+    test_obj = TestObject(self)
+    with self.assertRaises(adb_wrapper.CommandFailedError):
+      test_obj.alwaysRaisesCommandFailedError(retries=10)
+    self.assertEquals(11, test_obj.called)
+
+
+if __name__ == '__main__':
+  unittest.main()
+