Fuchsia: Add net test server support to the new-style runners.

This CL uses SSH reverse port forwarding to dynamically connect the test
suites running on Fuchsia to the Python-based test servers running on
the controlling machine.

Also adds a logging context to the spawning server library, so that the
verbosity of its logging output can be independently adjusted.

Change-Id: I6099ccc21efdd5e2e6b994e863f8d4b3dadaaae1
Reviewed-on: https://chromium-review.googlesource.com/907633
Commit-Queue: Kevin Marshall <kmarshall@chromium.org>
Reviewed-by: Scott Graham <scottmg@chromium.org>
Reviewed-by: Wez <wez@chromium.org>
Reviewed-by: Sergey Ulanov <sergeyu@chromium.org>
Cr-Commit-Position: refs/heads/master@{#535911}

build/fuchsia/runner_v2/common_args.py

@@ -45,6 +45,13 @@ def ConfigureLogging(args):
 
   logging.basicConfig(level=(logging.DEBUG if args.verbose else logging.INFO))
 
+  # The test server spawner is too noisy with INFO level logging, so tweak
+  # its verbosity a bit by adjusting its logging level.
+  if args.verbose:
+    logging.getLogger('chrome_test_server_spawner').setLevel(logging.DEBUG)
+  else:
+    logging.getLogger('chrome_test_server_spawner').setLevel(logging.WARN)
+
 
 def GetDeploymentTargetForArgs(args):
   """Constructs a deployment target object using parameters taken from

build/fuchsia/runner_v2/net_test_server.py

+# Copyright 2018 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 json
+import logging
+import os
+import re
+import select
+import socket
+import sys
+import subprocess
+import tempfile
+import time
+
+DIR_SOURCE_ROOT = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir))
+sys.path.append(os.path.join(DIR_SOURCE_ROOT, 'build', 'util', 'lib', 'common'))
+import chrome_test_server_spawner
+
+PORT_MAP_RE = re.compile('Allocated port (?P<port>\d+) for remote')
+GET_PORT_NUM_TIMEOUT_SECS = 5
+
+
+def _ConnectPortForwardingTask(target, local_port):
+  """Establishes a port forwarding SSH task to a localhost TCP endpoint hosted
+  at port |local_port|. Blocks until port forwarding is established.
+
+  Returns a tuple containing the remote port and the SSH task Popen object."""
+
+  forwarding_flags = ['-NT',  # Don't execute command; don't allocate terminal.
+                      '-R', '0:localhost:%d' % local_port]
+  task = target.RunCommandPiped([],
+                                ssh_args=forwarding_flags,
+                                stderr=subprocess.PIPE)
+
+  # SSH reports the remote dynamic port number over stderr.
+  # Unfortunately, the output is incompatible with Python's line buffered
+  # input (or vice versa), so we have to build our own buffered input system to
+  # pull bytes over the pipe.
+  poll_obj = select.poll()
+  poll_obj.register(task.stderr, select.POLLIN)
+  line = ''
+  timeout = time.time() + GET_PORT_NUM_TIMEOUT_SECS
+  while time.time() < timeout:
+    poll_result = poll_obj.poll(max(0, timeout - time.time()))
+    if poll_result:
+      next_char = task.stderr.read(1)
+      if not next_char:
+        break
+      line += next_char
+      if line.endswith('\n'):
+        line.strip()
+        matched = PORT_MAP_RE.match(line)
+        if matched:
+          device_port = int(matched.group('port'))
+          logging.debug('Port forwarding established (local=%d, device=%d)' %
+                        (local_port, device_port))
+          return (device_port, task)
+        line = ''
+
+  raise Exception('Could not establish a port forwarding connection.')
+
+
+# Implementation of chrome_test_server_spawner.PortForwarder that uses SSH's
+# remote port forwarding feature to forward ports.
+class SSHPortForwarder(chrome_test_server_spawner.PortForwarder):
+  def __init__(self, target):
+    self._target = target
+
+    # Maps the host (server) port to a tuple consisting of the device port
+    # number and a subprocess.Popen object for the forwarding SSH process.
+    self._port_mapping = {}
+
+  def Map(self, port_pairs):
+    for p in port_pairs:
+      _, host_port = p
+      self._port_mapping[host_port] = \
+          _ConnectPortForwardingTask(self._target, host_port)
+
+  def GetDevicePortForHostPort(self, host_port):
+    return self._port_mapping[host_port][0]
+
+  def Unmap(self, device_port):
+    for host_port, entry in self._port_mapping.iteritems():
+      if entry[0] == device_port:
+        entry[1].terminate()
+        entry[1].wait()
+        del self._port_mapping[host_port]
+        return
+
+    raise Exception('Unmap called for unknown port: %d' % device_port)
+
+
+def SetupTestServer(target, test_concurrency):
+  """Provisions a forwarding test server and configures |target| to use it.
+
+  Returns a tuple with a Popen opbject for the test server process,
+  and a Popen object for the port forwarding connection."""
+
+  logging.debug('Starting test server.')
+  spawning_server = chrome_test_server_spawner.SpawningServer(
+      0, SSHPortForwarder(target), test_concurrency)
+  forwarded_port, forwarding_process = _ConnectPortForwardingTask(
+      target, spawning_server.server_port)
+  spawning_server.Start()
+
+  logging.debug('Test server listening for connections (port=%d)' %
+                spawning_server.server_port)
+  logging.debug('Forwarded port is %d' % forwarded_port)
+
+  config_file = tempfile.NamedTemporaryFile(delete=True)
+
+  # Clean up the config JSON to only pass ports. See https://crbug.com/810209 .
+  config_file.write(json.dumps({
+    'name': 'testserver',
+    'address': '127.0.0.1',
+    'spawner_url_base': 'http://localhost:%d' % forwarded_port
+  }))
+
+  config_file.flush()
+  target.PutFile(config_file.name, '/system/net-test-server-config')
+
+  return spawning_server, forwarding_process

build/fuchsia/runner_v2/remote_cmd.py

@@ -43,7 +43,8 @@ def RunSsh(config_path, host, port, command, silent):
     return subprocess.call(ssh_command)
 
 
-def RunPipedSsh(config_path, host, port, command, **kwargs):
+def RunPipedSsh(config_path, host, port, command = None, ssh_args = None,
+                **kwargs):
   """Executes an SSH command on the remote host and returns a process object
   with access to the command's stdio streams. Does not block.
 
@@ -51,14 +52,20 @@ def RunPipedSsh(config_path, host, port, command, **kwargs):
   host: The hostname or IP address of the remote host.
   port: The port to connect to.
   command: A list of strings containing the command and its arguments.
+  ssh_args: Arguments that will be passed to SSH.
   kwargs: A dictionary of parameters to be passed to subprocess.Popen().
           The parameters can be used to override stdin and stdout, for example.
 
   Returns a Popen object for the command."""
 
+  if not command:
+    command = []
+  if not ssh_args:
+    ssh_args = []
+
   ssh_command = _SSH + ['-F', config_path,
                         host,
-                        '-p', str(port)] + command
+                        '-p', str(port)] + ssh_args + ['--'] + command
   logging.debug(' '.join(ssh_command))
   return subprocess.Popen(ssh_command, **kwargs)
 

build/fuchsia/runner_v2/test_runner.py

@@ -19,6 +19,7 @@ import tempfile
 import time
 
 from common_args import AddCommonArgs, ConfigureLogging, GetDeploymentTargetForArgs
+from net_test_server import SetupTestServer
 from run_package import RunPackage
 
 DEFAULT_TEST_CONCURRENCY = 4
@@ -56,6 +57,9 @@ def main():
                       help='Sets the number of parallel test jobs.')
   parser.add_argument('--test-launcher-summary-output',
                       help='Where the test launcher will output its json.')
+  parser.add_argument('--enable-test-server', action='store_true',
+                      default=False,
+                      help='Enable Chrome test server spawner.')
   parser.add_argument('child_args', nargs='*',
                       help='Arguments for the test process.')
   args = parser.parse_args()
@@ -86,15 +90,24 @@ def main():
     child_args.append('--test-launcher-summary-output=' + TEST_RESULT_PATH)
 
   with GetDeploymentTargetForArgs(args) as target:
+    target = GetDeploymentTargetForArgs(args)
     target.Start()
 
     if args.test_launcher_filter_file:
       target.PutFile(args.test_launcher_filter_file, TEST_FILTER_PATH)
       child_args.append('--test-launcher-filter-file=' + TEST_FILTER_PATH)
 
+    forwarder = None
+    if args.enable_test_server:
+      test_server, forwarder = SetupTestServer(target, test_concurrency)
+
     RunPackage(args.output_directory, target, args.package,
                child_args, args.package_manifest)
 
+    if forwarder:
+      forwarder.terminate()
+      forwarder.wait()
+
     if args.test_launcher_summary_output:
       target.GetFile(TEST_RESULT_PATH, args.test_launcher_summary_output)
 

build/util/lib/common/chrome_test_server_spawner.py

@@ -22,21 +22,6 @@ import time
 import urlparse
 
 
-DIR_SOURCE_ROOT = os.path.abspath(
-    os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir,
-                 os.pardir))
-
-
-# Path that are needed to import necessary modules when launching a testserver.
-os.environ['PYTHONPATH'] = os.environ.get('PYTHONPATH', '') + (':%s:%s:%s:%s:%s'
-    % (os.path.join(DIR_SOURCE_ROOT, 'third_party'),
-       os.path.join(DIR_SOURCE_ROOT, 'third_party', 'tlslite'),
-       os.path.join(DIR_SOURCE_ROOT, 'third_party', 'pyftpdlib', 'src'),
-       os.path.join(DIR_SOURCE_ROOT, 'net', 'tools', 'testserver'),
-       os.path.join(DIR_SOURCE_ROOT, 'components', 'sync', 'tools',
-                    'testserver')))
-
-
 SERVER_TYPES = {
     'http': '',
     'ftp': '-f',
@@ -47,8 +32,26 @@ SERVER_TYPES = {
 }
 
 
+_DIR_SOURCE_ROOT = os.path.abspath(
+    os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir,
+                 os.pardir))
+
+
+_logger = logging.getLogger(__name__)
+
+
+# Path that are needed to import necessary modules when launching a testserver.
+os.environ['PYTHONPATH'] = os.environ.get('PYTHONPATH', '') + (':%s:%s:%s:%s:%s'
+    % (os.path.join(_DIR_SOURCE_ROOT, 'third_party'),
+       os.path.join(_DIR_SOURCE_ROOT, 'third_party', 'tlslite'),
+       os.path.join(_DIR_SOURCE_ROOT, 'third_party', 'pyftpdlib', 'src'),
+       os.path.join(_DIR_SOURCE_ROOT, 'net', 'tools', 'testserver'),
+       os.path.join(_DIR_SOURCE_ROOT, 'components', 'sync', 'tools',
+                    'testserver')))
+
+
 # The timeout (in seconds) of starting up the Python test server.
-TEST_SERVER_STARTUP_TIMEOUT = 10
+_TEST_SERVER_STARTUP_TIMEOUT = 10
 
 
 def _GetServerTypeCommandLine(server_type):
@@ -137,9 +140,9 @@ class TestServerThread(threading.Thread):
     """
     assert self.host_port == 0 and self.pipe_out and self.pipe_in
     (in_fds, _, _) = select.select([self.pipe_in, ], [], [],
-                                   TEST_SERVER_STARTUP_TIMEOUT)
+                                   _TEST_SERVER_STARTUP_TIMEOUT)
     if len(in_fds) == 0:
-      logging.error('Failed to wait to the Python test server to be started.')
+      _logger.error('Failed to wait to the Python test server to be started.')
       return False
     # First read the data length as an unsigned 4-byte value.  This
     # is _not_ using network byte ordering since the Python test server packs
@@ -152,13 +155,13 @@ class TestServerThread(threading.Thread):
       (data_length,) = struct.unpack('=L', data_length)
       assert data_length
     if not data_length:
-      logging.error('Failed to get length of server data.')
+      _logger.error('Failed to get length of server data.')
       return False
     server_data_json = os.read(self.pipe_in, data_length)
     if not server_data_json:
-      logging.error('Failed to get server data.')
+      _logger.error('Failed to get server data.')
       return False
-    logging.info('Got port json data: %s', server_data_json)
+    _logger.info('Got port json data: %s', server_data_json)
 
     parsed_server_data = None
     try:
@@ -167,11 +170,11 @@ class TestServerThread(threading.Thread):
       pass
 
     if not isinstance(parsed_server_data, dict):
-      logging.error('Failed to parse server_data: %s' % server_data_json)
+      _logger.error('Failed to parse server_data: %s' % server_data_json)
       return False
 
     if not isinstance(parsed_server_data.get('port'), int):
-      logging.error('Failed to get port information from the server data.')
+      _logger.error('Failed to get port information from the server data.')
       return False
 
     self.host_port = parsed_server_data['port']
@@ -224,10 +227,10 @@ class TestServerThread(threading.Thread):
           pass
 
   def run(self):
-    logging.info('Start running the thread!')
+    _logger.info('Start running the thread!')
     self.wait_event.clear()
     self._GenerateCommandLineArguments()
-    command = DIR_SOURCE_ROOT
+    command = _DIR_SOURCE_ROOT
     if self.arguments['server-type'] == 'sync':
       command = [os.path.join(command, 'components', 'sync', 'tools',
                               'testserver',
@@ -235,17 +238,17 @@ class TestServerThread(threading.Thread):
     else:
       command = [os.path.join(command, 'net', 'tools', 'testserver',
                               'testserver.py')] + self.command_line
-    logging.info('Running: %s', command)
+    _logger.info('Running: %s', command)
 
     # Disable PYTHONUNBUFFERED because it has a bad interaction with the
     # testserver. Remove once this interaction is fixed.
     unbuf = os.environ.pop('PYTHONUNBUFFERED', None)
 
-    # Pass DIR_SOURCE_ROOT as the child's working directory so that relative
+    # Pass _DIR_SOURCE_ROOT as the child's working directory so that relative
     # paths in the arguments are resolved correctly.
     self.process = subprocess.Popen(
         command, preexec_fn=self._CloseUnnecessaryFDsForTestServerProcess,
-        cwd=DIR_SOURCE_ROOT)
+        cwd=_DIR_SOURCE_ROOT)
     if unbuf:
       os.environ['PYTHONUNBUFFERED'] = unbuf
     if self.process:
@@ -284,7 +287,7 @@ class TestServerThread(threading.Thread):
       os.close(self.pipe_out)
       self.pipe_in = None
       self.pipe_out = None
-    logging.info('Test-server has died.')
+    _logger.info('Test-server has died.')
     self.wait_event.set()
 
   def Stop(self):
@@ -325,7 +328,7 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
 
   def _StartTestServer(self):
     """Starts the test server thread."""
-    logging.info('Handling request to spawn a test server.')
+    _logger.info('Handling request to spawn a test server.')
     content_type = self.headers.getheader('content-type')
     if content_type != 'application/json':
       raise Exception('Bad content-type for start request.')
@@ -336,9 +339,9 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
       content_length = int(content_length)
     except:
       raise Exception('Bad content-length for start request.')
-    logging.info(content_length)
+    _logger.info(content_length)
     test_server_argument_json = self.rfile.read(content_length)
-    logging.info(test_server_argument_json)
+    _logger.info(test_server_argument_json)
 
     if len(self.server.test_servers) >= self.server.max_instances:
       self._SendResponse(400, 'Invalid request', {},
@@ -358,7 +361,7 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
       if new_server.forwarder_ocsp_device_port:
         response['ocsp_port'] = new_server.forwarder_ocsp_device_port
       self._SendResponse(200, 'OK', {}, json.dumps(response))
-      logging.info('Test server is running on port %d forwarded to %d.' %
+      _logger.info('Test server is running on port %d forwarded to %d.' %
               (new_server.forwarder_device_port, new_server.host_port))
       port = new_server.forwarder_device_port
       assert not self.server.test_servers.has_key(port)
@@ -366,7 +369,7 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
     else:
       new_server.Stop()
       self._SendResponse(500, 'Test Server Error.', {}, '')
-      logging.info('Encounter problem during starting a test server.')
+      _logger.info('Encounter problem during starting a test server.')
 
   def _KillTestServer(self, params):
     """Stops the test server instance."""
@@ -385,34 +388,40 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
 
     server = self.server.test_servers.pop(port)
 
-    logging.info('Handling request to kill a test server on port: %d.', port)
+    _logger.info('Handling request to kill a test server on port: %d.', port)
     server.Stop()
 
     # Make sure the status of test server is correct before sending response.
     if self.server.port_forwarder.WaitHostPortAvailable(port):
       self._SendResponse(200, 'OK', {}, 'killed')
-      logging.info('Test server on port %d is killed', port)
+      _logger.info('Test server on port %d is killed', port)
     else:
       self._SendResponse(500, 'Test Server Error.', {}, '')
-      logging.info('Encounter problem during killing a test server.')
+      _logger.info('Encounter problem during killing a test server.')
+
+  def log_message(self, format, *args):
+    # Suppress the default HTTP logging behavior if the logging level is higher
+    # than INFO.
+    if _logger.getEffectiveLevel() <= logging.INFO:
+      pass
 
   def do_POST(self):
     parsed_path = urlparse.urlparse(self.path)
     action = parsed_path.path
-    logging.info('Action for POST method is: %s.', action)
+    _logger.info('Action for POST method is: %s.', action)
     if action == '/start':
       self._StartTestServer()
     else:
       self._SendResponse(400, 'Unknown request.', {}, '')
-      logging.info('Encounter unknown request: %s.', action)
+      _logger.info('Encounter unknown request: %s.', action)
 
   def do_GET(self):
     parsed_path = urlparse.urlparse(self.path)
     action = parsed_path.path
     params = urlparse.parse_qs(parsed_path.query, keep_blank_values=1)
-    logging.info('Action for GET method is: %s.', action)
+    _logger.info('Action for GET method is: %s.', action)
     for param in params:
-      logging.info('%s=%s', param, params[param][0])
+      _logger.info('%s=%s', param, params[param][0])
     if action == '/kill':
       self._KillTestServer(params)
     elif action == '/ping':
@@ -420,10 +429,10 @@ class SpawningServerRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
       # to serve the requests. We don't need to test the status of the test
       # server when handling ping request.
       self._SendResponse(200, 'OK', {}, 'ready')
-      logging.info('Handled ping request and sent response.')
+      _logger.info('Handled ping request and sent response.')
     else:
       self._SendResponse(400, 'Unknown request', {}, '')
-      logging.info('Encounter unknown request: %s.', action)
+      _logger.info('Encounter unknown request: %s.', action)
 
 
 class SpawningServer(object):
@@ -433,14 +442,14 @@ class SpawningServer(object):
     self.server = BaseHTTPServer.HTTPServer(('', test_server_spawner_port),
                                             SpawningServerRequestHandler)
     self.server_port = self.server.server_port
-    logging.info('Started test server spawner on port: %d.', self.server_port)
+    _logger.info('Started test server spawner on port: %d.', self.server_port)
 
     self.server.port_forwarder = port_forwarder
     self.server.test_servers = {}
     self.server.max_instances = max_instances
 
   def _Listen(self):
-    logging.info('Starting test server spawner.')
+    _logger.info('Starting test server spawner.')
     self.server.serve_forever()
 
   def Start(self):
@@ -464,8 +473,8 @@ class SpawningServer(object):
     to avoid sharing the test server instance.
     """
     if self.server.test_servers:
-      logging.warning('Not all test servers were stopped.')
+      _logger.warning('Not all test servers were stopped.')
       for port in self.server.test_servers:
-        logging.warning('Stopping test server on port %d' % port)
+        _logger.warning('Stopping test server on port %d' % port)
         self.server.test_servers[port].Stop()
       self.server.test_servers = {}