deploy.py

#!/usr/bin/env python3

import os
import time
from uiautomator import Device
import re
import sys
import subprocess
import pathlib


VWS_CREDENTIALS = {
    'user': 'fairphonetesting@gmail.com',
    'password': 'aish3echi:uwaiSh'
}


PREBUILTS_PATH = '../../vendor/smartviser/viser/prebuilts/apk'

PREBUILT_PROXY_APK_PATTERN = (
    'com.lunarlabs.panda.proxy-latest-sdk{sdk}-{flavour}.apk')

PREBUILT_APKS = [
    'com.smartviser.demogame-latest.apk',
    'com.lunarlabs.panda-latest.apk',
]

FAIRPHONE_WIFI_NETWORKS = {
    'Fairphone Guest': {'security': 'WPA/WPA2 PSK', 'password': 'fairwifi'},
    'Fairphone DEV (2.4 GHz)': {
        'security': 'WPA/WPA2 PSK', 'password': 'fdev@adm'},
    'Fairphone DEV (5 GHz)': {
        'security': 'WPA/WPA2 PSK', 'password': 'fdev@adm'},
}

ADB_DEVICES_PATTERN = re.compile(r'^([a-z0-9-]+)\s+device$', flags=re.M)


class HostCommandError(BaseException):
    """An error happened while issuing a command on the host."""
    def __init__(self, command, error_message):
        self.command = command
        self.error_message = error_message
        message = 'Command `{}` failed: {}'.format(command, error_message)
        super(HostCommandError, self).__init__(message)


class DeviceCommandError(BaseException):
    """An error happened while sending a command to a device."""
    def __init__(self, serial, command, error_message):
        self.serial = serial
        self.command = command
        self.error_message = error_message
        message = 'Command `{}` failed on {}: {}'.format(
            command, serial, error_message)
        super(DeviceCommandError, self).__init__(message)


def adb(*args, serial = None, raise_on_error = True):
    """Run ADB command attached to serial.

    Example:
    >>> process = adb('shell', 'getprop', 'ro.build.fingerprint', serial='cc60c021')
    >>> process.returncode
    0
    >>> process.stdout.strip()
    'Fairphone/FP2/FP2:6.0.1/FP2-gms-18.02.0/FP2-gms-18.02.0:user/release-keys'

    :param *args:
        List of options to ADB (including command).
    :param str serial:
        Identifier for ADB connection to device.
    :param raise_on_error bool:
        Whether to raise a DeviceCommandError exception if the return code is
        less than 0.
    :returns subprocess.CompletedProcess:
        Completed process.
    :raises DeviceCommandError:
        If the command failed.
    """

    # Make sure the adb server is started to avoid the infamous "out of date"
    # message that pollutes stdout.
    ret = subprocess.run(
        ['adb', 'start-server'], stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        universal_newlines=True)
    if ret.returncode < 0:
        if raise_on_error:
            raise DeviceCommandError(
                serial if serial else '??', str(args), ret.stderr)
        else:
            return None

    command = ['adb']
    if serial:
        command += ['-s', serial]
    if args:
        command += list(args)
    ret = subprocess.run(
        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        universal_newlines=True)

    if raise_on_error and ret.returncode < 0:
        raise DeviceCommandError(
            serial if serial else '??', str(args), ret.stderr)

    return ret


def list_devices():
    """List serial numbers of devices attached to adb.

    Raises:
        DeviceCommandError: If the underlying adb command failed.
    """
    process = adb('devices')
    return ADB_DEVICES_PATTERN.findall(process.stdout)


def aapt(*args, raise_on_error = True):
    """Run an AAPT command.

    :param *args:
        The AAPT command with its options.
    :param raise_on_error bool:
        Whether to raise a DeviceCommandError exception if the return code is
        less than 0.
    :returns subprocess.CompletedProcess:
        Completed process.
    :raises HostCommandError:
        If the command failed.
    """
    command = ['aapt']
    if args:
        command += list(args)
    ret = subprocess.run(
        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
        universal_newlines=True)

    if raise_on_error and ret.returncode < 0:
        raise HostCommandError(str(args), ret.stderr)

    return ret


def force_awake(serial, always=True):
    """Force the device to stay awake.

    Raises:
        DeviceCommandError: If the underlying adb command failed.
    """
    adb('shell', 'svc power stayon {}'.format(
        'true' if always else 'false'), serial=serial)


def unlock(dut):
    """Wake-up the device and unlock it.

    Raises:
        DeviceCommandError: If the underlying adb commands failed.
    """
    if not dut.info['screenOn']:
        adb('shell', 'input keyevent KEYCODE_POWER', serial=dut.serial)
        time.sleep(1)
    # The KEYCODE_MENU input is enough to unlock a "swipe up to unlock"
    # lockscreen on Android 6, but unfortunately not Android 7. So we use a
    # swipe up (that depends on the screen resolution) instead.
    adb('shell', 'input touchscreen swipe 930 880 930 380', serial=dut.serial)
    time.sleep(1)
    adb('shell', 'input keyevent KEYCODE_HOME', serial=dut.serial)


def getprop(serial, key):
    """Get system property of device.

    Example:
    >>> getprop('167eb6e8', 'ro.build.id')
    'FP2-gms-18.02.0'

    :param str serial:
        Identifier for ADB connection to device.
    :param str key:
        Key of property to get.
    :returns str:
        Value of system property.
    :raise DeviceCommandError: If the underlying adb command failed.
    """
    process = adb('shell', 'getprop', key, serial=serial)
    return process.stdout.strip()


def is_gms_device(serial):
    """Test if device runs GMS or sibon.

    Example:
    >>> is_gms_device('167eb6e8')
    True

    :param str serial:
        Identifier for ADB connection to device.
    :returns bool:
        True if device runs GMS, false otherwise.
    :raise DeviceCommandError: If the underlying adb command failed.
    """
    return (
        getprop(serial, 'ro.build.id').startswith('FP2-gms-')
        or getprop(serial, 'ro.build.version.incremental').startswith('gms-'))


def uninstall_apk(serial, filename, prebuilts_dir):
    """Uninstall apk from prebuilts_dir on device.

    Raises:
        ValueError: If the package name could not be read from the apk.
        DeviceCommandError: If the uninstall command failed.
    """
    ret = aapt('dump', 'badging', '{}/{}'.format(prebuilts_dir, filename))
    package = None
    for line in ret.stdout.splitlines():
        if line.startswith('package'):
            for token in line.split(' '):
                if token.startswith('name='):
                    # Extract the package name out of the token
                    # (name='some.package.name')
                    package = token[6:-1]
                    break
    if not package:
        raise ValueError('Could not find package of app `{}`'.format(
            filename))

    adb('uninstall', package, serial=serial)


def install_apk(serial, filename, prebuilts_dir):
    """Install apk from prebuilts_dir on device.

    Raises:
        DeviceCommandError: If the install command failed.
    """
    path = os.path.join(prebuilts_dir, filename)
    adb('install', '-r', path, serial=serial)


def configure_wifi_networks(dut, networks):
    """Configure Wi-Fi networks.

        The `networks` parameters is a list of networks to configure hashed by
        their SSID. Each network value should have the following format:
          - security (str): The security value as can be found in the Wi-Fi
            settings dialog. Common values are 'None' and 'WPA/WPA2 PSK'.
          - password (str, optional): The network password if the security is
            'WPA/WPA2 PSK'.

        Parameters:
            dut (Device): The device object.
            networks (dict(dict(str))): The list of networks to configure.
        Raises:
            DeviceCommandError: If the UI automation fails.
    """
    # Open the Wi-Fi settings
    adb('shell', ('am start -a android.settings.WIFI_SETTINGS '
        '--activity-clear-task'), serial=dut.serial)

    # Make sure Wi-Fi is enabled
    wifi_enabler = dut(text='OFF',
        resourceId='com.android.settings:id/switch_widget')
    if wifi_enabler.exists:
        wifi_enabler.click()

    # Check for registered networks
    registered_networks = set()
    dut(description='More options').click()
    time.sleep(1)
    saved_networks = dut(text='Saved networks')
    if saved_networks.exists:
        saved_networks.click()
        for ssid in networks.keys():
            if dut(text=ssid).exists:
                registered_networks.add(ssid)
    dut.press.back()

    missing_networks = networks.keys() - registered_networks

    for ssid in registered_networks:
        print('Ignoring `{}` Wi-Fi network, already configured.'.format(ssid))

    for ssid in missing_networks:
        print('Configuring `{}` Wi-Fi network…'.format(ssid))

        dut(description='More options').click()
        dut(text='Add network').click()
        dut(resourceId='com.android.settings:id/ssid') \
            .set_text(ssid)
        dut(resourceId='com.android.settings:id/security') \
            .click()
        dut(text=networks[ssid]['security']) \
            .click()
        password_field = dut(resourceId='com.android.settings:id/password')
        if 'password' in networks[ssid] and networks[ssid]['password']:
            if not password_field.exists:
                dut(text='Cancel').click()
                raise DeviceCommandError(dut, 'UI: add Wi-Fi',
                    'missing password field')
            password_field.set_text(networks[ssid]['password'])
        elif password_field.exists:
            dut(text='Cancel').click()
            raise DeviceCommandError(dut, 'UI: add Wi-Fi',
                'missing password data')
        save_button = dut(text='Save')
        if not save_button.click():
            dut(text='Cancel').click()
            raise DeviceCommandError(dut, 'UI: add Wi-Fi',
                'could not save network')

    # Force the Wi-Fi on and off to pick the best network available
    dut(text='ON', resourceId='com.android.settings:id/switch_widget').click()
    dut(text='OFF', resourceId='com.android.settings:id/switch_widget').click()

    # Leave the settings
    dut.press.back()


def get_proxy_apk(android_sdk, flavour):
    if android_sdk >= 24:
        return PREBUILT_PROXY_APK_PATTERN.format(sdk=24, flavour=flavour)
    else:
        return PREBUILT_PROXY_APK_PATTERN.format(sdk=19, flavour=flavour)

# Prepare the DUT
def prepare_dut(dut, scenarios_dir, data_dir, prebuilts_dir):
    flavour = 'gms' if is_gms_device(dut.serial) else 'sibon'
    proxy_apk = get_proxy_apk(dut.sdk, flavour)

    # Uninstall the smartviser apps
    for app in PREBUILT_APKS + [proxy_apk]:
        print('Uninstalling `{}`…'.format(app))
        uninstall_apk(dut.serial, app, prebuilts_dir)

    # Copy the scenarios
    print('Pushing scenarios from `{}`…'.format(scenarios_dir))
    adb('push', scenarios_dir, '/sdcard/viser', serial=dut.serial)

    # Copy the scenarios data
    print('Pushing scenarios data from `{}`…'.format(data_dir))
    adb('push', data_dir, '/sdcard/viser/data', serial=dut.serial)

    # Install the smartviser apps (starting with the proxy app)
    for app in [proxy_apk] + PREBUILT_APKS:
        print('Installing `{}`…'.format(app))
        install_apk(dut.serial, app, prebuilts_dir)


# Grant the permissions through the UI
def configure_perms(dut):
    dut() \
        .child_by_text('You need to grant access for the Permissions Group '
                'Calendar Camera Contacts Microphone SMS Storage Telephone Location',
                resourceId='android:id/content') \
        .child(text='OK', className='android.widget.Button') \
        .click()

    # Accept any permission that is asked for
    prompt = dut(resourceId='com.android.packageinstaller:id/permission_allow_button',
            className='android.widget.Button')
    while prompt.exists:
        prompt.click()

    # Input the credentials
    dut(resourceId='android:id/content') \
        .child(text='Username') \
        .child(className='android.widget.EditText') \
        .set_text(VWS_CREDENTIALS['user'])
    dut(resourceId='android:id/content') \
        .child(text='Password') \
        .child(className='android.widget.EditText') \
        .set_text(VWS_CREDENTIALS['password'])

    # Sign in
    signin_label = 'SIGN IN' if dut.sdk >= 24 else 'Sign in'
    dut(resourceId='android:id/content') \
        .child(text=signin_label, className='android.widget.Button') \
        .click()

def configure_sms(dut):
    # TODO wait for the connection to be established and time-out
    prompt = dut(resourceId='android:id/content') \
        .child(text='Viser must be your SMS app to send messages')
    while not prompt.exists:
        time.sleep(1)

    # Make viser the default SMS app
    dut(resourceId='android:id/content') \
        .child_by_text('Viser must be your SMS app to send messages',
                className='android.widget.LinearLayout') \
        .child(text='OK', className='android.widget.Button') \
        .click()

    dut(resourceId='android:id/content') \
        .child_by_text('Change SMS app?', className='android.widget.LinearLayout') \
        .child(text='Yes', className='android.widget.Button') \
        .click()

def configure_settings(dut):
    # Set the e-mail account
    dut(text='Settings', className='android.widget.TextView') \
        .click()
    dut(resourceId='android:id/list') \
        .child_by_text('User settings', className='android.widget.LinearLayout') \
        .click()
    dut(resourceId='android:id/list') \
        .child_by_text('Email account', className='android.widget.LinearLayout') \
        .click()
    prompt = dut(resourceId='android:id/content') \
        .child_by_text('Email account', className='android.widget.LinearLayout')
    prompt.child(resourceId='android:id/edit') \
        .set_text('fairphone.viser@gmail.com')
    prompt.child(text='OK', className='android.widget.Button') \
        .click()
    dut(resourceId='android:id/list') \
        .child_by_text('Email password', className='android.widget.LinearLayout') \
        .click()
    dut(text='Password :') \
        .child(className='android.widget.EditText') \
        .set_text('fairphoneviser2017')
    dut(description='OK', className='android.widget.TextView') \
        .click()
    dut.press.back()
    dut.press.back()


def deploy():
    serials = []

    if len(sys.argv) > 1:
        serials.append(sys.argv[1])
    else:
        serials = list_devices()

    for serial in serials:
        print('Configuring device {}…'.format(serial))

        dut = Device(serial)
        # Work around the not-so-easy Device class
        dut.serial = serial
        # Cache the Android SDK version (dut.info fetches system properties)
        dut.sdk = dut.info['sdkInt']

        try:
            # Make sure the screen stays on - we're going to use UI automation
            force_awake(serial)
            unlock(dut)

            # Configure common Fairphone Wi-Fi networks
            if dut.sdk < 24:
                configure_wifi_networks(dut, FAIRPHONE_WIFI_NETWORKS)
            else:
                print('Uh oh, the device is running Android SDK {} on which we '
                    'do not deploy Wi-Fi networks yet.'.format(dut.sdk))

            # Push the scenarios, their data, and install the apps
            prepare_dut(
                dut, '../scenarios', '../scenarios-data', PREBUILTS_PATH)

            # Start the viser app
            adb(
                'shell', 'monkey', '-p', 'com.lunarlabs.panda', '-c',
                'android.intent.category.LAUNCHER', '1', serial=serial)

            configure_perms(dut)

            # TODO DO NOT DO THE FOLLOWING IF NO SIM CARD IS IN THE DUT
            # time.sleep(10)
            # configure_sms(dut)

            configure_settings(dut)
        except (HostCommandError, DeviceCommandError) as e:
            print('ERROR {}'.format(e), file=sys.stderr)
        finally:
            try:
                # Leave the device alone now
                force_awake(serial, always=False)
            except DeviceCommandError as e:
                print('WARNING {}'.format(e), file=sys.stderr)


if __name__ == '__main__':
    deploy()