xref: /external/chromium_org/build/android/pylib/device/decorators_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 decorators.py.
"""

# pylint: disable=W0613

import os
import sys
import time
import traceback
import unittest

from pylib import constants
from pylib.device import decorators
from pylib.device import device_errors
from pylib.utils import reraiser_thread

# TODO(jbudorick) Remove once the DeviceUtils implementations are no longer
#                 backed by AndroidCommands / android_testrunner.
sys.path.append(os.path.join(constants.DIR_SOURCE_ROOT, 'third_party',
                             'android_testrunner'))
import errors as old_errors

_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 testFunctionDecoratorTranslatesOldExceptions(self):
    """Tests that the explicit decorator translates old exceptions."""
    @decorators.WithTimeoutAndRetries
    def alwaysRaisesProvidedException(exception, timeout=None, retries=None):
      raise exception

    exception_desc = 'Old response timeout error'
    with self.assertRaises(device_errors.CommandTimeoutError) as e:
      alwaysRaisesProvidedException(
          old_errors.WaitForResponseTimedOutError(exception_desc),
          timeout=10, retries=1)
    self.assertEquals(exception_desc, str(e.exception))

    exception_desc = 'Old device error'
    with self.assertRaises(device_errors.DeviceUnreachableError) as e:
      alwaysRaisesProvidedException(
          old_errors.DeviceUnresponsiveError(exception_desc),
          timeout=10, retries=1)
    self.assertEquals(exception_desc, str(e.exception))

  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 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 testDefaultsFunctionDecoratorTranslatesOldExceptions(self):
    """Tests that the explicit decorator translates old exceptions."""
    @decorators.WithTimeoutAndRetriesDefaults(30, 10)
    def alwaysRaisesProvidedException(exception, timeout=None, retries=None):
      raise exception

    exception_desc = 'Old response timeout error'
    with self.assertRaises(device_errors.CommandTimeoutError) as e:
      alwaysRaisesProvidedException(
          old_errors.WaitForResponseTimedOutError(exception_desc))
    self.assertEquals(exception_desc, str(e.exception))

    exception_desc = 'Old device error'
    with self.assertRaises(device_errors.DeviceUnreachableError) as e:
      alwaysRaisesProvidedException(
          old_errors.DeviceUnresponsiveError(exception_desc))
    self.assertEquals(exception_desc, str(e.exception))

  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 testExplicitDecoratorTranslatesOldExceptions(self):
    """Tests that the explicit decorator translates old exceptions."""
    @decorators.WithExplicitTimeoutAndRetries(30, 10)
    def alwaysRaisesProvidedException(exception):
      raise exception

    exception_desc = 'Old response timeout error'
    with self.assertRaises(device_errors.CommandTimeoutError) as e:
      alwaysRaisesProvidedException(
          old_errors.WaitForResponseTimedOutError(exception_desc))
    self.assertEquals(exception_desc, str(e.exception))

    exception_desc = 'Old device error'
    with self.assertRaises(device_errors.DeviceUnreachableError) as e:
      alwaysRaisesProvidedException(
          old_errors.DeviceUnresponsiveError(exception_desc))
    self.assertEquals(exception_desc, str(e.exception))

  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=R0201

    @decorators.WithTimeoutAndRetriesFromInstance(
        'default_timeout', 'default_retries')
    def alwaysReturnsTimeout(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=R0201


  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 testMethodDecoratorTranslatesOldExceptions(self):
    test_obj = self._MethodDecoratorTestObject(self)

    exception_desc = 'Old response timeout error'
    with self.assertRaises(device_errors.CommandTimeoutError) as e:
      test_obj.alwaysRaisesProvidedException(
          old_errors.WaitForResponseTimedOutError(exception_desc))
    self.assertEquals(exception_desc, str(e.exception))

    exception_desc = 'Old device error'
    with self.assertRaises(device_errors.DeviceUnreachableError) as e:
      test_obj.alwaysRaisesProvidedException(
          old_errors.DeviceUnresponsiveError(exception_desc))
    self.assertEquals(exception_desc, str(e.exception))

  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)