testing/chromoting/chromoting_test_utilities.py

# Copyright 2015 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.

"""Utility script to run tests on the Chromoting bot."""

import hashlib
import os
from os.path import expanduser
import re
import shutil
import socket
import subprocess

import psutil

PROD_DIR_ID = '#PROD_DIR#'
CRD_ID = 'chrome-remote-desktop'  # Used in a few file/folder names
HOST_READY_INDICATOR = 'Host ready to receive connections.'
BROWSER_TEST_ID = 'browser_tests'
HOST_HASH_VALUE = hashlib.md5(socket.gethostname()).hexdigest()
NATIVE_MESSAGING_DIR = 'NativeMessagingHosts'
# On a Swarming bot where these tests are executed, a temp folder is created
# under which the files specified in an .isolate are copied. This temp folder
# has a random name, which we'll store here for use later.
# Note that the test-execution always starts from the testing/chromoting folder
# under the temp folder.
ISOLATE_CHROMOTING_HOST_PATH = 'remoting/host/linux/linux_me2me_host.py'
ISOLATE_TEMP_FOLDER = os.path.abspath(os.path.join(os.getcwd(), '../..'))
CHROMOTING_HOST_PATH = os.path.join(ISOLATE_TEMP_FOLDER,
                                    ISOLATE_CHROMOTING_HOST_PATH)
MAX_RETRIES = 1


class HostOperationFailedException(Exception):
  pass


def RunCommandInSubProcess(command):
  """Creates a subprocess with command-line that is passed in.

  Args:
    command: The text of command to be executed.
  Returns:
    results: stdout contents of executing the command.
  """

  cmd_line = [command]
  try:
    print 'Going to run:\n%s' % command
    results = subprocess.check_output(cmd_line, stderr=subprocess.STDOUT,
                                      shell=True)
  except subprocess.CalledProcessError, e:
    results = e.output
  finally:
    print results
  return results


def TestMachineCleanup(user_profile_dir, host_logs=None):
  """Cleans up test machine so as not to impact other tests.

  Args:
    user_profile_dir: the user-profile folder used by Chromoting tests.
    host_logs: List of me2me host logs; these will be deleted.
  """

  # Stop the host service.
  RunCommandInSubProcess(CHROMOTING_HOST_PATH + ' --stop')

  # Cleanup any host logs.
  if host_logs:
    for host_log in host_logs:
      RunCommandInSubProcess('rm %s' % host_log)

  # Remove the user-profile dir
  if os.path.exists(user_profile_dir):
    shutil.rmtree(user_profile_dir)


def InitialiseTestMachineForLinux(cfg_file):
  """Sets up a Linux machine for connect-to-host chromoting tests.

  Copy over me2me host-config to expected locations.
  By default, the Linux me2me host expects the host-config file to be under
  $HOME/.config/chrome-remote-desktop
  Its name is expected to have a hash that is specific to a machine.

  Args:
    cfg_file: location of test account's host-config file.

  Raises:
    Exception: if host did not start properly.
  """

  # First get home directory on current machine.
  home_dir = expanduser('~')
  default_config_file_location = os.path.join(home_dir, '.config', CRD_ID)
  if os.path.exists(default_config_file_location):
    shutil.rmtree(default_config_file_location)
  os.makedirs(default_config_file_location)

  # Copy over test host-config to expected location, with expected file-name.
  # The file-name should contain a hash-value that is machine-specific.
  default_config_file_name = 'host#%s.json' % HOST_HASH_VALUE
  config_file_src = os.path.join(os.getcwd(), cfg_file)
  shutil.copyfile(
      config_file_src,
      os.path.join(default_config_file_location, default_config_file_name))

  # Make sure chromoting host is running.
  RestartMe2MeHost()


def RestartMe2MeHost():
  """Stops and starts the Me2Me host on the test machine.

  Launches the me2me start-host command, and parses the stdout of the execution
  to obtain the host log-file name.

  Returns:
    log_file: Host log file.

  Raises:
    Exception: If host-log does not contain string indicating host is ready.
  """

  # To start the host, we want to be in the temp-folder for this test execution.
  # Store the current folder to return back to it later.
  previous_directory = os.getcwd()
  os.chdir(ISOLATE_TEMP_FOLDER)

  # Stop chromoting host.
  RunCommandInSubProcess(CHROMOTING_HOST_PATH + ' --stop')
  # Start chromoting host.
  print 'Starting chromoting host from %s' % CHROMOTING_HOST_PATH
  results = RunCommandInSubProcess(CHROMOTING_HOST_PATH + ' --start')

  os.chdir(previous_directory)

  # Get log file from results of above command printed to stdout. Example:
  # Log file: /tmp/tmp0c3EcP/chrome_remote_desktop_20150929_101525_B0o89t
  start_of_host_log = results.index('Log file: ') + len('Log file: ')
  log_file = results[start_of_host_log:].rstrip()

  # Confirm that the start process completed, and we got:
  # "Host ready to receive connections." in the log.
  if HOST_READY_INDICATOR not in results:
    # Host start failed. Print out host-log. Don't run any tests.
    with open(log_file, 'r') as f:
      print f.read()
    raise HostOperationFailedException('Host restart failed.')

  return log_file


def CleanupUserProfileDir(args):
  SetupUserProfileDir(args.me2me_manifest_file, args.it2me_manifest_file,
                      args.user_profile_dir)


def SetupUserProfileDir(me2me_manifest_file, it2me_manifest_file,
                        user_profile_dir):
  """Sets up the Google Chrome user profile directory.

  Delete the previous user profile directory if exists and create a new one.
  This invalidates any state changes by the previous test so each test can start
  with the same environment.

  When a user launches the remoting web-app, the native messaging host process
  is started. For this to work, this function places the me2me and it2me native
  messaging host manifest files in a specific folder under the user-profile dir.

  Args:
    me2me_manifest_file: location of me2me native messaging host manifest file.
    it2me_manifest_file: location of it2me native messaging host manifest file.
    user_profile_dir: Chrome user-profile-directory.
  """
  native_messaging_folder = os.path.join(user_profile_dir, NATIVE_MESSAGING_DIR)

  if os.path.exists(user_profile_dir):
    shutil.rmtree(user_profile_dir)
  os.makedirs(native_messaging_folder)

  manifest_files = [me2me_manifest_file, it2me_manifest_file]
  for manifest_file in manifest_files:
    manifest_file_src = os.path.join(os.getcwd(), manifest_file)
    manifest_file_dest = (
        os.path.join(native_messaging_folder, os.path.basename(manifest_file)))
    shutil.copyfile(manifest_file_src, manifest_file_dest)


def PrintRunningProcesses():
  processes = psutil.get_process_list()
  processes = sorted(processes, key=lambda process: process.name)

  print 'List of running processes:\n'
  for process in processes:
    print process.name


def PrintHostLogContents(host_log_files=None):
  if host_log_files:
    host_log_contents = ''
    for log_file in sorted(host_log_files):
      with open(log_file, 'r') as log:
        host_log_contents += '\nHOST LOG %s\n CONTENTS:\n%s' % (
            log_file, log.read())
    print host_log_contents


def TestCaseSetup(args):
  # Reset the user profile directory to start each test with a clean slate.
  CleanupUserProfileDir(args)
  # Stop+start me2me host process.
  return RestartMe2MeHost()


def GetJidListFromTestResults(results):
  """Parse the output of a test execution to obtain the JID used by the test.

  Args:
    results: stdio contents of test execution.

  Returns:
    jids_used: List of JIDs used by test; empty list if not found.
  """

  # Reg-ex defining the JID information in the string being parsed.
  jid_re = '(Connecting to )(.*.gserviceaccount.com/chromoting.*)(. Local.*)'
  jids_used = []
  for line in results.split('\n'):
    match = re.search(jid_re, line)
    if match:
      jid_used = match.group(2)
      if jid_used not in jids_used:
        jids_used.append(jid_used)

  return jids_used


def GetJidFromHostLog(host_log_file):
  """Parse the me2me host log to obtain the JID that the host registered.

  Args:
    host_log_file: path to host-log file that should be parsed for a JID.

  Returns:
    host_jid: host-JID if found in host-log, else None
  """
  host_jid = None
  with open(host_log_file, 'r') as log_file:
    for line in log_file:
      # The host JID will be recorded in a line saying 'Signaling
      # connected'.
      if 'Signaling connected. ' in line:
        components = line.split(':')
        host_jid = components[-1].lstrip()
        break

  return host_jid