tools/adb.py

#!/usr/bin/env python3
# Copyright (c) 2020, the R8 project authors. Please see the AUTHORS file
# for details. All rights reserved. Use of this source code is governed by a
# BSD-style license that can be found in the LICENSE file.

import subprocess
import time
import utils


def install_apk_on_emulator(apk, emulator_id, quiet=False):
    cmd = ['adb', '-s', emulator_id, 'install', '-r', '-d', apk]
    if quiet:
        subprocess.check_output(cmd)
    else:
        subprocess.check_call(cmd)


def uninstall_apk_on_emulator(app_id, emulator_id):
    process = subprocess.Popen(['adb', '-s', emulator_id, 'uninstall', app_id],
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
    stdout, stderr = process.communicate()
    stdout = stdout.decode('UTF-8')
    stderr = stderr.decode('UTF-8')

    if stdout.strip() == 'Success':
        # Successfully uninstalled
        return

    if 'Unknown package: {}'.format(app_id) in stderr:
        # Application not installed
        return

    # Check if the app is listed in packages
    packages = subprocess.check_output(
        ['adb', 'shell', 'pm', 'list', 'packages'])
    if not 'package:' + app_id in packages:
        return

    raise Exception(
        'Unexpected result from `adb uninstall {}\nStdout: {}\nStderr: {}'.
        format(app_id, stdout, stderr))


def wait_for_emulator(emulator_id):
    stdout = subprocess.check_output(['adb', 'devices']).decode('UTF-8')
    if '{}\tdevice'.format(emulator_id) in stdout:
        return True

    print('Emulator \'{}\' not connected; waiting for connection'.format(
        emulator_id))

    time_waited = 0
    while True:
        time.sleep(10)
        time_waited += 10
        stdout = subprocess.check_output(['adb', 'devices']).decode('UTF-8')
        if '{}\tdevice'.format(emulator_id) not in stdout:
            print('... still waiting for connection')
            if time_waited >= 5 * 60:
                return False
        else:
            return True


def run_monkey(app_id, emulator_id, apk, monkey_events, quiet, enable_logging):
    if not wait_for_emulator(emulator_id):
        return False

    install_apk_on_emulator(apk, emulator_id, quiet)

    # Intentionally using a constant seed such that the monkey generates the same
    # event sequence for each shrinker.
    random_seed = 42

    cmd = [
        'adb', '-s', emulator_id, 'shell', 'monkey', '-p', app_id, '-s',
        str(random_seed),
        str(monkey_events)
    ]

    try:
        stdout = utils.RunCmd(cmd, quiet=quiet, logging=enable_logging)
        succeeded = ('Events injected: {}'.format(monkey_events) in stdout)
    except subprocess.CalledProcessError as e:
        succeeded = False

    uninstall_apk_on_emulator(app_id, emulator_id)

    return succeeded


def run_instrumented(app_id,
                     test_id,
                     emulator_id,
                     apk,
                     test_apk,
                     quiet,
                     enable_logging,
                     test_runner='androidx.test.runner.AndroidJUnitRunner'):
    if not wait_for_emulator(emulator_id):
        return None

    install_apk_on_emulator(apk, emulator_id, quiet)
    install_apk_on_emulator(test_apk, emulator_id, quiet)

    cmd = [
        'adb', '-s', emulator_id, 'shell', 'am', 'instrument', '-w',
        '{}/{}'.format(test_id, test_runner)
    ]

    try:
        stdout = utils.RunCmd(cmd, quiet=quiet, logging=enable_logging)
        # The runner will print OK (X tests) if completed succesfully
        succeeded = any("OK (" in s for s in stdout)
    except subprocess.CalledProcessError as e:
        succeeded = False

    uninstall_apk_on_emulator(test_id, emulator_id)
    uninstall_apk_on_emulator(app_id, emulator_id)

    return succeeded