build/android/pylib/base/test_dispatcher_unittest.py

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