devlib/utils/android.py

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


"""
Utility functions for working with Android devices through adb.

"""
# pylint: disable=E1103
import os
import pexpect
import time
import subprocess
import logging
import re
import threading
import tempfile
import Queue
from collections import defaultdict

from devlib.exception import TargetError, HostError, DevlibError
from devlib.utils.misc import check_output, which, memoized, ABI_MAP
from devlib.utils.misc import escape_single_quotes, escape_double_quotes
from devlib import host


logger = logging.getLogger('android')

MAX_ATTEMPTS = 5
AM_START_ERROR = re.compile(r"Error: Activity.*")

# See:
# http://developer.android.com/guide/topics/manifest/uses-sdk-element.html#ApiLevels
ANDROID_VERSION_MAP = {
    23: 'MARSHMALLOW',
    22: 'LOLLYPOP_MR1',
    21: 'LOLLYPOP',
    20: 'KITKAT_WATCH',
    19: 'KITKAT',
    18: 'JELLY_BEAN_MR2',
    17: 'JELLY_BEAN_MR1',
    16: 'JELLY_BEAN',
    15: 'ICE_CREAM_SANDWICH_MR1',
    14: 'ICE_CREAM_SANDWICH',
    13: 'HONEYCOMB_MR2',
    12: 'HONEYCOMB_MR1',
    11: 'HONEYCOMB',
    10: 'GINGERBREAD_MR1',
    9: 'GINGERBREAD',
    8: 'FROYO',
    7: 'ECLAIR_MR1',
    6: 'ECLAIR_0_1',
    5: 'ECLAIR',
    4: 'DONUT',
    3: 'CUPCAKE',
    2: 'BASE_1_1',
    1: 'BASE',
}


# Initialized in functions near the botton of the file
android_home = None
platform_tools = None
adb = None
aapt = None
fastboot = None


class AndroidProperties(object):

    def __init__(self, text):
        self._properties = {}
        self.parse(text)

    def parse(self, text):
        self._properties = dict(re.findall(r'\[(.*?)\]:\s+\[(.*?)\]', text))

    def iteritems(self):
        return self._properties.iteritems()

    def __iter__(self):
        return iter(self._properties)

    def __getattr__(self, name):
        return self._properties.get(name)

    __getitem__ = __getattr__


class AdbDevice(object):

    def __init__(self, name, status):
        self.name = name
        self.status = status

    def __cmp__(self, other):
        if isinstance(other, AdbDevice):
            return cmp(self.name, other.name)
        else:
            return cmp(self.name, other)

    def __str__(self):
        return 'AdbDevice({}, {})'.format(self.name, self.status)

    __repr__ = __str__


class ApkInfo(object):

    version_regex = re.compile(r"name='(?P<name>[^']+)' versionCode='(?P<vcode>[^']+)' versionName='(?P<vname>[^']+)'")
    name_regex = re.compile(r"name='(?P<name>[^']+)'")

    def __init__(self, path=None):
        self.path = path
        self.package = None
        self.activity = None
        self.label = None
        self.version_name = None
        self.version_code = None
        self.native_code = None
        self.parse(path)

    def parse(self, apk_path):
        _check_env()
        command = [aapt, 'dump', 'badging', apk_path]
        logger.debug(' '.join(command))
        try:
            output = subprocess.check_output(command, stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as e:
            raise HostError('Error parsing APK file {}. `aapt` says:\n{}'
                            .format(apk_path, e.output))
        for line in output.split('\n'):
            if line.startswith('application-label:'):
                self.label = line.split(':')[1].strip().replace('\'', '')
            elif line.startswith('package:'):
                match = self.version_regex.search(line)
                if match:
                    self.package = match.group('name')
                    self.version_code = match.group('vcode')
                    self.version_name = match.group('vname')
            elif line.startswith('launchable-activity:'):
                match = self.name_regex.search(line)
                self.activity = match.group('name')
            elif line.startswith('native-code'):
                apk_abis = [entry.strip() for entry in line.split(':')[1].split("'") if entry.strip()]
                mapped_abis = []
                for apk_abi in apk_abis:
                    found = False
                    for abi, architectures in ABI_MAP.iteritems():
                        if apk_abi in architectures:
                            mapped_abis.append(abi)
                            found = True
                            break
                    if not found:
                        mapped_abis.append(apk_abi)
                self.native_code = mapped_abis
            else:
                pass  # not interested


class AdbConnection(object):

    # maintains the count of parallel active connections to a device, so that
    # adb disconnect is not invoked untill all connections are closed
    active_connections = defaultdict(int)
    default_timeout = 10
    ls_command = 'ls'

    @property
    def name(self):
        return self.device

    @property
    @memoized
    def newline_separator(self):
        output = adb_command(self.device,
                             "shell '({}); echo \"\n$?\"'".format(self.ls_command), adb_server=self.adb_server)
        if output.endswith('\r\n'):
            return '\r\n'
        elif output.endswith('\n'):
            return '\n'
        else:
            raise DevlibError("Unknown line ending")

    # Again, we need to handle boards where the default output format from ls is
    # single column *and* boards where the default output is multi-column.
    # We need to do this purely because the '-1' option causes errors on older
    # versions of the ls tool in Android pre-v7.
    def _setup_ls(self):
        command = "shell '(ls -1); echo \"\n$?\"'"
        try:
            output = adb_command(self.device, command, timeout=self.timeout, adb_server=self.adb_server)
        except subprocess.CalledProcessError as e:
            raise HostError(
                'Failed to set up ls command on Android device. Output:\n'
                + e.output)
        lines = output.splitlines()
        retval = lines[-1].strip()
        if int(retval) == 0:
            self.ls_command = 'ls -1'
        else:
            self.ls_command = 'ls'
        logger.debug("ls command is set to {}".format(self.ls_command))

    def __init__(self, device=None, timeout=None, platform=None, adb_server=None):
        self.timeout = timeout if timeout is not None else self.default_timeout
        if device is None:
            device = adb_get_device(timeout=timeout, adb_server=adb_server)
        self.device = device
        self.adb_server = adb_server
        adb_connect(self.device)
        AdbConnection.active_connections[self.device] += 1
        self._setup_ls()

    def push(self, source, dest, timeout=None):
        if timeout is None:
            timeout = self.timeout
        command = "push '{}' '{}'".format(source, dest)
        if not os.path.exists(source):
            raise HostError('No such file "{}"'.format(source))
        return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)

    def pull(self, source, dest, timeout=None):
        if timeout is None:
            timeout = self.timeout
        # Pull all files matching a wildcard expression
        if os.path.isdir(dest) and \
           ('*' in source or '?' in source):
            command = 'shell {} {}'.format(self.ls_command, source)
            output = adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
            for line in output.splitlines():
                command = "pull '{}' '{}'".format(line.strip(), dest)
                adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
            return
        command = "pull '{}' '{}'".format(source, dest)
        return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)

    def execute(self, command, timeout=None, check_exit_code=False,
                as_root=False, strip_colors=True):
        return adb_shell(self.device, command, timeout, check_exit_code,
                         as_root, self.newline_separator,adb_server=self.adb_server)

    def background(self, command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, as_root=False):
        return adb_background_shell(self.device, command, stdout, stderr, as_root)

    def close(self):
        AdbConnection.active_connections[self.device] -= 1
        if AdbConnection.active_connections[self.device] <= 0:
            adb_disconnect(self.device)
            del AdbConnection.active_connections[self.device]

    def cancel_running_command(self):
        # adbd multiplexes commands so that they don't interfer with each
        # other, so there is no need to explicitly cancel a running command
        # before the next one can be issued.
        pass


def fastboot_command(command, timeout=None, device=None):
    _check_env()
    target = '-s {}'.format(device) if device else ''
    full_command = 'fastboot {} {}'.format(target, command)
    logger.debug(full_command)
    output, _ = check_output(full_command, timeout, shell=True)
    return output


def fastboot_flash_partition(partition, path_to_image):
    command = 'flash {} {}'.format(partition, path_to_image)
    fastboot_command(command)


def adb_get_device(timeout=None, adb_server=None):
    """
    Returns the serial number of a connected android device.

    If there are more than one device connected to the machine, or it could not
    find any device connected, :class:`devlib.exceptions.HostError` is raised.
    """
    # TODO this is a hacky way to issue a adb command to all listed devices

    # Ensure server is started so the 'daemon started successfully' message
    # doesn't confuse the parsing below
    adb_command(None, 'start-server', adb_server=adb_server)

    # The output of calling adb devices consists of a heading line then
    # a list of the devices sperated by new line
    # The last line is a blank new line. in otherwords, if there is a device found
    # then the output length is 2 + (1 for each device)
    start = time.time()
    while True:
        output = adb_command(None, "devices", adb_server=adb_server).splitlines()  # pylint: disable=E1103
        output_length = len(output)
        if output_length == 3:
            # output[1] is the 2nd line in the output which has the device name
            # Splitting the line by '\t' gives a list of two indexes, which has
            # device serial in 0 number and device type in 1.
            return output[1].split('\t')[0]
        elif output_length > 3:
            message = '{} Android devices found; either explicitly specify ' +\
                      'the device you want, or make sure only one is connected.'
            raise HostError(message.format(output_length - 2))
        else:
            if timeout < time.time() - start:
                raise HostError('No device is connected and available')
            time.sleep(1)


def adb_connect(device, timeout=None, attempts=MAX_ATTEMPTS):
    _check_env()
    # Connect is required only for ADB-over-IP
    if "." not in device:
        logger.debug('Device connected via USB, connect not required')
        return
    tries = 0
    output = None
    while tries <= attempts:
        tries += 1
        if device:
            command = 'adb connect {}'.format(device)
            logger.debug(command)
            output, _ = check_output(command, shell=True, timeout=timeout)
        if _ping(device):
            break
        time.sleep(10)
    else:  # did not connect to the device
        message = 'Could not connect to {}'.format(device or 'a device')
        if output:
            message += '; got: "{}"'.format(output)
        raise HostError(message)


def adb_disconnect(device):
    _check_env()
    if not device:
        return
    if ":" in device and device in adb_list_devices():
        command = "adb disconnect " + device
        logger.debug(command)
        retval = subprocess.call(command, stdout=open(os.devnull, 'wb'), shell=True)
        if retval:
            raise TargetError('"{}" returned {}'.format(command, retval))


def _ping(device):
    _check_env()
    device_string = ' -s {}'.format(device) if device else ''
    command = "adb{} shell \"ls / > /dev/null\"".format(device_string)
    logger.debug(command)
    result = subprocess.call(command, stderr=subprocess.PIPE, shell=True)
    if not result:
        return True
    else:
        return False


def adb_shell(device, command, timeout=None, check_exit_code=False,
              as_root=False, newline_separator='\r\n', adb_server=None):  # NOQA
    _check_env()
    if as_root:
        command = 'echo \'{}\' | su'.format(escape_single_quotes(command))
    device_part = []
    if adb_server:
        device_part = ['-H', adb_server]
    device_part += ['-s', device] if device else []

    # On older combinations of ADB/Android versions, the adb host command always
    # exits with 0 if it was able to run the command on the target, even if the
    # command failed (https://code.google.com/p/android/issues/detail?id=3254).
    # Homogenise this behaviour by running the command then echoing the exit
    # code.
    adb_shell_command = '({}); echo \"\n$?\"'.format(command)
    actual_command = ['adb'] + device_part + ['shell', adb_shell_command]
    logger.debug('adb {} shell {}'.format(' '.join(device_part), command))
    raw_output, error = check_output(actual_command, timeout, shell=False)
    if raw_output:
        try:
            output, exit_code, _ = raw_output.rsplit(newline_separator, 2)
        except ValueError:
            exit_code, _ = raw_output.rsplit(newline_separator, 1)
            output = ''
    else:  # raw_output is empty
        exit_code = '969696'  # just because
        output = ''

    if check_exit_code:
        exit_code = exit_code.strip()
        re_search = AM_START_ERROR.findall('{}\n{}'.format(output, error))
        if exit_code.isdigit():
            if int(exit_code):
                message = ('Got exit code {}\nfrom target command: {}\n'
                           'STDOUT: {}\nSTDERR: {}')
                raise TargetError(message.format(exit_code, command, output, error))
            elif re_search:
                message = 'Could not start activity; got the following:\n{}'
                raise TargetError(message.format(re_search[0]))
        else:  # not all digits
            if re_search:
                message = 'Could not start activity; got the following:\n{}'
                raise TargetError(message.format(re_search[0]))
            else:
                message = 'adb has returned early; did not get an exit code. '\
                          'Was kill-server invoked?\nOUTPUT:\n-----\n{}\n'\
                          '-----\nERROR:\n-----\n{}\n-----'
                raise TargetError(message.format(raw_output, error))

    return output


def adb_background_shell(device, command,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE,
                         as_root=False):
    """Runs the sepcified command in a subprocess, returning the the Popen object."""
    _check_env()
    if as_root:
        command = 'echo \'{}\' | su'.format(escape_single_quotes(command))
    device_string = ' -s {}'.format(device) if device else ''
    full_command = 'adb{} shell "{}"'.format(device_string, escape_double_quotes(command))
    logger.debug(full_command)
    return subprocess.Popen(full_command, stdout=stdout, stderr=stderr, shell=True)


def adb_list_devices(adb_server=None):
    output = adb_command(None, 'devices',adb_server=adb_server)
    devices = []
    for line in output.splitlines():
        parts = [p.strip() for p in line.split()]
        if len(parts) == 2:
            devices.append(AdbDevice(*parts))
    return devices


def get_adb_command(device, command, timeout=None,adb_server=None):
    _check_env()
    device_string = ""
    if adb_server != None:
        device_string = ' -H {}'.format(adb_server)
    device_string += ' -s {}'.format(device) if device else ''
    return "adb{} {}".format(device_string, command)

def adb_command(device, command, timeout=None,adb_server=None):
    full_command = get_adb_command(device, command, timeout, adb_server)
    logger.debug(full_command)
    output, _ = check_output(full_command, timeout, shell=True)
    return output

def grant_app_permissions(target, package):
    """
    Grant an app all the permissions it may ask for
    """
    dumpsys = target.execute('dumpsys package {}'.format(package))

    permissions = re.search(
        'requested permissions:\s*(?P<permissions>(android.permission.+\s*)+)', dumpsys
    )
    if permissions is None:
        return
    permissions = permissions.group('permissions').replace(" ", "").splitlines()

    for permission in permissions:
        try:
            target.execute('pm grant {} {}'.format(package, permission))
        except TargetError:
            logger.debug('Cannot grant {}'.format(permission))


# Messy environment initialisation stuff...

class _AndroidEnvironment(object):

    def __init__(self):
        self.android_home = None
        self.platform_tools = None
        self.adb = None
        self.aapt = None
        self.fastboot = None


def _initialize_with_android_home(env):
    logger.debug('Using ANDROID_HOME from the environment.')
    env.android_home = android_home
    env.platform_tools = os.path.join(android_home, 'platform-tools')
    os.environ['PATH'] = env.platform_tools + os.pathsep + os.environ['PATH']
    _init_common(env)
    return env


def _initialize_without_android_home(env):
    adb_full_path = which('adb')
    if adb_full_path:
        env.adb = 'adb'
    else:
        raise HostError('ANDROID_HOME is not set and adb is not in PATH. '
                        'Have you installed Android SDK?')
    logger.debug('Discovering ANDROID_HOME from adb path.')
    env.platform_tools = os.path.dirname(adb_full_path)
    env.android_home = os.path.dirname(env.platform_tools)
    _init_common(env)
    return env


def _init_common(env):
    logger.debug('ANDROID_HOME: {}'.format(env.android_home))
    build_tools_directory = os.path.join(env.android_home, 'build-tools')
    if not os.path.isdir(build_tools_directory):
        msg = '''ANDROID_HOME ({}) does not appear to have valid Android SDK install
                 (cannot find build-tools)'''
        raise HostError(msg.format(env.android_home))
    versions = os.listdir(build_tools_directory)
    for version in reversed(sorted(versions)):
        aapt_path = os.path.join(build_tools_directory, version, 'aapt')
        if os.path.isfile(aapt_path):
            logger.debug('Using aapt for version {}'.format(version))
            env.aapt = aapt_path
            break
    else:
        raise HostError('aapt not found. Please make sure at least one Android '
                        'platform is installed.')


def _check_env():
    global android_home, platform_tools, adb, aapt  # pylint: disable=W0603
    if not android_home:
        android_home = os.getenv('ANDROID_HOME')
        if android_home:
            _env = _initialize_with_android_home(_AndroidEnvironment())
        else:
            _env = _initialize_without_android_home(_AndroidEnvironment())
        android_home = _env.android_home
        platform_tools = _env.platform_tools
        adb = _env.adb
        aapt = _env.aapt

class LogcatMonitor(object):
    """
    Helper class for monitoring Anroid's logcat

    :param target: Android target to monitor
    :type target: :class:`AndroidTarget`

    :param regexps: List of uncompiled regular expressions to filter on the
                    device. Logcat entries that don't match any will not be
                    seen. If omitted, all entries will be sent to host.
    :type regexps: list(str)
    """

    @property
    def logfile(self):
        return self._logfile

    def __init__(self, target, regexps=None):
        super(LogcatMonitor, self).__init__()

        self.target = target
        self._regexps = regexps

    def start(self, outfile=None):
        """
        Start logcat and begin monitoring

        :param outfile: Optional path to file to store all logcat entries
        :type outfile: str
        """
        if outfile:
            self._logfile = open(outfile)
        else:
            self._logfile = tempfile.NamedTemporaryFile()

        self.target.clear_logcat()

        logcat_cmd = 'logcat'

        # Join all requested regexps with an 'or'
        if self._regexps:
            regexp = '{}'.format('|'.join(self._regexps))
            if len(self._regexps) > 1:
                regexp = '({})'.format(regexp)
            logcat_cmd = '{} -e "{}"'.format(logcat_cmd, regexp)

        logcat_cmd = get_adb_command(self.target.conn.device, logcat_cmd)

        logger.debug('logcat command ="{}"'.format(logcat_cmd))
        self._logcat = pexpect.spawn(logcat_cmd, logfile=self._logfile)

    def stop(self):
        self._logcat.terminate()
        self._logfile.close()

    def get_log(self):
        """
        Return the list of lines found by the monitor
        """
        with open(self._logfile.name) as fh:
            return [line for line in fh]

    def clear_log(self):
        with open(self._logfile.name, 'w') as fh:
            pass

    def search(self, regexp):
        """
        Search a line that matches a regexp in the logcat log
        Return immediatly
        """
        return [line for line in self.get_log() if re.match(regexp, line)]

    def wait_for(self, regexp, timeout=30):
        """
        Search a line that matches a regexp in the logcat log
        Wait for it to appear if it's not found

        :param regexp: regexp to search
        :type regexp: str

        :param timeout: Timeout in seconds, before rasing RuntimeError.
                        ``None`` means wait indefinitely
        :type timeout: number

        :returns: List of matched strings
        """
        log = self.get_log()
        res = [line for line in log if re.match(regexp, line)]

        # Found some matches, return them
        if len(res) > 0:
            return res

        # Store the number of lines we've searched already, so we don't have to
        # re-grep them after 'expect' returns
        next_line_num = len(log)

        try:
            self._logcat.expect(regexp, timeout=timeout)
        except pexpect.TIMEOUT:
            raise RuntimeError('Logcat monitor timeout ({}s)'.format(timeout))

        return [line for line in self.get_log()[next_line_num:]
                if re.match(regexp, line)]