Index: build/android/pylib/python_test_sharder.py |
diff --git a/build/android/pylib/python_test_sharder.py b/build/android/pylib/python_test_sharder.py |
new file mode 100644 |
index 0000000000000000000000000000000000000000..7bab235a8ba48ea595c11dc0371b17b084230c06 |
--- /dev/null |
+++ b/build/android/pylib/python_test_sharder.py |
@@ -0,0 +1,200 @@ |
+# 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 |