text: # 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 decorators.py.
  """
  
  # pylint: disable=W0613
  
  import time
  import traceback
  import unittest
  
  from devil.android import decorators
  from devil.android import device_errors
  from devil.utils import reraiser_thread
  
  _DEFAULT_TIMEOUT = 30
  _DEFAULT_RETRIES = 3
  
  
  class DecoratorsTest(unittest.TestCase):
    _decorated_function_called_count = 0
  
    def testFunctionDecoratorDoesTimeouts(self):
      """Tests that the base decorator handles the timeout logic."""
      DecoratorsTest._decorated_function_called_count = 0
  
      @decorators.WithTimeoutAndRetries
      def alwaysTimesOut(timeout=None, retries=None):
        DecoratorsTest._decorated_function_called_count += 1
        time.sleep(100)
  
      start_time = time.time()
      with self.assertRaises(device_errors.CommandTimeoutError):
        alwaysTimesOut(timeout=1, retries=0)
      elapsed_time = time.time() - start_time
      self.assertTrue(elapsed_time >= 1)
      self.assertEquals(1, DecoratorsTest._decorated_function_called_count)
  
    def testFunctionDecoratorDoesRetries(self):
      """Tests that the base decorator handles the retries logic."""
      DecoratorsTest._decorated_function_called_count = 0
  
      @decorators.WithTimeoutAndRetries
      def alwaysRaisesCommandFailedError(timeout=None, retries=None):
        DecoratorsTest._decorated_function_called_count += 1
        raise device_errors.CommandFailedError('testCommand failed')
  
      with self.assertRaises(device_errors.CommandFailedError):
        alwaysRaisesCommandFailedError(timeout=30, retries=10)
      self.assertEquals(11, DecoratorsTest._decorated_function_called_count)
  
    def testFunctionDecoratorRequiresParams(self):
      """Tests that the base decorator requires timeout and retries params."""
  
      @decorators.WithTimeoutAndRetries
      def requiresExplicitTimeoutAndRetries(timeout=None, retries=None):
        return (timeout, retries)
  
      with self.assertRaises(KeyError):
        requiresExplicitTimeoutAndRetries()
      with self.assertRaises(KeyError):
        requiresExplicitTimeoutAndRetries(timeout=10)
      with self.assertRaises(KeyError):
        requiresExplicitTimeoutAndRetries(retries=0)
      expected_timeout = 10
      expected_retries = 1
      (actual_timeout, actual_retries) = (requiresExplicitTimeoutAndRetries(
          timeout=expected_timeout, retries=expected_retries))
      self.assertEquals(expected_timeout, actual_timeout)
      self.assertEquals(expected_retries, actual_retries)
  
    def testFunctionDecoratorTranslatesReraiserExceptions(self):
      """Tests that the explicit decorator translates reraiser exceptions."""
  
      @decorators.WithTimeoutAndRetries
      def alwaysRaisesProvidedException(exception, timeout=None, retries=None):
        raise exception
  
      exception_desc = 'Reraiser thread timeout error'
      with self.assertRaises(device_errors.CommandTimeoutError) as e:
        alwaysRaisesProvidedException(
            reraiser_thread.TimeoutError(exception_desc), timeout=10, retries=1)
      self.assertEquals(exception_desc, str(e.exception))
  
    def testConditionalRetriesDecoratorRetries(self):
      def do_not_retry_no_adb_error(exc):
        return not isinstance(exc, device_errors.NoAdbError)
  
      actual_tries = [0]
  
      @decorators.WithTimeoutAndConditionalRetries(do_not_retry_no_adb_error)
      def alwaysRaisesCommandFailedError(timeout=None, retries=None):
        actual_tries[0] += 1
        raise device_errors.CommandFailedError('Command failed :(')
  
      with self.assertRaises(device_errors.CommandFailedError):
        alwaysRaisesCommandFailedError(timeout=10, retries=10)
      self.assertEquals(11, actual_tries[0])
  
    def testConditionalRetriesDecoratorDoesntRetry(self):
      def do_not_retry_no_adb_error(exc):
        return not isinstance(exc, device_errors.NoAdbError)
  
      actual_tries = [0]
  
      @decorators.WithTimeoutAndConditionalRetries(do_not_retry_no_adb_error)
      def alwaysRaisesNoAdbError(timeout=None, retries=None):
        actual_tries[0] += 1
        raise device_errors.NoAdbError()
  
      with self.assertRaises(device_errors.NoAdbError):
        alwaysRaisesNoAdbError(timeout=10, retries=10)
      self.assertEquals(1, actual_tries[0])
  
    def testDefaultsFunctionDecoratorDoesTimeouts(self):
      """Tests that the defaults decorator handles timeout logic."""
      DecoratorsTest._decorated_function_called_count = 0
  
      @decorators.WithTimeoutAndRetriesDefaults(1, 0)
      def alwaysTimesOut(timeout=None, retries=None):
        DecoratorsTest._decorated_function_called_count += 1
        time.sleep(100)
  
      start_time = time.time()
      with self.assertRaises(device_errors.CommandTimeoutError):
        alwaysTimesOut()
      elapsed_time = time.time() - start_time
      self.assertTrue(elapsed_time >= 1)
      self.assertEquals(1, DecoratorsTest._decorated_function_called_count)
  
      DecoratorsTest._decorated_function_called_count = 0
      with self.assertRaises(device_errors.CommandTimeoutError):
        alwaysTimesOut(timeout=2)
      elapsed_time = time.time() - start_time
      self.assertTrue(elapsed_time >= 2)
      self.assertEquals(1, DecoratorsTest._decorated_function_called_count)
  
    def testDefaultsFunctionDecoratorDoesRetries(self):
      """Tests that the defaults decorator handles retries logic."""
      DecoratorsTest._decorated_function_called_count = 0
  
      @decorators.WithTimeoutAndRetriesDefaults(30, 10)
      def alwaysRaisesCommandFailedError(timeout=None, retries=None):
        DecoratorsTest._decorated_function_called_count += 1
        raise device_errors.CommandFailedError('testCommand failed')
  
      with self.assertRaises(device_errors.CommandFailedError):
        alwaysRaisesCommandFailedError()
      self.assertEquals(11, DecoratorsTest._decorated_function_called_count)
  
      DecoratorsTest._decorated_function_called_count = 0
      with self.assertRaises(device_errors.CommandFailedError):
        alwaysRaisesCommandFailedError(retries=5)
      self.assertEquals(6, DecoratorsTest._decorated_function_called_count)
  
    def testDefaultsFunctionDecoratorPassesValues(self):
      """Tests that the defaults decorator passes timeout and retries kwargs."""
  
      @decorators.WithTimeoutAndRetriesDefaults(30, 10)
      def alwaysReturnsTimeouts(timeout=None, retries=None):
        return timeout
  
      self.assertEquals(30, alwaysReturnsTimeouts())
      self.assertEquals(120, alwaysReturnsTimeouts(timeout=120))
  
      @decorators.WithTimeoutAndRetriesDefaults(30, 10)
      def alwaysReturnsRetries(timeout=None, retries=None):
        return retries
  
      self.assertEquals(10, alwaysReturnsRetries())
      self.assertEquals(1, alwaysReturnsRetries(retries=1))
  
    def testDefaultsFunctionDecoratorTranslatesReraiserExceptions(self):
      """Tests that the explicit decorator translates reraiser exceptions."""
  
      @decorators.WithTimeoutAndRetriesDefaults(30, 10)
      def alwaysRaisesProvidedException(exception, timeout=None, retries=None):
        raise exception
  
      exception_desc = 'Reraiser thread timeout error'
      with self.assertRaises(device_errors.CommandTimeoutError) as e:
        alwaysRaisesProvidedException(
            reraiser_thread.TimeoutError(exception_desc))
      self.assertEquals(exception_desc, str(e.exception))
  
    def testExplicitFunctionDecoratorDoesTimeouts(self):
      """Tests that the explicit decorator handles timeout logic."""
      DecoratorsTest._decorated_function_called_count = 0
  
      @decorators.WithExplicitTimeoutAndRetries(1, 0)
      def alwaysTimesOut():
        DecoratorsTest._decorated_function_called_count += 1
        time.sleep(100)
  
      start_time = time.time()
      with self.assertRaises(device_errors.CommandTimeoutError):
        alwaysTimesOut()
      elapsed_time = time.time() - start_time
      self.assertTrue(elapsed_time >= 1)
      self.assertEquals(1, DecoratorsTest._decorated_function_called_count)
  
    def testExplicitFunctionDecoratorDoesRetries(self):
      """Tests that the explicit decorator handles retries logic."""
      DecoratorsTest._decorated_function_called_count = 0
  
      @decorators.WithExplicitTimeoutAndRetries(30, 10)
      def alwaysRaisesCommandFailedError():
        DecoratorsTest._decorated_function_called_count += 1
        raise device_errors.CommandFailedError('testCommand failed')
  
      with self.assertRaises(device_errors.CommandFailedError):
        alwaysRaisesCommandFailedError()
      self.assertEquals(11, DecoratorsTest._decorated_function_called_count)
  
    def testExplicitDecoratorTranslatesReraiserExceptions(self):
      """Tests that the explicit decorator translates reraiser exceptions."""
  
      @decorators.WithExplicitTimeoutAndRetries(30, 10)
      def alwaysRaisesProvidedException(exception):
        raise exception
  
      exception_desc = 'Reraiser thread timeout error'
      with self.assertRaises(device_errors.CommandTimeoutError) as e:
        alwaysRaisesProvidedException(
            reraiser_thread.TimeoutError(exception_desc))
      self.assertEquals(exception_desc, str(e.exception))
  
    class _MethodDecoratorTestObject(object):
      """An object suitable for testing the method decorator."""
  
      def __init__(self,
                   test_case,
                   default_timeout=_DEFAULT_TIMEOUT,
                   default_retries=_DEFAULT_RETRIES):
        self._test_case = test_case
        self.default_timeout = default_timeout
        self.default_retries = default_retries
        self.function_call_counters = {
            'alwaysRaisesCommandFailedError': 0,
            'alwaysTimesOut': 0,
            'requiresExplicitTimeoutAndRetries': 0,
        }
  
      @decorators.WithTimeoutAndRetriesFromInstance('default_timeout',
                                                    'default_retries')
      def alwaysTimesOut(self, timeout=None, retries=None):
        self.function_call_counters['alwaysTimesOut'] += 1
        time.sleep(100)
        self._test_case.assertFalse(True, msg='Failed to time out?')
  
      @decorators.WithTimeoutAndRetriesFromInstance('default_timeout',
                                                    'default_retries')
      def alwaysRaisesCommandFailedError(self, timeout=None, retries=None):
        self.function_call_counters['alwaysRaisesCommandFailedError'] += 1
        raise device_errors.CommandFailedError('testCommand failed')
  
      # pylint: disable=no-self-use
  
      @decorators.WithTimeoutAndRetriesFromInstance('default_timeout',
                                                    'default_retries')
      def alwaysReturnsTimeout(self, timeout=None, retries=None):
        return timeout
  
      @decorators.WithTimeoutAndRetriesFromInstance(
          'default_timeout', 'default_retries', min_default_timeout=100)
      def alwaysReturnsTimeoutWithMin(self, timeout=None, retries=None):
        return timeout
  
      @decorators.WithTimeoutAndRetriesFromInstance('default_timeout',
                                                    'default_retries')
      def alwaysReturnsRetries(self, timeout=None, retries=None):
        return retries
  
      @decorators.WithTimeoutAndRetriesFromInstance('default_timeout',
                                                    'default_retries')
      def alwaysRaisesProvidedException(self,
                                        exception,
                                        timeout=None,
                                        retries=None):
        raise exception
  
      # pylint: enable=no-self-use
  
    def testMethodDecoratorDoesTimeout(self):
      """Tests that the method decorator handles timeout logic."""
      test_obj = self._MethodDecoratorTestObject(self)
      start_time = time.time()
      with self.assertRaises(device_errors.CommandTimeoutError):
        try:
          test_obj.alwaysTimesOut(timeout=1, retries=0)
        except:
          traceback.print_exc()
          raise
      elapsed_time = time.time() - start_time
      self.assertTrue(elapsed_time >= 1)
      self.assertEquals(1, test_obj.function_call_counters['alwaysTimesOut'])
  
    def testMethodDecoratorDoesRetries(self):
      """Tests that the method decorator handles retries logic."""
      test_obj = self._MethodDecoratorTestObject(self)
      with self.assertRaises(device_errors.CommandFailedError):
        try:
          test_obj.alwaysRaisesCommandFailedError(retries=10)
        except:
          traceback.print_exc()
          raise
      self.assertEquals(
          11, test_obj.function_call_counters['alwaysRaisesCommandFailedError'])
  
    def testMethodDecoratorPassesValues(self):
      """Tests that the method decorator passes timeout and retries kwargs."""
      test_obj = self._MethodDecoratorTestObject(
          self, default_timeout=42, default_retries=31)
      self.assertEquals(42, test_obj.alwaysReturnsTimeout())
      self.assertEquals(41, test_obj.alwaysReturnsTimeout(timeout=41))
      self.assertEquals(31, test_obj.alwaysReturnsRetries())
      self.assertEquals(32, test_obj.alwaysReturnsRetries(retries=32))
  
    def testMethodDecoratorUsesMiniumumTimeout(self):
      test_obj = self._MethodDecoratorTestObject(
          self, default_timeout=42, default_retries=31)
      self.assertEquals(100, test_obj.alwaysReturnsTimeoutWithMin())
      self.assertEquals(41, test_obj.alwaysReturnsTimeoutWithMin(timeout=41))
  
    def testMethodDecoratorTranslatesReraiserExceptions(self):
      test_obj = self._MethodDecoratorTestObject(self)
  
      exception_desc = 'Reraiser thread timeout error'
      with self.assertRaises(device_errors.CommandTimeoutError) as e:
        test_obj.alwaysRaisesProvidedException(
            reraiser_thread.TimeoutError(exception_desc))
      self.assertEquals(exception_desc, str(e.exception))
  
  
  if __name__ == '__main__':
    unittest.main(verbosity=2)
  