xref: /aosp_15_r20/external/cronet/testing/chromoting/chromoting_test_utilities.py (revision 6777b5387eb2ff775bb5750e3f5d96f37fb7352b)

# Copyright 2015 The Chromium Authors
# 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."""

from __future__ import print_function

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 as 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