295 lines
12 KiB
Python
Executable file
295 lines
12 KiB
Python
Executable file
#!/usr/bin/python
|
|
#pylint: disable-msg=C0111
|
|
"""Unit Tests for autotest.client.common_lib.test"""
|
|
|
|
__author__ = 'gps@google.com (Gregory P. Smith)'
|
|
|
|
import unittest
|
|
import common
|
|
from autotest_lib.client.common_lib import test
|
|
from autotest_lib.client.common_lib.test_utils import mock
|
|
from autotest_lib.client.common_lib import error as common_lib_error
|
|
|
|
class TestTestCase(unittest.TestCase):
|
|
class _neutered_base_test(test.base_test):
|
|
"""A child class of base_test to avoid calling the constructor."""
|
|
def __init__(self, *args, **kwargs):
|
|
class MockJob(object):
|
|
pass
|
|
class MockProfilerManager(object):
|
|
def active(self):
|
|
return False
|
|
def present(self):
|
|
return True
|
|
self.job = MockJob()
|
|
self.job.default_profile_only = False
|
|
self.job.profilers = MockProfilerManager()
|
|
self.job.test_retry = 0
|
|
self._new_keyval = False
|
|
self.iteration = 0
|
|
self.before_iteration_hooks = []
|
|
self.after_iteration_hooks = []
|
|
|
|
|
|
def setUp(self):
|
|
self.god = mock.mock_god()
|
|
self.test = self._neutered_base_test()
|
|
|
|
|
|
def tearDown(self):
|
|
self.god.unstub_all()
|
|
|
|
|
|
|
|
class Test_base_test_execute(TestTestCase):
|
|
# Test the various behaviors of the base_test.execute() method.
|
|
def setUp(self):
|
|
TestTestCase.setUp(self)
|
|
self.god.stub_function(self.test, 'run_once_profiling')
|
|
self.god.stub_function(self.test, 'postprocess')
|
|
self.god.stub_function(self.test, 'process_failed_constraints')
|
|
|
|
|
|
def test_call_run_once(self):
|
|
# setup
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
self.god.stub_function(self.test, 'run_once')
|
|
self.god.stub_function(self.test, 'postprocess_iteration')
|
|
self.god.stub_function(self.test, 'analyze_perf_constraints')
|
|
before_hook = self.god.create_mock_function('before_hook')
|
|
after_hook = self.god.create_mock_function('after_hook')
|
|
self.test.register_before_iteration_hook(before_hook)
|
|
self.test.register_after_iteration_hook(after_hook)
|
|
|
|
# tests the test._call_run_once implementation
|
|
self.test.drop_caches_between_iterations.expect_call()
|
|
before_hook.expect_call(self.test)
|
|
self.test.run_once.expect_call(1, 2, arg='val')
|
|
self.test.postprocess_iteration.expect_call()
|
|
self.test.analyze_perf_constraints.expect_call([])
|
|
after_hook.expect_call(self.test)
|
|
self.test._call_run_once([], False, None, (1, 2), {'arg': 'val'})
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_call_run_once_with_exception(self):
|
|
# setup
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
self.god.stub_function(self.test, 'run_once')
|
|
before_hook = self.god.create_mock_function('before_hook')
|
|
after_hook = self.god.create_mock_function('after_hook')
|
|
self.test.register_before_iteration_hook(before_hook)
|
|
self.test.register_after_iteration_hook(after_hook)
|
|
error = Exception('fail')
|
|
|
|
# tests the test._call_run_once implementation
|
|
self.test.drop_caches_between_iterations.expect_call()
|
|
before_hook.expect_call(self.test)
|
|
self.test.run_once.expect_call(1, 2, arg='val').and_raises(error)
|
|
after_hook.expect_call(self.test)
|
|
try:
|
|
self.test._call_run_once([], False, None, (1, 2), {'arg': 'val'})
|
|
except:
|
|
pass
|
|
self.god.check_playback()
|
|
|
|
|
|
def _setup_failed_test_calls(self, fail_count, error):
|
|
"""
|
|
Set up failed test calls for use with call_run_once_with_retry.
|
|
|
|
@param fail_count: The amount of times to mock a failure.
|
|
@param error: The error to raise while failing.
|
|
"""
|
|
self.god.stub_function(self.test.job, 'record')
|
|
self.god.stub_function(self.test, '_call_run_once')
|
|
# tests the test._call_run_once implementation
|
|
for run in xrange(0, fail_count):
|
|
self.test._call_run_once.expect_call([], False, None, (1, 2),
|
|
{'arg': 'val'}).and_raises(
|
|
error)
|
|
info_str = 'Run %s failed with %s' % (run, error)
|
|
# On the final run we do not emit this message.
|
|
if run != self.test.job.test_retry and isinstance(error,
|
|
common_lib_error.TestFailRetry):
|
|
self.test.job.record.expect_call('INFO', None, None, info_str)
|
|
|
|
|
|
def test_call_run_once_with_retry_exception(self):
|
|
"""
|
|
Test call_run_once_with_retry duplicating a test that will always fail.
|
|
"""
|
|
self.test.job.test_retry = 5
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
self.god.stub_function(self.test, 'run_once')
|
|
before_hook = self.god.create_mock_function('before_hook')
|
|
after_hook = self.god.create_mock_function('after_hook')
|
|
self.test.register_before_iteration_hook(before_hook)
|
|
self.test.register_after_iteration_hook(after_hook)
|
|
error = common_lib_error.TestFailRetry('fail')
|
|
self._setup_failed_test_calls(self.test.job.test_retry+1, error)
|
|
try:
|
|
self.test._call_run_once_with_retry([], False, None, (1, 2),
|
|
{'arg': 'val'})
|
|
except Exception as err:
|
|
if err != error:
|
|
raise
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_call_run_once_with_retry_exception_unretryable(self):
|
|
"""
|
|
Test call_run_once_with_retry duplicating a test that will always fail
|
|
with a non-retryable exception.
|
|
"""
|
|
self.test.job.test_retry = 5
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
self.god.stub_function(self.test, 'run_once')
|
|
before_hook = self.god.create_mock_function('before_hook')
|
|
after_hook = self.god.create_mock_function('after_hook')
|
|
self.test.register_before_iteration_hook(before_hook)
|
|
self.test.register_after_iteration_hook(after_hook)
|
|
error = common_lib_error.TestFail('fail')
|
|
self._setup_failed_test_calls(1, error)
|
|
try:
|
|
self.test._call_run_once_with_retry([], False, None, (1, 2),
|
|
{'arg': 'val'})
|
|
except Exception as err:
|
|
if err != error:
|
|
raise
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_call_run_once_with_retry_exception_and_pass(self):
|
|
"""
|
|
Test call_run_once_with_retry duplicating a test that fails at first
|
|
and later passes.
|
|
"""
|
|
# Stubbed out for the write_keyval call.
|
|
self.test.outputdir = '/tmp'
|
|
self.test.job._tap = None
|
|
|
|
num_to_fail = 2
|
|
self.test.job.test_retry = 5
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
self.god.stub_function(self.test, 'run_once')
|
|
before_hook = self.god.create_mock_function('before_hook')
|
|
after_hook = self.god.create_mock_function('after_hook')
|
|
self.god.stub_function(self.test, '_call_run_once')
|
|
self.test.register_before_iteration_hook(before_hook)
|
|
self.test.register_after_iteration_hook(after_hook)
|
|
self.god.stub_function(self.test.job, 'record')
|
|
# tests the test._call_run_once implementation
|
|
error = common_lib_error.TestFailRetry('fail')
|
|
self._setup_failed_test_calls(num_to_fail, error)
|
|
# Passing call
|
|
self.test._call_run_once.expect_call([], False, None, (1, 2),
|
|
{'arg': 'val'})
|
|
self.test._call_run_once_with_retry([], False, None, (1, 2),
|
|
{'arg': 'val'})
|
|
self.god.check_playback()
|
|
|
|
|
|
def _expect_call_run_once(self):
|
|
self.test._call_run_once.expect_call((), False, None, (), {})
|
|
|
|
|
|
def test_execute_test_length(self):
|
|
# test that test_length overrides iterations and works.
|
|
self.god.stub_function(self.test, '_call_run_once')
|
|
|
|
self._expect_call_run_once()
|
|
self._expect_call_run_once()
|
|
self._expect_call_run_once()
|
|
self.test.run_once_profiling.expect_call(None)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
|
|
fake_time = iter(xrange(4)).next
|
|
self.test.execute(iterations=1, test_length=3, _get_time=fake_time)
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_execute_iterations(self):
|
|
# test that iterations works.
|
|
self.god.stub_function(self.test, '_call_run_once')
|
|
|
|
iterations = 2
|
|
for _ in range(iterations):
|
|
self._expect_call_run_once()
|
|
self.test.run_once_profiling.expect_call(None)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
|
|
self.test.execute(iterations=iterations)
|
|
self.god.check_playback()
|
|
|
|
|
|
def _mock_calls_for_execute_no_iterations(self):
|
|
self.test.run_once_profiling.expect_call(None)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
|
|
|
|
def test_execute_iteration_zero(self):
|
|
# test that iterations=0 works.
|
|
self._mock_calls_for_execute_no_iterations()
|
|
|
|
self.test.execute(iterations=0)
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_execute_profile_only(self):
|
|
# test that profile_only=True works.
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
self.test.drop_caches_between_iterations.expect_call()
|
|
self.test.run_once_profiling.expect_call(None)
|
|
self.test.drop_caches_between_iterations.expect_call()
|
|
self.test.run_once_profiling.expect_call(None)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
self.test.execute(profile_only=True, iterations=2)
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_execute_default_profile_only(self):
|
|
# test that profile_only=True works.
|
|
self.god.stub_function(self.test, 'drop_caches_between_iterations')
|
|
for _ in xrange(3):
|
|
self.test.drop_caches_between_iterations.expect_call()
|
|
self.test.run_once_profiling.expect_call(None)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
self.test.job.default_profile_only = True
|
|
self.test.execute(iterations=3)
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_execute_postprocess_profiled_false(self):
|
|
# test that postprocess_profiled_run=False works
|
|
self.god.stub_function(self.test, '_call_run_once')
|
|
|
|
self.test._call_run_once.expect_call((), False, False, (), {})
|
|
self.test.run_once_profiling.expect_call(False)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
|
|
self.test.execute(postprocess_profiled_run=False, iterations=1)
|
|
self.god.check_playback()
|
|
|
|
|
|
def test_execute_postprocess_profiled_true(self):
|
|
# test that postprocess_profiled_run=True works
|
|
self.god.stub_function(self.test, '_call_run_once')
|
|
|
|
self.test._call_run_once.expect_call((), False, True, (), {})
|
|
self.test.run_once_profiling.expect_call(True)
|
|
self.test.postprocess.expect_call()
|
|
self.test.process_failed_constraints.expect_call()
|
|
|
|
self.test.execute(postprocess_profiled_run=True, iterations=1)
|
|
self.god.check_playback()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|