[Android] Converts host driven tests to common test_dispatcher

build/android/pylib/host_driven/python_test_sharder.py

-# Copyright (c) 2012 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.
-
-"""Takes care of sharding the python-drive tests in multiple devices."""
-
-import copy
-import logging
-import multiprocessing
-
-from pylib.base import base_test_result
-from pylib.base import sharded_tests_queue
-from pylib.forwarder import Forwarder
-
-from python_test_caller import CallPythonTest
-
-
-def SetTestsContainer(tests_container):
-  """Sets PythonTestSharder as a top-level field.
-
-  PythonTestSharder uses multiprocessing.Pool, which creates a pool of
-  processes. This is used to initialize each worker in the pool, ensuring that
-  each worker has access to this shared pool of tests.
-
-  The multiprocessing module requires that this be a top-level method.
-
-  Args:
-    tests_container: the container for all the tests.
-  """
-  PythonTestSharder.tests_container = tests_container
-
-
-def _DefaultRunnable(test_runner):
-  """A default runnable for a PythonTestRunner.
-
-  Args:
-    test_runner: A PythonTestRunner which will run tests.
-
-  Returns:
-    The test results.
-  """
-  return test_runner.RunTests()
-
-
-class PythonTestRunner(object):
-  """Thin wrapper around a list of PythonTestBase instances.
-
-  This is meant to be a long-lived object which can run multiple Python tests
-  within its lifetime. Tests will receive the device_id and shard_index.
-
-  The shard index affords the ability to create unique port numbers (e.g.
-  DEFAULT_PORT + shard_index) if the test so wishes.
-  """
-
-  def __init__(self, options):
-    """Constructor.
-
-    Args:
-      options: Options to use for setting up tests.
-    """
-    self.options = options
-
-  def RunTests(self):
-    """Runs tests from the shared pool of tests, aggregating results.
-
-    Returns:
-      A list of test results for all of the tests which this runner executed.
-    """
-    tests = PythonTestSharder.tests_container
-
-    results = base_test_result.TestRunResults()
-    for t in tests:
-      results.AddTestRunResults(CallPythonTest(t, self.options))
-    return results
-
-
-class PythonTestSharder(object):
-  """Runs Python tests in parallel on multiple devices.
-
-  This is lifted more or less wholesale from BaseTestRunner.
-
-  Under the covers, it creates a pool of long-lived PythonTestRunners, which
-  execute tests from the pool of tests.
-
-  Args:
-    attached_devices: a list of device IDs attached to the host.
-    available_tests: a list of tests to run which subclass PythonTestBase.
-    options: Options to use for setting up tests.
-
-  Returns:
-    An aggregated list of test results.
-  """
-  tests_container = None
-
-  def __init__(self, attached_devices, available_tests, options):
-    self.options = options
-    self.attached_devices = attached_devices
-    self.retries = options.num_retries
-    self.tests = available_tests
-
-  def _SetupSharding(self, tests):
-    """Creates the shared pool of tests and makes it available to test runners.
-
-    Args:
-      tests: the list of tests which will be consumed by workers.
-    """
-    SetTestsContainer(sharded_tests_queue.ShardedTestsQueue(
-        len(self.attached_devices), tests))
-
-  def RunShardedTests(self):
-    """Runs tests in parallel using a pool of workers.
-
-    Returns:
-      A list of test results aggregated from all test runs.
-    """
-    logging.warning('*' * 80)
-    logging.warning('Sharding in ' + str(len(self.attached_devices)) +
-                    ' devices.')
-    logging.warning('Note that the output is not synchronized.')
-    logging.warning('Look for the "Final result" banner in the end.')
-    logging.warning('*' * 80)
-    final_results = base_test_result.TestRunResults()
-    tests_to_run = self.tests
-
-    Forwarder.UseMultiprocessing()
-
-    for retry in xrange(self.retries):
-      logging.warning('Try %d of %d', retry + 1, self.retries)
-      self._SetupSharding(self.tests)
-      test_runners = self._MakeTestRunners(self.attached_devices)
-      logging.warning('Starting...')
-      pool = multiprocessing.Pool(len(self.attached_devices),
-                                  SetTestsContainer,
-                                  [PythonTestSharder.tests_container])
-
-      # List of TestRunResults objects from each test execution.
-      try:
-        results_lists = pool.map(_DefaultRunnable, test_runners)
-      except Exception:
-        logging.exception('Unable to run tests. Something with the '
-                          'PythonTestRunners has gone wrong.')
-        raise Exception('PythonTestRunners were unable to run tests.')
-
-      test_results = base_test_result.TestRunResults()
-      for t in results_lists:
-        test_results.AddTestRunResults(t)
-      # Accumulate passing results.
-      final_results.AddResults(test_results.GetPass())
-      # If we have failed tests, map them to tests to retry.
-      failed_tests = [t.GetName() for t in test_results.GetNotPass()]
-      tests_to_run = self._GetTestsToRetry(self.tests, failed_tests)
-
-      # Bail out early if we have no more tests. This can happen if all tests
-      # pass before we're out of retries, for example.
-      if not tests_to_run:
-        break
-
-    # all_passed has accumulated all passing test results.
-    # test_results will have the results from the most recent run, which could
-    # include a variety of failure modes (unknown, crashed, failed, etc).
-    test_results.AddResults(final_results.GetPass())
-    final_results = test_results
-
-    return final_results
-
-  def _MakeTestRunners(self, attached_devices):
-    """Initialize and return a list of PythonTestRunners.
-
-    Args:
-      attached_devices: list of device IDs attached to host.
-
-    Returns:
-      A list of PythonTestRunners, one for each device.
-    """
-    test_runners = []
-    for index, device in enumerate(attached_devices):
-      logging.warning('*' * 80)
-      logging.warning('Creating shard %d for %s', index, device)
-      logging.warning('*' * 80)
-      # Bind the PythonTestRunner to a device & shard index. Give it the
-      # runnable which it will use to actually execute the tests.
-      test_options = copy.deepcopy(self.options)
-      test_options.ensure_value('device_id', device)
-      test_options.ensure_value('shard_index', index)
-      test_runner = PythonTestRunner(test_options)
-      test_runners.append(test_runner)
-
-    return test_runners
-
-  def _GetTestsToRetry(self, available_tests, failed_test_names):
-    """Infers a list of tests to retry from failed tests and available tests.
-
-    Args:
-      available_tests: a list of tests which subclass PythonTestBase.
-      failed_test_names: a list of failed test names.
-
-    Returns:
-      A list of test objects which correspond to test names found in
-      failed_test_names, or an empty list if there is no correspondence.
-    """
-    tests_to_retry = [t for t in available_tests
-                      if t.qualified_name in failed_test_names]
-    return tests_to_retry