blob: ad457362f7b0da16af86ca032bdf1ef6885870b1 [file] [log] [blame]
# 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.assertEqual(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.assertEqual(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.assertEqual(expected_timeout, actual_timeout)
self.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(11, DecoratorsTest._decorated_function_called_count)
DecoratorsTest._decorated_function_called_count = 0
with self.assertRaises(device_errors.CommandFailedError):
alwaysRaisesCommandFailedError(retries=5)
self.assertEqual(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.assertEqual(30, alwaysReturnsTimeouts())
self.assertEqual(120, alwaysReturnsTimeouts(timeout=120))
@decorators.WithTimeoutAndRetriesDefaults(30, 10)
def alwaysReturnsRetries(timeout=None, retries=None):
return retries
self.assertEqual(10, alwaysReturnsRetries())
self.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(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.assertEqual(
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.assertEqual(42, test_obj.alwaysReturnsTimeout())
self.assertEqual(41, test_obj.alwaysReturnsTimeout(timeout=41))
self.assertEqual(31, test_obj.alwaysReturnsRetries())
self.assertEqual(32, test_obj.alwaysReturnsRetries(retries=32))
def testMethodDecoratorUsesMiniumumTimeout(self):
test_obj = self._MethodDecoratorTestObject(
self, default_timeout=42, default_retries=31)
self.assertEqual(100, test_obj.alwaysReturnsTimeoutWithMin())
self.assertEqual(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.assertEqual(exception_desc, str(e.exception))
if __name__ == '__main__':
unittest.main(verbosity=2)