import time
from pylib import android_commands
-from pylib import constants
+from pylib import forwarder
+from pylib import valgrind_tools
from pylib.base import base_test_result
from pylib.instrumentation import test_package
from pylib.instrumentation import test_result
test_name: The name of the method to run as the test.
instrumentation_options: An InstrumentationOptions object.
"""
- self.test_name = test_name
class_name = self.__class__.__name__
- self.qualified_name = '%s.%s' % (class_name, self.test_name)
+ self.adb = None
+ self.cleanup_test_files = False
+ self.device_id = ''
+ self.has_forwarded_ports = False
+ self.instrumentation_options = instrumentation_options
+ self.ports_to_forward = []
+ self.push_deps = False
+ self.shard_index = 0
+
# Use tagged_name when creating results, so that we can identify host-driven
# tests in the overall results.
+ self.test_name = test_name
+ self.qualified_name = '%s.%s' % (class_name, self.test_name)
self.tagged_name = '%s_%s' % (self._HOST_DRIVEN_TAG, self.qualified_name)
- self.instrumentation_options = instrumentation_options
- self.ports_to_forward = []
-
+ # TODO(bulach): make ports_to_forward not optional and move the Forwarder
+ # mapping here.
def SetUp(self, device, shard_index, push_deps,
- cleanup_test_files):
+ cleanup_test_files, ports_to_forward=None):
+ if not ports_to_forward:
+ ports_to_forward = []
self.device_id = device
self.shard_index = shard_index
self.adb = android_commands.AndroidCommands(self.device_id)
self.push_deps = push_deps
self.cleanup_test_files = cleanup_test_files
+ if ports_to_forward:
+ self.ports_to_forward = ports_to_forward
def TearDown(self):
pass
# Get the test method on the derived class and execute it
return getattr(self, self.test_name)()
+ @staticmethod
+ def __GetHostForwarderLog():
+ return ('-- Begin Full HostForwarder log\n'
+ '%s\n'
+ '--End Full HostForwarder log\n' % forwarder.Forwarder.GetHostLog())
+
+ def __StartForwarder(self):
+ logging.warning('Forwarding %s %s', self.ports_to_forward,
+ self.has_forwarded_ports)
+ if self.ports_to_forward and not self.has_forwarded_ports:
+ self.has_forwarded_ports = True
+ tool = valgrind_tools.CreateTool(None, self.adb)
+ forwarder.Forwarder.Map([(port, port) for port in self.ports_to_forward],
+ self.adb, tool)
+
def __RunJavaTest(self, test, test_pkg, additional_flags=None):
"""Runs a single Java test in a Java TestRunner.
Returns:
TestRunResults object with a single test result.
"""
+ # TODO(bulach): move this to SetUp() stage.
+ self.__StartForwarder()
+
java_test_runner = test_runner.TestRunner(self.instrumentation_options,
self.device_id,
self.shard_index, test_pkg,
- self.ports_to_forward,
additional_flags=additional_flags)
try:
java_test_runner.SetUp()
start_ms = int(time.time()) * 1000
done = False
for test_filter in test_filters:
- tests = test_pkg._GetAllMatchingTests(None, None, test_filter)
+ tests = test_pkg.GetAllMatchingTests(None, None, test_filter)
# Filters should always result in >= 1 test.
if len(tests) == 0:
raise Exception('Java test filter "%s" returned no tests.'
if not java_result.DidRunPass():
result = java_result.GetNotPass().pop()
log = result.GetLog()
+ log += self.__GetHostForwarderLog()
test_type = result.GetType()
done = True
break