from pylib.utils import watchdog_timer
-
class TestException(Exception):
pass
class MockRunner(object):
"""A mock TestRunner."""
def __init__(self, device='0', shard_index=0):
- self.device = device
+ self.device_serial = device
self.shard_index = shard_index
self.setups = 0
self.teardowns = 0
for i in xrange(5):
self.assertEqual(counter.GetAndIncrement(), i)
+ def testApplyMaxPerRun(self):
+ self.assertEqual(
+ ['A:B', 'C:D', 'E', 'F:G', 'H:I'],
+ test_dispatcher.ApplyMaxPerRun(['A:B', 'C:D:E', 'F:G:H:I'], 2))
+
class TestThreadGroupFunctions(unittest.TestCase):
"""Tests test_dispatcher._RunAllTests and test_dispatcher._CreateRunners."""
runners = test_dispatcher._CreateRunners(MockRunner, ['0', '1'])
for runner in runners:
self.assertEqual(runner.setups, 1)
- self.assertEqual(set([r.device for r in runners]),
+ self.assertEqual(set([r.device_serial for r in runners]),
set(['0', '1']))
self.assertEqual(set([r.shard_index for r in runners]),
set([0, 1]))
self.assertEqual(len(results.GetAll()), 0)
self.assertEqual(exit_code, constants.ERROR_EXIT_CODE)
- def testTestsRemainWithAllDevicesOffline(self):
- attached_devices = android_commands.GetAttachedDevices
- android_commands.GetAttachedDevices = lambda: []
- try:
- with self.assertRaises(AssertionError):
- _results, _exit_code = TestShard._RunShard(MockRunner)
- finally:
- android_commands.GetAttachedDevices = attached_devices
-
class TestReplicate(unittest.TestCase):
"""Tests test_dispatcher.RunTests with replication."""