Home | History | Annotate | Download | only in common_lib
      1 #!/usr/bin/python
      2 #pylint: disable-msg=C0111
      3 """Unit Tests for autotest.client.common_lib.test"""
      4 
      5 __author__ = 'gps (at] google.com (Gregory P. Smith)'
      6 
      7 import unittest
      8 import common
      9 from autotest_lib.client.common_lib import test
     10 from autotest_lib.client.common_lib.test_utils import mock
     11 from autotest_lib.client.common_lib import error as common_lib_error
     12 
     13 class TestTestCase(unittest.TestCase):
     14     class _neutered_base_test(test.base_test):
     15         """A child class of base_test to avoid calling the constructor."""
     16         def __init__(self, *args, **kwargs):
     17             class MockJob(object):
     18                 pass
     19             class MockProfilerManager(object):
     20                 def active(self):
     21                     return False
     22                 def present(self):
     23                     return True
     24             self.job = MockJob()
     25             self.job.default_profile_only = False
     26             self.job.profilers = MockProfilerManager()
     27             self.job.test_retry = 0
     28             self._new_keyval = False
     29             self.iteration = 0
     30             self.before_iteration_hooks = []
     31             self.after_iteration_hooks = []
     32 
     33 
     34     def setUp(self):
     35         self.god = mock.mock_god()
     36         self.test = self._neutered_base_test()
     37 
     38 
     39     def tearDown(self):
     40         self.god.unstub_all()
     41 
     42 
     43 
     44 class Test_base_test_execute(TestTestCase):
     45     # Test the various behaviors of the base_test.execute() method.
     46     def setUp(self):
     47         TestTestCase.setUp(self)
     48         self.god.stub_function(self.test, 'run_once_profiling')
     49         self.god.stub_function(self.test, 'postprocess')
     50         self.god.stub_function(self.test, 'process_failed_constraints')
     51 
     52 
     53     def test_call_run_once(self):
     54         # setup
     55         self.god.stub_function(self.test, 'drop_caches_between_iterations')
     56         self.god.stub_function(self.test, 'run_once')
     57         self.god.stub_function(self.test, 'postprocess_iteration')
     58         self.god.stub_function(self.test, 'analyze_perf_constraints')
     59         before_hook = self.god.create_mock_function('before_hook')
     60         after_hook = self.god.create_mock_function('after_hook')
     61         self.test.register_before_iteration_hook(before_hook)
     62         self.test.register_after_iteration_hook(after_hook)
     63 
     64         # tests the test._call_run_once implementation
     65         self.test.drop_caches_between_iterations.expect_call()
     66         before_hook.expect_call(self.test)
     67         self.test.run_once.expect_call(1, 2, arg='val')
     68         self.test.postprocess_iteration.expect_call()
     69         self.test.analyze_perf_constraints.expect_call([])
     70         after_hook.expect_call(self.test)
     71         self.test._call_run_once([], False, None, (1, 2), {'arg': 'val'})
     72         self.god.check_playback()
     73 
     74 
     75     def test_call_run_once_with_exception(self):
     76         # setup
     77         self.god.stub_function(self.test, 'drop_caches_between_iterations')
     78         self.god.stub_function(self.test, 'run_once')
     79         before_hook = self.god.create_mock_function('before_hook')
     80         after_hook = self.god.create_mock_function('after_hook')
     81         self.test.register_before_iteration_hook(before_hook)
     82         self.test.register_after_iteration_hook(after_hook)
     83         error = Exception('fail')
     84 
     85         # tests the test._call_run_once implementation
     86         self.test.drop_caches_between_iterations.expect_call()
     87         before_hook.expect_call(self.test)
     88         self.test.run_once.expect_call(1, 2, arg='val').and_raises(error)
     89         after_hook.expect_call(self.test)
     90         try:
     91             self.test._call_run_once([], False, None, (1, 2), {'arg': 'val'})
     92         except:
     93             pass
     94         self.god.check_playback()
     95 
     96 
     97     def _setup_failed_test_calls(self, fail_count, error):
     98         """
     99         Set up failed test calls for use with call_run_once_with_retry.
    100 
    101         @param fail_count: The amount of times to mock a failure.
    102         @param error: The error to raise while failing.
    103         """
    104         self.god.stub_function(self.test.job, 'record')
    105         self.god.stub_function(self.test, '_call_run_once')
    106         # tests the test._call_run_once implementation
    107         for run in xrange(0, fail_count):
    108             self.test._call_run_once.expect_call([], False, None, (1, 2),
    109                                                  {'arg': 'val'}).and_raises(
    110                                                           error)
    111             info_str = 'Run %s failed with %s' % (run, error)
    112             # On the final run we do not emit this message.
    113             if run != self.test.job.test_retry and isinstance(error,
    114                                                common_lib_error.TestFailRetry):
    115                 self.test.job.record.expect_call('INFO', None, None, info_str)
    116 
    117 
    118     def test_call_run_once_with_retry_exception(self):
    119         """
    120         Test call_run_once_with_retry duplicating a test that will always fail.
    121         """
    122         self.test.job.test_retry = 5
    123         self.god.stub_function(self.test, 'drop_caches_between_iterations')
    124         self.god.stub_function(self.test, 'run_once')
    125         before_hook = self.god.create_mock_function('before_hook')
    126         after_hook = self.god.create_mock_function('after_hook')
    127         self.test.register_before_iteration_hook(before_hook)
    128         self.test.register_after_iteration_hook(after_hook)
    129         error = common_lib_error.TestFailRetry('fail')
    130         self._setup_failed_test_calls(self.test.job.test_retry+1, error)
    131         try:
    132             self.test._call_run_once_with_retry([], False, None, (1, 2),
    133                                                 {'arg': 'val'})
    134         except Exception as err:
    135             if err != error:
    136                 raise
    137         self.god.check_playback()
    138 
    139 
    140     def test_call_run_once_with_retry_exception_unretryable(self):
    141         """
    142         Test call_run_once_with_retry duplicating a test that will always fail
    143         with a non-retryable exception.
    144         """
    145         self.test.job.test_retry = 5
    146         self.god.stub_function(self.test, 'drop_caches_between_iterations')
    147         self.god.stub_function(self.test, 'run_once')
    148         before_hook = self.god.create_mock_function('before_hook')
    149         after_hook = self.god.create_mock_function('after_hook')
    150         self.test.register_before_iteration_hook(before_hook)
    151         self.test.register_after_iteration_hook(after_hook)
    152         error = common_lib_error.TestFail('fail')
    153         self._setup_failed_test_calls(1, error)
    154         try:
    155             self.test._call_run_once_with_retry([], False, None, (1, 2),
    156                                                 {'arg': 'val'})
    157         except Exception as err:
    158             if err != error:
    159                 raise
    160         self.god.check_playback()
    161 
    162 
    163     def test_call_run_once_with_retry_exception_and_pass(self):
    164         """
    165         Test call_run_once_with_retry duplicating a test that fails at first
    166         and later passes.
    167         """
    168         # Stubbed out for the write_keyval call.
    169         self.test.outputdir = '/tmp'
    170         self.test.job._tap = None
    171 
    172         num_to_fail = 2
    173         self.test.job.test_retry = 5
    174         self.god.stub_function(self.test, 'drop_caches_between_iterations')
    175         self.god.stub_function(self.test, 'run_once')
    176         before_hook = self.god.create_mock_function('before_hook')
    177         after_hook = self.god.create_mock_function('after_hook')
    178         self.god.stub_function(self.test, '_call_run_once')
    179         self.test.register_before_iteration_hook(before_hook)
    180         self.test.register_after_iteration_hook(after_hook)
    181         self.god.stub_function(self.test.job, 'record')
    182         # tests the test._call_run_once implementation
    183         error = common_lib_error.TestFailRetry('fail')
    184         self._setup_failed_test_calls(num_to_fail, error)
    185         # Passing call
    186         self.test._call_run_once.expect_call([], False, None, (1, 2),
    187                                              {'arg': 'val'})
    188         self.test._call_run_once_with_retry([], False, None, (1, 2),
    189                                             {'arg': 'val'})
    190         self.god.check_playback()
    191 
    192 
    193     def _expect_call_run_once(self):
    194         self.test._call_run_once.expect_call((), False, None, (), {})
    195 
    196 
    197     def test_execute_test_length(self):
    198         # test that test_length overrides iterations and works.
    199         self.god.stub_function(self.test, '_call_run_once')
    200 
    201         self._expect_call_run_once()
    202         self._expect_call_run_once()
    203         self._expect_call_run_once()
    204         self.test.run_once_profiling.expect_call(None)
    205         self.test.postprocess.expect_call()
    206         self.test.process_failed_constraints.expect_call()
    207 
    208         fake_time = iter(xrange(4)).next
    209         self.test.execute(iterations=1, test_length=3, _get_time=fake_time)
    210         self.god.check_playback()
    211 
    212 
    213     def test_execute_iterations(self):
    214         # test that iterations works.
    215         self.god.stub_function(self.test, '_call_run_once')
    216 
    217         iterations = 2
    218         for _ in range(iterations):
    219             self._expect_call_run_once()
    220         self.test.run_once_profiling.expect_call(None)
    221         self.test.postprocess.expect_call()
    222         self.test.process_failed_constraints.expect_call()
    223 
    224         self.test.execute(iterations=iterations)
    225         self.god.check_playback()
    226 
    227 
    228     def _mock_calls_for_execute_no_iterations(self):
    229         self.test.run_once_profiling.expect_call(None)
    230         self.test.postprocess.expect_call()
    231         self.test.process_failed_constraints.expect_call()
    232 
    233 
    234     def test_execute_iteration_zero(self):
    235         # test that iterations=0 works.
    236         self._mock_calls_for_execute_no_iterations()
    237 
    238         self.test.execute(iterations=0)
    239         self.god.check_playback()
    240 
    241 
    242     def test_execute_profile_only(self):
    243         # test that profile_only=True works.
    244         self.god.stub_function(self.test, 'drop_caches_between_iterations')
    245         self.test.drop_caches_between_iterations.expect_call()
    246         self.test.run_once_profiling.expect_call(None)
    247         self.test.drop_caches_between_iterations.expect_call()
    248         self.test.run_once_profiling.expect_call(None)
    249         self.test.postprocess.expect_call()
    250         self.test.process_failed_constraints.expect_call()
    251         self.test.execute(profile_only=True, iterations=2)
    252         self.god.check_playback()
    253 
    254 
    255     def test_execute_default_profile_only(self):
    256         # test that profile_only=True works.
    257         self.god.stub_function(self.test, 'drop_caches_between_iterations')
    258         for _ in xrange(3):
    259             self.test.drop_caches_between_iterations.expect_call()
    260             self.test.run_once_profiling.expect_call(None)
    261         self.test.postprocess.expect_call()
    262         self.test.process_failed_constraints.expect_call()
    263         self.test.job.default_profile_only = True
    264         self.test.execute(iterations=3)
    265         self.god.check_playback()
    266 
    267 
    268     def test_execute_postprocess_profiled_false(self):
    269         # test that postprocess_profiled_run=False works
    270         self.god.stub_function(self.test, '_call_run_once')
    271 
    272         self.test._call_run_once.expect_call((), False, False, (), {})
    273         self.test.run_once_profiling.expect_call(False)
    274         self.test.postprocess.expect_call()
    275         self.test.process_failed_constraints.expect_call()
    276 
    277         self.test.execute(postprocess_profiled_run=False, iterations=1)
    278         self.god.check_playback()
    279 
    280 
    281     def test_execute_postprocess_profiled_true(self):
    282         # test that postprocess_profiled_run=True works
    283         self.god.stub_function(self.test, '_call_run_once')
    284 
    285         self.test._call_run_once.expect_call((), False, True, (), {})
    286         self.test.run_once_profiling.expect_call(True)
    287         self.test.postprocess.expect_call()
    288         self.test.process_failed_constraints.expect_call()
    289 
    290         self.test.execute(postprocess_profiled_run=True, iterations=1)
    291         self.god.check_playback()
    292 
    293 
    294 if __name__ == '__main__':
    295     unittest.main()
    296