deploy.py

#!/usr/bin/env python3

import abc
import contextlib
import dataclasses
import logging
import pathlib
import re
import subprocess
import sys
import time
from typing import Sequence

import uiautomator

_LOG = logging.getLogger(__name__)

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

ANDROID_6_SDK = 23
ANDROID_7_SDK = 24


class BaseError(Exception, abc.ABC):
    """An abstract error."""


class HostCommandError(BaseError):
    """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(BaseError):
    """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)


class InvalidIntentError(DeviceCommandError):
    """An intent was rejected by a device."""


class UnlicensedDeviceError(BaseError):
    """A device is missing a license."""

    def __init__(self, device, provider):
        self.device = device
        super().__init__(f"{device} is missing a license for {provider}")


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.
    """
    _LOG.info("Disable the Privacy Impact screen")
    with device.launch("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()


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

    serial: str
    """The device serial number (adb/fastboot)."""
    os_flavour: str
    """The Fairphone-specific OS flavour."""
    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"]
        # Cache the OS flavour
        self.os_flavour = "gms" if self.is_gms_device() else "sibon"

    def __str__(self) -> str:
        return f"{DeviceUnderTest.__name__}({self.serial})"

    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 uninstall_package(self, package_name: str) -> None:
        """Uninstall an app from its package name.

        :raise DeviceCommandError: If the uninstall command failed.
        """
        self.adb("uninstall", package_name)

    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 >= ANDROID_6_SDK:
            # 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)

    @contextlib.contextmanager
    def launch(self, activity: str):
        """Launch an app and eventually goes back to the home screen.

        Example:

        >>> device = DeviceUnderTest(...)
        >>> with device.launch("com.android.settings/.Settings"):
        ...     device.ui(text="Bluetooth").exists
        True
        >>> device.ui(text="Bluetooth").exists
        False

        :param activity: The qualified activity name.
        :raise DeviceCommandError: If the app manager command failed.
        :raise InvalidIntentError: If the activity could not be
            resolved.
        """
        command = ["shell", "am", "start", "-a", "android.intent.action.MAIN", activity]

        try:
            process = self.adb(*command)
            # Ignore warnings as they should not impede the app launch.  For instance,
            # Android is verbose about apps not started *again*, because they are e.g.
            # already active in the background or in the foreground, and that's fine.
            if process.stderr and not process.stderr.startswith(
                "Warning: Activity not started"
            ):
                raise InvalidIntentError(self.serial, " ".join(command), process.stderr)
            yield
        finally:
            self.ui.press.home()


@dataclasses.dataclass(frozen=True)
class AndroidApp:
    """An installable Android app."""

    package: str
    """The app package name."""
    apk: str
    """The app package (APK) filename."""

    def __str__(self) -> str:
        return self.package


@dataclasses.dataclass(frozen=True)
class Credentials:
    """Credentials to access a service."""

    username: str
    password: str


@dataclasses.dataclass(frozen=True)
class ViserSuite:
    """A SmartViser viSer app suite.

    Example:

    >>> device = DeviceUnderTest(...)
    >>> suite = ViserSuite(...)
    >>> suite.deploy(device)

    """

    ANDROID_APPS = [
        AndroidApp("com.smartviser.demogame", "com.smartviser.demogame-latest.apk"),
        AndroidApp("com.lunarlabs.panda", "com.lunarlabs.panda-latest.apk"),
    ]

    credentials: Credentials
    prebuilts_path: pathlib.Path
    scenarios_path: pathlib.Path
    scenarios_data_path: pathlib.Path
    settings_file: pathlib.Path
    target_path: pathlib.Path = pathlib.Path("/sdcard/Viser")

    def __str__(self) -> str:
        return "SmartViser viSer app suite"

    @property
    def target_scenarios_path(self) -> pathlib.Path:
        return self.target_path

    @property
    def target_scenarios_data_path(self) -> pathlib.Path:
        return self.target_path / "Data"

    @property
    def target_settings_file(self) -> pathlib.Path:
        return self.target_path / "Settings" / self.settings_file.name

    def resolve_apps(self, device: DeviceUnderTest) -> Sequence[AndroidApp]:
        """Resolve the apps based on the target device properties.

        :param device: The device to target.
        :returns: The sequence of apps suitable for the target device.
            The sequence is ordered to satisfy the dependency graph
            (i.e. required apps come first in the sequence).
        """
        return self.ANDROID_APPS

    def deploy(self, device: DeviceUnderTest) -> None:
        """Deploy the suite on a device.

        Copy the test scenarios and their data, install the
        different apps composing the suite, and configure the suite.

        The previous configuration and data (i.e. reports) tied to the
        app suite, if any, is deleted before hand.

        :param device: The device to deploy the suite to.
        """
        self.cleanup_previous_deployment(device)
        self.copy_scenarios(device)
        self.install_suite(device)
        self.configure_suite(device)

    def cleanup_previous_deployment(self, device: DeviceUnderTest) -> None:
        """Clean-up a previous deployment of the suite on a device.

        :param device: The device to clean-up.
        """
        # Uninstall the apps in the reverse order to cater for dependencies.
        for app in reversed(self.resolve_apps(device)):
            _LOG.info("Uninstall %s", app)
            device.uninstall_package(app.package)

        _LOG.info("Delete data from previous deployment")
        device.remove(self.target_path, recurse=True)

    def copy_scenarios(self, device: DeviceUnderTest) -> None:
        """Copy the suite scenarios and their data on a device.

        :param device: The device to copy the scenarios to.
        """
        _LOG.info(
            "Copy scenarios: %s → %s", self.scenarios_path, self.target_scenarios_path
        )
        device.push(self.scenarios_path, self.target_scenarios_path)

        _LOG.info(
            "Copy scenarios data: %s → %s",
            self.scenarios_data_path,
            self.target_scenarios_data_path,
        )
        device.push(self.scenarios_data_path, self.target_scenarios_data_path)

    def install_suite(self, device: DeviceUnderTest) -> None:
        """Install the suite apps on a device.

        Input the credentials in the viSer app to authenticate the
        device.

        :param device: The device to install the suite apps on.
        """
        for app in self.resolve_apps(device):
            _LOG.info("Install %s", app)
            device.install(self.prebuilts_path / app.apk)

        with device.launch("com.lunarlabs.panda/.ggb.Activity.MainActivity"):
            _LOG.info("Initiate first run of viSer")

            # Input the credentials
            device.ui(resourceId="android:id/content").child(text="Username").child(
                className="android.widget.EditText"
            ).set_text(self.credentials.username)
            device.ui(resourceId="android:id/content").child(text="Password").child(
                className="android.widget.EditText"
            ).set_text(self.credentials.password)

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

            # Wait for log-in to complete
            progress_bar = device.ui(resourceId="android:id/content").child(
                className="android.widget.ProgressBar"
            )
            progress_bar.wait.gone(timeout=10000)

            # Is the app licensed?
            license_failed_info = device.ui(resourceId="android:id/content").child(
                textContains="Licence verification failed"
            )
            if license_failed_info.exists:
                raise UnlicensedDeviceError(device, self)

            self._maybe_accept_viser_dialer(device)

    def configure_suite(self, device: DeviceUnderTest) -> None:
        """Configure the suite on a device.

        :param device: The device to configure the suite on.
        """
        _LOG.info(
            "Copy settings: %s → %s", self.settings_file, self.target_settings_file
        )
        device.push(self.settings_file, self.target_settings_file)

        with device.launch("com.lunarlabs.panda/.ggb.Activity.MainActivity"):
            device.ui(description="Open navigation drawer").click()
            device.ui(
                text="Settings", className="android.widget.CheckedTextView"
            ).click()
            device.ui(className="android.support.v7.widget.RecyclerView").child_by_text(
                "Settings management", className="android.widget.LinearLayout"
            ).click()
            device.ui(className="android.support.v7.widget.RecyclerView").child_by_text(
                "Load settings", className="android.widget.LinearLayout"
            ).click()
            device.ui(resourceId="android:id/list").child_by_text(
                self.settings_file.name, className="android.widget.TextView"
            ).click()
            device.ui(
                textMatches="(?i)Select", className="android.widget.Button"
            ).click()

            self._maybe_accept_viser_dialer(device)

    def _maybe_accept_viser_dialer(self, device: DeviceUnderTest) -> None:
        """Accept the built-in viSer dialer as the default dialer, if necessary.

        This method tries to accept the built-in viSer dialer from a
        currently displayed modal. Two types of modals are handled, the
        modal spawned by the viSer app itself, and the modal spawned by
        the system UI.

        This method is idempotent and will pass even if the modals are
        missing.

        :param device: The device to accept the modal on.
        """
        dialer_set = False

        # Optional modal to set the the built-in dialer as default
        set_dialer_modal = device.ui(resourceId="android:id/content").child(
            textContains="Do you want to use viSer dialer as default ?"
        )
        if set_dialer_modal.exists:
            device.ui(resourceId="android:id/content").child(
                textMatches="(?i)Yes", className="android.widget.Button"
            ).click()
            dialer_set = True

        # Optional system modal to really set the built-in dialer as default
        set_dialer_system_modal = device.ui(resourceId="android:id/content").child(
            text="Make viSer your default Phone app?"
        )
        if set_dialer_system_modal.exists:
            device.ui(resourceId="android:id/content").child(
                textMatches="(?i)Set default", className="android.widget.Button"
            ).click()
            dialer_set = True

        # Optional system modal to change the default dialer
        change_dialer_system_modal = device.ui(resourceId="android:id/content").child(
            text="Change default Dialer app?"
        )
        if change_dialer_system_modal.exists:
            device.ui(resourceId="android:id/content").child(
                textMatches="(?i)OK", className="android.widget.Button"
            ).click()
            dialer_set = True

        if dialer_set:
            _LOG.info("Set the built-in viSer dialer as default dialer")


def deploy():
    serials = []
    suite = ViserSuite(
        Credentials("fairphonetesting@gmail.com", "aish3echi:uwaiSh"),
        pathlib.Path("../../vendor/smartviser/viser/prebuilts/apk"),
        pathlib.Path("../scenarios"),
        pathlib.Path("../scenarios-data"),
        pathlib.Path("settings/fairphone-bot_settings"),
    )

    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 < ANDROID_6_SDK:
                disable_privacy_impact_popup(device)

            # Push the scenarios, their data, and install the apps
            suite.deploy(device)
        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()