The Tor Browser: media/webrtc/trunk/build/android/pylib/python_test

media/webrtc/trunk/build/android/pylib/python_test_sharder.py@8bccb770b82d (annotated)

media/webrtc/trunk/build/android/pylib/python_test_sharder.py

Wed, 31 Dec 2014 13:27:57 +0100

author: Michael Schloh von Bennewitz <michael@schloh.com>
date: Wed, 31 Dec 2014 13:27:57 +0100
branch: TOR_BUG_3246
changeset 6: 8bccb770b82d
permissions: -rw-r--r--

Ignore runtime configuration files generated during quality assurance.

 # 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

The Tor Browser / annotate

media/webrtc/trunk/build/android/pylib/python_test_sharder.py@8bccb770b82d (annotated)

media/webrtc/trunk/build/android/pylib/python_test_sharder.py