deploy.py

#!/usr/bin/env python3

import os
import pathlib
import re
import subprocess
import sys
import time

from uiautomator import Device

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(dut, filename, prebuilts_dir):
    """Install apk from prebuilts_dir on device.

    Raises:
        DeviceCommandError: If the install command failed.
    """
    path = os.path.join(prebuilts_dir, filename)
    command = ["install", "-r"]
    if dut.sdk >= 23:
        # From Marshmallow onwards, adb has a flag to grant default permissions
        command.append("-g")
    adb(*command, path, serial=dut.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.wait()
    time.sleep(1)
    saved_networks = dut(text="Saved networks")
    if saved_networks.exists:
        saved_networks.click.wait()
        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.wait()
        dut(text="Add network").click.wait()
        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")
        time.sleep(1)
        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 disable_privacy_impact_popup(dut):
    """Disable Privacy Impact popup on Android 5.

    This simplifies UI automation. Disabling the feature globally is more robust
    than clicking through the the Privacy Impact screen per app.
    """
    print("Disabling the Privacy Impact screen…")
    adb(
        "shell",
        (
            "am start -a android.intent.action.MAIN "
            "com.fairphone.privacyimpact/.PrivacyImpactPreferenceActivity"
        ),
        serial=dut.serial,
    )
    disable_privacy_impact_checkbox = dut(className="android.widget.CheckBox")
    if not disable_privacy_impact_checkbox.checked:
        disable_privacy_impact_checkbox.click()


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, app, prebuilts_dir)


# Grant the permissions through the UI
def configure_perms(dut):
    # 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)
                )

            # Disable Privacy Impact popup on Android 5.
            if dut.sdk <= 22:
                disable_privacy_impact_popup(dut)

            # 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()