deploy.py

#!/usr/bin/env python3

import logging
import pathlib
import re
import subprocess
import sys
import time

import uiautomator

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"


_LOG = logging.getLogger(__name__)

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

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 disable_privacy_impact_popup(device):
    """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…")
    device.adb(
        "shell",
        (
            "am start -a android.intent.action.MAIN "
            "com.fairphone.privacyimpact/.PrivacyImpactPreferenceActivity"
        ),
    )
    disable_privacy_impact_checkbox = device.ui(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(device, scenarios_dir, data_dir, prebuilts_dir):
    flavour = "gms" if device.is_gms_device() else "sibon"
    proxy_apk = get_proxy_apk(device.sdk, flavour)
    prebuilts_dir = pathlib.Path(prebuilts_dir)

    # Uninstall the smartviser apps
    for app in PREBUILT_APKS + [proxy_apk]:
        print("Uninstalling `{}`…".format(app))
        device.uninstall(prebuilts_dir / app)

    # Copy the scenarios
    print("Pushing scenarios from `{}`…".format(scenarios_dir))
    device.push(scenarios_dir, pathlib.Path("/sdcard/Viser"))

    # Copy the scenarios data
    print("Pushing scenarios data from `{}`…".format(data_dir))
    device.push(data_dir, pathlib.Path("/sdcard/Viser/Data"))

    # Install the smartviser apps (starting with the proxy app)
    for app in [proxy_apk] + PREBUILT_APKS:
        print("Installing `{}`…".format(app))
        device.install(prebuilts_dir / app)


# Grant the permissions through the UI
def configure_perms(device):
    # Input the credentials
    device.ui(resourceId="android:id/content").child(text="Username").child(
        className="android.widget.EditText"
    ).set_text(VWS_CREDENTIALS["user"])
    device.ui(resourceId="android:id/content").child(text="Password").child(
        className="android.widget.EditText"
    ).set_text(VWS_CREDENTIALS["password"])

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


def configure_settings(device):
    # Set the e-mail account
    device.ui(text="Settings", className="android.widget.TextView").click()
    device.ui(resourceId="android:id/list").child_by_text(
        "User settings", className="android.widget.LinearLayout"
    ).click()
    device.ui(resourceId="android:id/list").child_by_text(
        "Email account", className="android.widget.LinearLayout"
    ).click()
    prompt = device.ui(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()
    device.ui(resourceId="android:id/list").child_by_text(
        "Email password", className="android.widget.LinearLayout"
    ).click()
    device.ui(text="Password :").child(className="android.widget.EditText").set_text(
        "fairphoneviser2017"
    )
    device.ui(description="OK", className="android.widget.TextView").click()
    device.ui.press.back()
    device.ui.press.back()


class DeviceUnderTest:
    """An Android device under test."""

    serial: str
    """The device serial number (adb/fastboot)."""
    sdk: int
    """The Android SDK version number."""
    ui: uiautomator.Device
    """The UI Automator handle piloting the device."""

    def __init__(self, serial: str):
        self.serial = serial
        self.ui = uiautomator.Device(serial)

        # Cache the Android SDK version
        self.sdk = self.ui.info["sdkInt"]

    def adb(self, *args) -> subprocess.CompletedProcess:
        """Execute an adb command on this device.

        :returns: The completed process.
        :raise DeviceCommandError: If the underlying adb command failed.
        """
        return adb(*args, serial=self.serial)

    def force_awake(self, always=True) -> None:
        """Force the device to stay awake.

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

    def unlock(self) -> None:
        """Wake-up the device and unlock it.

        :raise DeviceCommandError: If the underlying adb commands failed.
        """
        if not self.ui.info["screenOn"]:
            self.adb("shell", "input keyevent KEYCODE_POWER")
            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.
        self.adb("shell", "input touchscreen swipe 930 880 930 380")
        time.sleep(1)
        self.adb("shell", "input keyevent KEYCODE_HOME")

    def getprop(self, key: str) -> str:
        """Get a system property.

        Example:
        >>> self.getprop('ro.build.id')
        'FP2-gms-18.02.0'

        :param key: Key of property to get.
        :returns: Value of system property.
        :raise DeviceCommandError: If the underlying adb command failed.
        """
        process = self.adb("shell", "getprop", key)
        return process.stdout.strip()

    def is_gms_device(self) -> bool:
        """Whether the device runs GMS or sibon.

        Example:
        >>> self.is_gms_device()
        True

        :returns: True if device runs GMS, false otherwise.
        :raise DeviceCommandError: If the underlying adb command failed.
        """
        return self.getprop("ro.build.id").startswith("FP2-gms-") or self.getprop(
            "ro.build.version.incremental"
        ).startswith("gms-")

    def uninstall(self, apk: pathlib.Path) -> None:
        """Uninstall an app from an APK file.

        :raise ValueError: If the package name could not be read from the apk.
        :raise DeviceCommandError: If the uninstall command failed.
        """
        ret = aapt("dump", "badging", str(apk))
        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(apk.name))

        self.adb("uninstall", package)

    def install(self, apk: pathlib.Path) -> None:
        """Install an app from an APK file.

        :raise DeviceCommandError: If the install command failed.
        """
        command = ["install", "-r"]
        if self.sdk >= 23:
            # From Marshmallow onwards, adb has a flag to grant default permissions
            command.append("-g")

        self.adb(*command, str(apk))

    def push(self, source: pathlib.Path, target: pathlib.Path) -> None:
        """Push a file or directory to the device.

        The target directory (if the source is a directory), or its
        parent (if the source if a file) will be created on the device,
        always.

        :raise DeviceCommandError: If the push command failed.
        """
        if source.is_dir():
            # `adb push` skips empty directories so we have to manually create the
            # target directory itself.
            self.adb("shell", "mkdir", "-p", str(target))
            # `adb push` expects the target to be the host directory not the target
            # directory itself. So we'll just copy the source directory content instead.
            sources = [str(child) for child in source.iterdir()]
            self.adb("push", *sources, str(target))
        else:
            self.adb("shell", "mkdir", "-p", str(target.parent))
            self.adb("push", str(source), str(target))

    def remove(self, target: pathlib.Path, *, recurse: bool = False) -> None:
        """Remove a file or directory from the device.

        :raise DeviceCommandError: If the remove command failed.
        """
        command = ["shell", "rm", "-f"]
        if recurse:
            command.append("-r")
        command.append(str(target))
        self.adb(*command)


def deploy():
    serials = []

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

    for serial in serials:
        _LOG.info("Deploy to device %s", serial)

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

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

            # Push the scenarios, their data, and install the apps
            prepare_dut(
                device,
                pathlib.Path("../scenarios"),
                pathlib.Path("../scenarios-data"),
                PREBUILTS_PATH,
            )

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

            configure_perms(device)

            configure_settings(device)
        except (HostCommandError, DeviceCommandError, uiautomator.JsonRPCError):
            _LOG.error("Failed to execute deployment", exc_info=True)
        finally:
            try:
                # Leave the device alone now
                device.force_awake(always=False)
            except DeviceCommandError:
                _LOG.warning("Failed to tear down device", exc_info=True)


if __name__ == "__main__":
    logging.basicConfig(
        format="%(asctime)-15s:%(levelname)s:%(message)s", level=logging.INFO
    )
    deploy()