#!/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()