108 строки
3.4 KiB
Python
108 строки
3.4 KiB
Python
# 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.
|
|
|
|
|
|
import logging
|
|
import multiprocessing
|
|
|
|
from test_result import *
|
|
|
|
|
|
def _ShardedTestRunnable(test):
|
|
"""Standalone function needed by multiprocessing.Pool."""
|
|
log_format = '[' + test.device + '] # %(asctime)-15s: %(message)s'
|
|
if logging.getLogger().handlers:
|
|
logging.getLogger().handlers[0].setFormatter(logging.Formatter(log_format))
|
|
else:
|
|
logging.basicConfig(format=log_format)
|
|
return test.Run()
|
|
|
|
|
|
def SetTestsContainer(tests_container):
|
|
"""Sets tests container.
|
|
|
|
multiprocessing.Queue can't be pickled across processes, so we need to set
|
|
this as a 'global', per process, via multiprocessing.Pool.
|
|
"""
|
|
BaseTestSharder.tests_container = tests_container
|
|
|
|
|
|
class BaseTestSharder(object):
|
|
"""Base class for sharding tests across multiple devices.
|
|
|
|
Args:
|
|
attached_devices: A list of attached devices.
|
|
"""
|
|
# See more in SetTestsContainer.
|
|
tests_container = None
|
|
|
|
def __init__(self, attached_devices):
|
|
self.attached_devices = attached_devices
|
|
self.retries = 1
|
|
self.tests = []
|
|
|
|
def CreateShardedTestRunner(self, device, index):
|
|
"""Factory function to create a suite-specific test runner.
|
|
|
|
Args:
|
|
device: Device serial where this shard will run
|
|
index: Index of this device in the pool.
|
|
|
|
Returns:
|
|
An object of BaseTestRunner type (that can provide a "Run()" method).
|
|
"""
|
|
pass
|
|
|
|
def SetupSharding(self, tests):
|
|
"""Called before starting the shards."""
|
|
pass
|
|
|
|
def OnTestsCompleted(self, test_runners, test_results):
|
|
"""Notifies that we completed the tests."""
|
|
pass
|
|
|
|
def RunShardedTests(self):
|
|
"""Runs the tests in all connected devices.
|
|
|
|
Returns:
|
|
A TestResults object.
|
|
"""
|
|
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 = TestResults()
|
|
for retry in xrange(self.retries):
|
|
logging.warning('Try %d of %d' % (retry + 1, self.retries))
|
|
self.SetupSharding(self.tests)
|
|
test_runners = []
|
|
for index, device in enumerate(self.attached_devices):
|
|
logging.warning('*' * 80)
|
|
logging.warning('Creating shard %d for %s' % (index, device))
|
|
logging.warning('*' * 80)
|
|
test_runner = self.CreateShardedTestRunner(device, index)
|
|
test_runners += [test_runner]
|
|
logging.warning('Starting...')
|
|
pool = multiprocessing.Pool(len(self.attached_devices),
|
|
SetTestsContainer,
|
|
[BaseTestSharder.tests_container])
|
|
results_lists = pool.map(_ShardedTestRunnable, test_runners)
|
|
test_results = TestResults.FromTestResults(results_lists)
|
|
if retry == self.retries - 1:
|
|
all_passed = final_results.ok + test_results.ok
|
|
final_results = test_results
|
|
final_results.ok = all_passed
|
|
break
|
|
else:
|
|
final_results.ok += test_results.ok
|
|
self.tests = []
|
|
for t in test_results.GetAllBroken():
|
|
self.tests += [t.name]
|
|
if not self.tests:
|
|
break
|
|
self.OnTestsCompleted(test_runners, final_results)
|
|
return final_results
|