build/android/pylib/base/test_dispatcher_unittest.py

#!/usr/bin/env python
# Copyright 2013 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unittests for test_dispatcher.py."""
# pylint: disable=R0201
# pylint: disable=W0212

import os
import sys
import unittest


from devil.android import device_utils
from devil.android.sdk import adb_wrapper
from devil.utils import watchdog_timer
from pylib import constants
from pylib.base import base_test_result
from pylib.base import test_collection
from pylib.base import test_dispatcher

sys.path.append(
    os.path.join(constants.DIR_SOURCE_ROOT, 'third_party', 'pymock'))
import mock # pylint: disable=import-error


class TestException(Exception):
  pass


def _MockDevice(serial):
  d = mock.MagicMock(spec=device_utils.DeviceUtils)
  d.__str__.return_value = serial
  d.adb = mock.MagicMock(spec=adb_wrapper.AdbWrapper)
  d.adb.GetDeviceSerial = mock.MagicMock(return_value=serial)
  d.IsOnline = mock.MagicMock(return_value=True)
  return d


class MockRunner(object):
  """A mock TestRunner."""
  def __init__(self, device=None, shard_index=0):
    self.device = device or _MockDevice('0')
    self.device_serial = self.device.adb.GetDeviceSerial()
    self.shard_index = shard_index
    self.setups = 0
    self.teardowns = 0

  def RunTest(self, test):
    results = base_test_result.TestRunResults()
    results.AddResult(
        base_test_result.BaseTestResult(test, base_test_result.ResultType.PASS))
    return (results, None)

  def SetUp(self):
    self.setups += 1

  def TearDown(self):
    self.teardowns += 1


class MockRunnerFail(MockRunner):
  def RunTest(self, test):
    results = base_test_result.TestRunResults()
    results.AddResult(
        base_test_result.BaseTestResult(test, base_test_result.ResultType.FAIL))
    return (results, test)


class MockRunnerFailTwice(MockRunner):
  def __init__(self, device=None, shard_index=0):
    super(MockRunnerFailTwice, self).__init__(device, shard_index)
    self._fails = 0

  def RunTest(self, test):
    self._fails += 1
    results = base_test_result.TestRunResults()
    if self._fails <= 2:
      results.AddResult(base_test_result.BaseTestResult(
          test, base_test_result.ResultType.FAIL))
      return (results, test)
    else:
      results.AddResult(base_test_result.BaseTestResult(
          test, base_test_result.ResultType.PASS))
      return (results, None)


class MockRunnerException(MockRunner):
  def RunTest(self, test):
    raise TestException


class TestFunctions(unittest.TestCase):
  """Tests test_dispatcher._RunTestsFromQueue."""
  @staticmethod
  def _RunTests(mock_runner, tests):
    results = []
    tests = test_collection.TestCollection(
        [test_dispatcher._Test(t) for t in tests])
    test_dispatcher._RunTestsFromQueue(mock_runner, tests, results,
                                       watchdog_timer.WatchdogTimer(None), 2)
    run_results = base_test_result.TestRunResults()
    for r in results:
      run_results.AddTestRunResults(r)
    return run_results

  def testRunTestsFromQueue(self):
    results = TestFunctions._RunTests(MockRunner(), ['a', 'b'])
    self.assertEqual(len(results.GetPass()), 2)
    self.assertEqual(len(results.GetNotPass()), 0)

  def testRunTestsFromQueueRetry(self):
    results = TestFunctions._RunTests(MockRunnerFail(), ['a', 'b'])
    self.assertEqual(len(results.GetPass()), 0)
    self.assertEqual(len(results.GetFail()), 2)

  def testRunTestsFromQueueFailTwice(self):
    results = TestFunctions._RunTests(MockRunnerFailTwice(), ['a', 'b'])
    self.assertEqual(len(results.GetPass()), 2)
    self.assertEqual(len(results.GetNotPass()), 0)

  def testSetUp(self):
    runners = []
    counter = test_dispatcher._ThreadSafeCounter()
    test_dispatcher._SetUp(MockRunner, _MockDevice('0'), runners, counter)
    self.assertEqual(len(runners), 1)
    self.assertEqual(runners[0].setups, 1)

  def testThreadSafeCounter(self):
    counter = test_dispatcher._ThreadSafeCounter()
    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."""
  def setUp(self):
    self.tests = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
    shared_test_collection = test_collection.TestCollection(
        [test_dispatcher._Test(t) for t in self.tests])
    self.test_collection_factory = lambda: shared_test_collection

  def testCreate(self):
    runners = test_dispatcher._CreateRunners(
        MockRunner, [_MockDevice('0'), _MockDevice('1')])
    for runner in runners:
      self.assertEqual(runner.setups, 1)
    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]))

  def testRun(self):
    runners = [MockRunner(_MockDevice('0')), MockRunner(_MockDevice('1'))]
    results, exit_code = test_dispatcher._RunAllTests(
        runners, self.test_collection_factory, 0)
    self.assertEqual(len(results.GetPass()), len(self.tests))
    self.assertEqual(exit_code, 0)

  def testTearDown(self):
    runners = [MockRunner(_MockDevice('0')), MockRunner(_MockDevice('1'))]
    test_dispatcher._TearDownRunners(runners)
    for runner in runners:
      self.assertEqual(runner.teardowns, 1)

  def testRetry(self):
    runners = test_dispatcher._CreateRunners(
        MockRunnerFail, [_MockDevice('0'), _MockDevice('1')])
    results, exit_code = test_dispatcher._RunAllTests(
        runners, self.test_collection_factory, 0)
    self.assertEqual(len(results.GetFail()), len(self.tests))
    self.assertEqual(exit_code, constants.ERROR_EXIT_CODE)

  def testReraise(self):
    runners = test_dispatcher._CreateRunners(
        MockRunnerException, [_MockDevice('0'), _MockDevice('1')])
    with self.assertRaises(TestException):
      test_dispatcher._RunAllTests(runners, self.test_collection_factory, 0)


class TestShard(unittest.TestCase):
  """Tests test_dispatcher.RunTests with sharding."""
  @staticmethod
  def _RunShard(runner_factory):
    return test_dispatcher.RunTests(
        ['a', 'b', 'c'], runner_factory, [_MockDevice('0'), _MockDevice('1')],
        shard=True)

  def testShard(self):
    results, exit_code = TestShard._RunShard(MockRunner)
    self.assertEqual(len(results.GetPass()), 3)
    self.assertEqual(exit_code, 0)

  def testFailing(self):
    results, exit_code = TestShard._RunShard(MockRunnerFail)
    self.assertEqual(len(results.GetPass()), 0)
    self.assertEqual(len(results.GetFail()), 3)
    self.assertEqual(exit_code, constants.ERROR_EXIT_CODE)

  def testNoTests(self):
    results, exit_code = test_dispatcher.RunTests(
        [], MockRunner, [_MockDevice('0'), _MockDevice('1')], shard=True)
    self.assertEqual(len(results.GetAll()), 0)
    self.assertEqual(exit_code, constants.ERROR_EXIT_CODE)


class TestReplicate(unittest.TestCase):
  """Tests test_dispatcher.RunTests with replication."""
  @staticmethod
  def _RunReplicate(runner_factory):
    return test_dispatcher.RunTests(
        ['a', 'b', 'c'], runner_factory, [_MockDevice('0'), _MockDevice('1')],
        shard=False)

  def testReplicate(self):
    results, exit_code = TestReplicate._RunReplicate(MockRunner)
    # We expect 6 results since each test should have been run on every device
    self.assertEqual(len(results.GetPass()), 6)
    self.assertEqual(exit_code, 0)

  def testFailing(self):
    results, exit_code = TestReplicate._RunReplicate(MockRunnerFail)
    self.assertEqual(len(results.GetPass()), 0)
    self.assertEqual(len(results.GetFail()), 6)
    self.assertEqual(exit_code, constants.ERROR_EXIT_CODE)

  def testNoTests(self):
    results, exit_code = test_dispatcher.RunTests(
        [], MockRunner, [_MockDevice('0'), _MockDevice('1')], shard=False)
    self.assertEqual(len(results.GetAll()), 0)
    self.assertEqual(exit_code, constants.ERROR_EXIT_CODE)


if __name__ == '__main__':
  unittest.main()