android-13.0.0_r83/s

# Copyright 2016 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
import re
import subprocess
import threading
import time

from mobly import utils

# Command to use for running ADB commands.
ADB = 'adb'

# adb gets confused if we try to manage bound ports in parallel, so anything to
# do with port forwarding must happen under this lock.
ADB_PORT_LOCK = threading.Lock()

# Number of attempts to execute "adb root", and seconds for interval time of
# this commands.
ADB_ROOT_RETRY_ATTMEPTS = 3
ADB_ROOT_RETRY_ATTEMPT_INTERVAL_SEC = 10

# Qualified class name of the default instrumentation test runner.
DEFAULT_INSTRUMENTATION_RUNNER = 'com.android.common.support.test.runner.AndroidJUnitRunner'

# Adb getprop call should never take too long.
DEFAULT_GETPROP_TIMEOUT_SEC = 5
DEFAULT_GETPROPS_ATTEMPTS = 3
DEFAULT_GETPROPS_RETRY_SLEEP_SEC = 1

# The regex pattern indicating the `adb connect` command did not fail.
PATTERN_ADB_CONNECT_SUCCESS = re.compile(
    r'^connected to .*|^already connected to .*')


class Error(Exception):
  """Base error type for adb proxy module."""


class AdbError(Error):
  """Raised when an adb command encounters an error.

  Attributes:
    cmd: list of strings, the adb command executed.
    stdout: byte string, the raw stdout of the command.
    stderr: byte string, the raw stderr of the command.
    ret_code: int, the return code of the command.
    serial: string, the serial of the device the command is executed on.
      This is an empty string if the adb command is not specific to a
      device.
  """

  def __init__(self, cmd, stdout, stderr, ret_code, serial=''):
    super().__init__()
    self.cmd = cmd
    self.stdout = stdout
    self.stderr = stderr
    self.ret_code = ret_code
    self.serial = serial

  def __str__(self):
    return ('Error executing adb cmd "%s". ret: %d, stdout: %s, stderr: %s') % (
        utils.cli_cmd_to_string(
            self.cmd), self.ret_code, self.stdout, self.stderr)


class AdbTimeoutError(Error):
  """Raised when an command did not complete within expected time.

  Attributes:
    cmd: list of strings, the adb command that timed out
    timeout: float, the number of seconds passed before timing out.
    serial: string, the serial of the device the command is executed on.
      This is an empty string if the adb command is not specific to a
      device.
  """

  def __init__(self, cmd, timeout, serial=''):
    super().__init__()
    self.cmd = cmd
    self.timeout = timeout
    self.serial = serial

  def __str__(self):
    return 'Timed out executing command "%s" after %ss.' % (
        utils.cli_cmd_to_string(self.cmd), self.timeout)


def is_adb_available():
  """Checks if adb is available as a command line tool.

  Returns:
    True if adb binary is available in console, False otherwise.
  """
  ret, out, err = utils.run_command('which adb', shell=True)
  clean_out = out.decode('utf-8').strip()
  if clean_out:
    return True
  return False


def list_occupied_adb_ports():
  """Lists all the host ports occupied by adb forward.

  This is useful because adb will silently override the binding if an attempt
  to bind to a port already used by adb was made, instead of throwing binding
  error. So one should always check what ports adb is using before trying to
  bind to a port with adb.

  Returns:
    A list of integers representing occupied host ports.
  """
  out = AdbProxy().forward('--list')
  clean_lines = str(out, 'utf-8').strip().split('\n')
  used_ports = []
  for line in clean_lines:
    tokens = line.split(' tcp:')
    if len(tokens) != 3:
      continue
    used_ports.append(int(tokens[1]))
  return used_ports


class AdbProxy:
  """Proxy class for ADB.

  For syntactic reasons, the '-' in adb commands need to be replaced with
  '_'. Can directly execute adb commands on an object:
  >> adb = AdbProxy(<serial>)
  >> adb.start_server()
  >> adb.devices() # will return the console output of "adb devices".

  By default, command args are expected to be an iterable which is passed
  directly to subprocess.Popen():
  >> adb.shell(['echo', 'a', 'b'])

  This way of launching commands is recommended by the subprocess
  documentation to avoid shell injection vulnerabilities and avoid having to
  deal with multiple layers of shell quoting and different shell environments
  between different OSes.

  If you really want to run the command through the system shell, this is
  possible by supplying shell=True, but try to avoid this if possible:
  >> adb.shell('cat /foo > /tmp/file', shell=True)
  """

  def __init__(self, serial=''):
    self.serial = serial

  def _exec_cmd(self, args, shell, timeout, stderr) -> bytes:
    """Executes adb commands.

    Args:
      args: string or list of strings, program arguments.
        See subprocess.Popen() documentation.
      shell: bool, True to run this command through the system shell,
        False to invoke it directly. See subprocess.Popen() docs.
      timeout: float, the number of seconds to wait before timing out.
        If not specified, no timeout takes effect.
      stderr: a Byte stream, like io.BytesIO, stderr of the command will
        be written to this object if provided.

    Returns:
      The output of the adb command run if exit code is 0.

    Raises:
      ValueError: timeout value is invalid.
      AdbError: The adb command exit code is not 0.
      AdbTimeoutError: The adb command timed out.
    """
    if timeout and timeout <= 0:
      raise ValueError('Timeout is not a positive value: %s' % timeout)
    try:
      (ret, out, err) = utils.run_command(args, shell=shell, timeout=timeout)
    except subprocess.TimeoutExpired:
      raise AdbTimeoutError(cmd=args, timeout=timeout, serial=self.serial)

    if stderr:
      stderr.write(err)
    logging.debug('cmd: %s, stdout: %s, stderr: %s, ret: %s',
                  utils.cli_cmd_to_string(args), out, err, ret)
    if ret == 0:
      return out
    else:
      raise AdbError(cmd=args,
                     stdout=out,
                     stderr=err,
                     ret_code=ret,
                     serial=self.serial)

  def _execute_and_process_stdout(self, args, shell, handler) -> bytes:
    """Executes adb commands and processes the stdout with a handler.

    Args:
      args: string or list of strings, program arguments.
        See subprocess.Popen() documentation.
      shell: bool, True to run this command through the system shell,
        False to invoke it directly. See subprocess.Popen() docs.
      handler: func, a function to handle adb stdout line by line.

    Returns:
      The stderr of the adb command run if exit code is 0.

    Raises:
      AdbError: The adb command exit code is not 0.
    """
    proc = subprocess.Popen(args,
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            shell=shell,
                            bufsize=1)
    out = '[elided, processed via handler]'
    try:
      # Even if the process dies, stdout.readline still works
      # and will continue until it runs out of stdout to process.
      while True:
        line = proc.stdout.readline()
        if line:
          handler(line)
        else:
          break
    finally:
      # Note, communicate will not contain any buffered output.
      (unexpected_out, err) = proc.communicate()
      if unexpected_out:
        out = '[unexpected stdout] %s' % unexpected_out
        for line in unexpected_out.splitlines():
          handler(line)

    ret = proc.returncode
    logging.debug('cmd: %s, stdout: %s, stderr: %s, ret: %s',
                  utils.cli_cmd_to_string(args), out, err, ret)
    if ret == 0:
      return err
    else:
      raise AdbError(cmd=args, stdout=out, stderr=err, ret_code=ret)

  def _construct_adb_cmd(self, raw_name, args, shell):
    """Constructs an adb command with arguments for a subprocess call.

    Args:
      raw_name: string, the raw unsanitized name of the adb command to
        format.
      args: string or list of strings, arguments to the adb command.
        See subprocess.Proc() documentation.
      shell: bool, True to run this command through the system shell,
        False to invoke it directly. See subprocess.Proc() docs.

    Returns:
      The adb command in a format appropriate for subprocess. If shell is
        True, then this is a string; otherwise, this is a list of
        strings.
    """
    args = args or ''
    name = raw_name.replace('_', '-')
    if shell:
      args = utils.cli_cmd_to_string(args)
      # Add quotes around "adb" in case the ADB path contains spaces. This
      # is pretty common on Windows (e.g. Program Files).
      if self.serial:
        adb_cmd = '"%s" -s "%s" %s %s' % (ADB, self.serial, name, args)
      else:
        adb_cmd = '"%s" %s %s' % (ADB, name, args)
    else:
      adb_cmd = [ADB]
      if self.serial:
        adb_cmd.extend(['-s', self.serial])
      adb_cmd.append(name)
      if args:
        if isinstance(args, str):
          adb_cmd.append(args)
        else:
          adb_cmd.extend(args)
    return adb_cmd

  def _exec_adb_cmd(self, name, args, shell, timeout, stderr) -> bytes:
    adb_cmd = self._construct_adb_cmd(name, args, shell=shell)
    out = self._exec_cmd(adb_cmd, shell=shell, timeout=timeout, stderr=stderr)
    return out

  def _execute_adb_and_process_stdout(self, name, args, shell, handler) -> bytes:
    adb_cmd = self._construct_adb_cmd(name, args, shell=shell)
    err = self._execute_and_process_stdout(adb_cmd,
                                           shell=shell,
                                           handler=handler)
    return err

  def _parse_getprop_output(self, output):
    """Parses the raw output of `adb shell getprop` into a dictionary.

    Args:
      output: byte str, the raw output of the `adb shell getprop` call.

    Returns:
      dict, name-value pairs of the properties.
    """
    output = output.decode('utf-8', errors='ignore').replace('\r\n', '\n')
    results = {}
    for line in output.split(']\n'):
      if not line:
        continue
      try:
        name, value = line.split(': ', 1)
      except ValueError:
        logging.debug('Failed to parse adb getprop line %s', line)
        continue
      name = name.strip()[1:-1]
      # Remove any square bracket from either end of the value string.
      if value and value[0] == '[':
        value = value[1:]
      results[name] = value
    return results

  @property
  def current_user_id(self) -> int:
    """The integer ID of the current Android user.

    Some adb commands require specifying a user ID to work properly. Use
    this to get the current user ID.

    Note a "user" is not the same as an "account" in Android. See AOSP's
    documentation for details.
    https://source.android.com/devices/tech/admin/multi-user
    """
    sdk_int = int(self.getprop('ro.build.version.sdk'))
    if sdk_int >= 24:
      return int(self.shell(['am', 'get-current-user']))
    if sdk_int >= 21:
      user_info_str = self.shell(['dumpsys', 'user']).decode('utf-8')
      return int(re.findall(r'\{(\d+):', user_info_str)[0])
    # Multi-user is not supported in SDK < 21, only user 0 exists.
    return 0

  def connect(self, address) -> bytes:
    """Executes the `adb connect` command with proper status checking.

    Args:
      address: string, the address of the Android instance to connect to.

    Returns:
      The stdout content.

    Raises:
      AdbError: if the connection failed.
    """
    stdout = self._exec_adb_cmd('connect',
                                address,
                                shell=False,
                                timeout=None,
                                stderr=None)
    if PATTERN_ADB_CONNECT_SUCCESS.match(stdout.decode('utf-8')) is None:
      raise AdbError(cmd=f'connect {address}',
                     stdout=stdout,
                     stderr='',
                     ret_code=0)
    return stdout

  def getprop(self, prop_name):
    """Get a property of the device.

    This is a convenience wrapper for `adb shell getprop xxx`.

    Args:
      prop_name: A string that is the name of the property to get.

    Returns:
      A string that is the value of the property, or None if the property
      doesn't exist.
    """
    return self.shell(
        ['getprop', prop_name],
        timeout=DEFAULT_GETPROP_TIMEOUT_SEC).decode('utf-8').strip()

  def getprops(self, prop_names):
    """Get multiple properties of the device.

    This is a convenience wrapper for `adb shell getprop`. Use this to
    reduce the number of adb calls when getting multiple properties.

    Args:
      prop_names: list of strings, the names of the properties to get.

    Returns:
      A dict containing name-value pairs of the properties requested, if
      they exist.
    """
    attempts = DEFAULT_GETPROPS_ATTEMPTS
    results = {}
    for attempt in range(attempts):
      # The ADB getprop command can randomly return empty string, so try
      # multiple times. This value should always be non-empty if the device
      # in a working state.
      raw_output = self.shell(['getprop'], timeout=DEFAULT_GETPROP_TIMEOUT_SEC)
      properties = self._parse_getprop_output(raw_output)
      if properties:
        for name in prop_names:
          if name in properties:
            results[name] = properties[name]
        break
      # Don't call sleep on the last attempt.
      if attempt < attempts - 1:
        time.sleep(DEFAULT_GETPROPS_RETRY_SLEEP_SEC)
    return results

  def has_shell_command(self, command) -> bool:
    """Checks to see if a given check command exists on the device.

    Args:
      command: A string that is the name of the command to check.

    Returns:
      A boolean that is True if the command exists and False otherwise.
    """
    try:
      output = self.shell(['command', '-v', command]).decode('utf-8').strip()
      return command in output
    except AdbError:
      # If the command doesn't exist, then 'command -v' can return
      # an exit code > 1.
      return False

  def forward(self, args=None, shell=False) -> bytes:
    with ADB_PORT_LOCK:
      return self._exec_adb_cmd('forward',
                                args,
                                shell,
                                timeout=None,
                                stderr=None)

  def instrument(self, package, options=None, runner=None, handler=None) -> bytes:
    """Runs an instrumentation command on the device.

    This is a convenience wrapper to avoid parameter formatting.

    Example:

    .. code-block:: python

      device.instrument(
        'com.my.package.test',
        options = {
          'class': 'com.my.package.test.TestSuite',
        },
      )

    Args:
      package: string, the package of the instrumentation tests.
      options: dict, the instrumentation options including the test
        class.
      runner: string, the test runner name, which defaults to
        DEFAULT_INSTRUMENTATION_RUNNER.
      handler: optional func, when specified the function is used to parse
        the instrumentation stdout line by line as the output is
        generated; otherwise, the stdout is simply returned once the
        instrumentation is finished.

    Returns:
      The stdout of instrumentation command or the stderr if the handler
        is set.
    """
    if runner is None:
      runner = DEFAULT_INSTRUMENTATION_RUNNER
    if options is None:
      options = {}

    options_list = []
    for option_key, option_value in options.items():
      options_list.append('-e %s %s' % (option_key, option_value))
    options_string = ' '.join(options_list)

    instrumentation_command = 'am instrument -r -w %s %s/%s' % (options_string,
                                                                package, runner)
    logging.info('AndroidDevice|%s: Executing adb shell %s', self.serial,
                 instrumentation_command)
    if handler is None:
      return self._exec_adb_cmd('shell',
                                instrumentation_command,
                                shell=False,
                                timeout=None,
                                stderr=None)
    else:
      return self._execute_adb_and_process_stdout('shell',
                                                  instrumentation_command,
                                                  shell=False,
                                                  handler=handler)

  def root(self) -> bytes:
    """Enables ADB root mode on the device.

    This method will retry to execute the command `adb root` when an
    AdbError occurs, since sometimes the error `adb: unable to connect
    for root: closed` is raised when executing `adb root` immediately after
    the device is booted to OS.

    Returns:
      A string that is the stdout of root command.

    Raises:
      AdbError: If the command exit code is not 0.
    """
    for attempt in range(ADB_ROOT_RETRY_ATTMEPTS):
      try:
        return self._exec_adb_cmd('root',
                                  args=None,
                                  shell=False,
                                  timeout=None,
                                  stderr=None)
      except AdbError as e:
        if attempt + 1 < ADB_ROOT_RETRY_ATTMEPTS:
          logging.debug('Retry the command "%s" since Error "%s" occurred.' %
                        (utils.cli_cmd_to_string(
                            e.cmd), e.stderr.decode('utf-8').strip()))
          # Buffer between "adb root" commands.
          time.sleep(ADB_ROOT_RETRY_ATTEMPT_INTERVAL_SEC)
        else:
          raise e

  def __getattr__(self, name):

    def adb_call(args=None, shell=False, timeout=None, stderr=None) -> bytes:
      """Wrapper for an ADB command.

      Args:
        args: string or list of strings, arguments to the adb command.
          See subprocess.Proc() documentation.
        shell: bool, True to run this command through the system shell,
          False to invoke it directly. See subprocess.Proc() docs.
        timeout: float, the number of seconds to wait before timing out.
          If not specified, no timeout takes effect.
        stderr: a Byte stream, like io.BytesIO, stderr of the command
          will be written to this object if provided.

      Returns:
        The output of the adb command run if exit code is 0.
      """
      return self._exec_adb_cmd(name,
                                args,
                                shell=shell,
                                timeout=timeout,
                                stderr=stderr)

    return adb_call