Android: adds instrumentation test runners.

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 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, device_id, shard_index):
+    """Constructor.
+
+    Args:
+      device_id: ID of the device which this test will talk to.
+      shard_index: shard index, used to create such as unique port numbers.
+    """
+    self.device_id = device_id
+    self.shard_index = shard_index
+
+  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.device_id, self.shard_index)
+      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.
+    shard_retries: number of retries for any given test.
+    available_tests: a list of tests to run which subclass PythonTestBase.
+
+  Returns:
+    An aggregated list of test results.
+  """
+  tests_container = None
+
+  def __init__(self, attached_devices, shard_retries, available_tests):
+    self.attached_devices = attached_devices
+    self.retries = 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_runner = PythonTestRunner(device, index)
+      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