Reorganize Android's test scripts into packages.

build/android/pylib/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 python_test_caller import CallPythonTest
-from run_java_tests import FatalTestException
-import sharded_tests_queue
-from test_result import TestResults
-
-
-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 = []
-    for t in tests:
-      res = CallPythonTest(t, self.options)
-      results.append(res)
-
-    return TestResults.FromTestResults(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.shard_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)
-    all_passed = []
-    test_results = TestResults()
-    tests_to_run = self.tests
-    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 TestResults 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 FatalTestException('PythonTestRunners were unable to run tests.')
-
-      test_results = TestResults.FromTestResults(results_lists)
-      # Accumulate passing results.
-      all_passed += test_results.ok
-      # If we have failed tests, map them to tests to retry.
-      failed_tests = test_results.GetAllBroken()
-      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
-
-    final_results = TestResults()
-    # 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).
-    final_results = test_results
-    final_results.ok = all_passed
-
-    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_tests):
-    """Infers a list of tests to retry from failed tests and available tests.
-
-    Args:
-      available_tests: a list of tests which subclass PythonTestBase.
-      failed_tests: a list of SingleTestResults representing failed tests.
-
-    Returns:
-      A list of test objects which correspond to test names found in
-      failed_tests, or an empty list if there is no correspondence.
-    """
-    failed_test_names = map(lambda t: t.test_name, failed_tests)
-    tests_to_retry = [t for t in available_tests
-                      if t.qualified_name in failed_test_names]
-    return tests_to_retry