1 # Copyright 2013 The Chromium Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 """Unittests for test_dispatcher.py.""" 6 # pylint: disable=R0201 7 # pylint: disable=W0212 8 9 import os 10 import sys 11 import unittest 12 13 sys.path.append(os.path.join(os.path.dirname(os.path.realpath(__file__)), 14 os.pardir, os.pardir)) 15 16 # Mock out android_commands.GetAttachedDevices(). 17 from pylib import android_commands 18 android_commands.GetAttachedDevices = lambda: ['0', '1'] 19 from pylib import constants 20 from pylib.base import base_test_result 21 from pylib.base import test_dispatcher 22 from pylib.utils import watchdog_timer 23 24 25 class TestException(Exception): 26 pass 27 28 29 class MockRunner(object): 30 """A mock TestRunner.""" 31 def __init__(self, device='0', shard_index=0): 32 self.device_serial = device 33 self.shard_index = shard_index 34 self.setups = 0 35 self.teardowns = 0 36 37 def RunTest(self, test): 38 results = base_test_result.TestRunResults() 39 results.AddResult( 40 base_test_result.BaseTestResult(test, base_test_result.ResultType.PASS)) 41 return (results, None) 42 43 def SetUp(self): 44 self.setups += 1 45 46 def TearDown(self): 47 self.teardowns += 1 48 49 50 class MockRunnerFail(MockRunner): 51 def RunTest(self, test): 52 results = base_test_result.TestRunResults() 53 results.AddResult( 54 base_test_result.BaseTestResult(test, base_test_result.ResultType.FAIL)) 55 return (results, test) 56 57 58 class MockRunnerFailTwice(MockRunner): 59 def __init__(self, device='0', shard_index=0): 60 super(MockRunnerFailTwice, self).__init__(device, shard_index) 61 self._fails = 0 62 63 def RunTest(self, test): 64 self._fails += 1 65 results = base_test_result.TestRunResults() 66 if self._fails <= 2: 67 results.AddResult(base_test_result.BaseTestResult( 68 test, base_test_result.ResultType.FAIL)) 69 return (results, test) 70 else: 71 results.AddResult(base_test_result.BaseTestResult( 72 test, base_test_result.ResultType.PASS)) 73 return (results, None) 74 75 76 class MockRunnerException(MockRunner): 77 def RunTest(self, test): 78 raise TestException 79 80 81 class TestFunctions(unittest.TestCase): 82 """Tests test_dispatcher._RunTestsFromQueue.""" 83 @staticmethod 84 def _RunTests(mock_runner, tests): 85 results = [] 86 tests = test_dispatcher._TestCollection( 87 [test_dispatcher._Test(t) for t in tests]) 88 test_dispatcher._RunTestsFromQueue(mock_runner, tests, results, 89 watchdog_timer.WatchdogTimer(None), 2) 90 run_results = base_test_result.TestRunResults() 91 for r in results: 92 run_results.AddTestRunResults(r) 93 return run_results 94 95 def testRunTestsFromQueue(self): 96 results = TestFunctions._RunTests(MockRunner(), ['a', 'b']) 97 self.assertEqual(len(results.GetPass()), 2) 98 self.assertEqual(len(results.GetNotPass()), 0) 99 100 def testRunTestsFromQueueRetry(self): 101 results = TestFunctions._RunTests(MockRunnerFail(), ['a', 'b']) 102 self.assertEqual(len(results.GetPass()), 0) 103 self.assertEqual(len(results.GetFail()), 2) 104 105 def testRunTestsFromQueueFailTwice(self): 106 results = TestFunctions._RunTests(MockRunnerFailTwice(), ['a', 'b']) 107 self.assertEqual(len(results.GetPass()), 2) 108 self.assertEqual(len(results.GetNotPass()), 0) 109 110 def testSetUp(self): 111 runners = [] 112 counter = test_dispatcher._ThreadSafeCounter() 113 test_dispatcher._SetUp(MockRunner, '0', runners, counter) 114 self.assertEqual(len(runners), 1) 115 self.assertEqual(runners[0].setups, 1) 116 117 def testThreadSafeCounter(self): 118 counter = test_dispatcher._ThreadSafeCounter() 119 for i in xrange(5): 120 self.assertEqual(counter.GetAndIncrement(), i) 121 122 def testApplyMaxPerRun(self): 123 self.assertEqual( 124 ['A:B', 'C:D', 'E', 'F:G', 'H:I'], 125 test_dispatcher.ApplyMaxPerRun(['A:B', 'C:D:E', 'F:G:H:I'], 2)) 126 127 128 class TestThreadGroupFunctions(unittest.TestCase): 129 """Tests test_dispatcher._RunAllTests and test_dispatcher._CreateRunners.""" 130 def setUp(self): 131 self.tests = ['a', 'b', 'c', 'd', 'e', 'f', 'g'] 132 shared_test_collection = test_dispatcher._TestCollection( 133 [test_dispatcher._Test(t) for t in self.tests]) 134 self.test_collection_factory = lambda: shared_test_collection 135 136 def testCreate(self): 137 runners = test_dispatcher._CreateRunners(MockRunner, ['0', '1']) 138 for runner in runners: 139 self.assertEqual(runner.setups, 1) 140 self.assertEqual(set([r.device_serial for r in runners]), 141 set(['0', '1'])) 142 self.assertEqual(set([r.shard_index for r in runners]), 143 set([0, 1])) 144 145 def testRun(self): 146 runners = [MockRunner('0'), MockRunner('1')] 147 results, exit_code = test_dispatcher._RunAllTests( 148 runners, self.test_collection_factory, 0) 149 self.assertEqual(len(results.GetPass()), len(self.tests)) 150 self.assertEqual(exit_code, 0) 151 152 def testTearDown(self): 153 runners = [MockRunner('0'), MockRunner('1')] 154 test_dispatcher._TearDownRunners(runners) 155 for runner in runners: 156 self.assertEqual(runner.teardowns, 1) 157 158 def testRetry(self): 159 runners = test_dispatcher._CreateRunners(MockRunnerFail, ['0', '1']) 160 results, exit_code = test_dispatcher._RunAllTests( 161 runners, self.test_collection_factory, 0) 162 self.assertEqual(len(results.GetFail()), len(self.tests)) 163 self.assertEqual(exit_code, constants.ERROR_EXIT_CODE) 164 165 def testReraise(self): 166 runners = test_dispatcher._CreateRunners(MockRunnerException, ['0', '1']) 167 with self.assertRaises(TestException): 168 test_dispatcher._RunAllTests(runners, self.test_collection_factory, 0) 169 170 171 class TestShard(unittest.TestCase): 172 """Tests test_dispatcher.RunTests with sharding.""" 173 @staticmethod 174 def _RunShard(runner_factory): 175 return test_dispatcher.RunTests( 176 ['a', 'b', 'c'], runner_factory, ['0', '1'], shard=True) 177 178 def testShard(self): 179 results, exit_code = TestShard._RunShard(MockRunner) 180 self.assertEqual(len(results.GetPass()), 3) 181 self.assertEqual(exit_code, 0) 182 183 def testFailing(self): 184 results, exit_code = TestShard._RunShard(MockRunnerFail) 185 self.assertEqual(len(results.GetPass()), 0) 186 self.assertEqual(len(results.GetFail()), 3) 187 self.assertEqual(exit_code, constants.ERROR_EXIT_CODE) 188 189 def testNoTests(self): 190 results, exit_code = test_dispatcher.RunTests( 191 [], MockRunner, ['0', '1'], shard=True) 192 self.assertEqual(len(results.GetAll()), 0) 193 self.assertEqual(exit_code, constants.ERROR_EXIT_CODE) 194 195 196 class TestReplicate(unittest.TestCase): 197 """Tests test_dispatcher.RunTests with replication.""" 198 @staticmethod 199 def _RunReplicate(runner_factory): 200 return test_dispatcher.RunTests( 201 ['a', 'b', 'c'], runner_factory, ['0', '1'], shard=False) 202 203 def testReplicate(self): 204 results, exit_code = TestReplicate._RunReplicate(MockRunner) 205 # We expect 6 results since each test should have been run on every device 206 self.assertEqual(len(results.GetPass()), 6) 207 self.assertEqual(exit_code, 0) 208 209 def testFailing(self): 210 results, exit_code = TestReplicate._RunReplicate(MockRunnerFail) 211 self.assertEqual(len(results.GetPass()), 0) 212 self.assertEqual(len(results.GetFail()), 6) 213 self.assertEqual(exit_code, constants.ERROR_EXIT_CODE) 214 215 def testNoTests(self): 216 results, exit_code = test_dispatcher.RunTests( 217 [], MockRunner, ['0', '1'], shard=False) 218 self.assertEqual(len(results.GetAll()), 0) 219 self.assertEqual(exit_code, constants.ERROR_EXIT_CODE) 220 221 222 if __name__ == '__main__': 223 unittest.main() 224
