| #!/usr/bin/env python3 |
| |
| 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__) |
| |
| VWS_CREDENTIALS = {"user": "fairphonetesting@gmail.com", "password": "aish3echi:uwaiSh"} |
| |
| 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) |
| |
| |
| class InvalidIntentError(DeviceCommandError): |
| """An intent was rejected by a device.""" |
| |
| |
| 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() |
| |
| |
| # 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).""" |
| 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 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 >= 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) |
| |
| @contextlib.contextmanager |
| def launch(self, package_or_activity: str): |
| """Launch an app and eventually goes back to the home screen. |
| |
| There are two ways to start an application: |
| |
| 1. Start the main activity (advertised in the launcher), only |
| the package name is needed (e.g. `com.android.settings`); |
| 2. Start a specific activity, the qualified activity name is |
| needed (e.g. `com.android.settings/.Settings`). |
| |
| Example: |
| |
| >>> device = DeviceUnderTest(...) |
| >>> with device.launch("com.android.settings"): |
| ... device.ui(text="Bluetooth").exists |
| True |
| >>> device.ui(text="Bluetooth").exists |
| False |
| |
| :param package_or_activity: The package name or the qualified |
| activity name. |
| :raise DeviceCommandError: If the app manager command failed. |
| :raise InvalidIntentError: If the package or activity could not |
| be resolved. |
| """ |
| if "/" in package_or_activity: |
| [package, activity] = package_or_activity.split("/", maxsplit=1) |
| else: |
| package, activity = package_or_activity, "" |
| |
| if activity: |
| command = [ |
| "shell", |
| "am", |
| "start", |
| "-a", |
| "android.intent.action.MAIN", |
| package_or_activity, |
| ] |
| else: |
| # Use the monkey to avoid specifying the exact activity name. |
| command = [ |
| "shell", |
| "monkey", |
| "-p", |
| package, |
| "-c", |
| "android.intent.category.LAUNCHER", |
| "1", |
| ] |
| |
| try: |
| process = self.adb(*command) |
| if process.stderr: |
| 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 ViserProxyApp: |
| """A template for the viSer Proxy app.""" |
| |
| package: str |
| """The app package name.""" |
| apk_filename_template: str |
| """The string template of the APK filename, in the `str.format()` style (`{}`). |
| |
| The following place-holders can be used and will be replaced at runtime: |
| - `{sdk}`: The Android SDK number, e.g. `25` for Android 7.1. |
| - `{flavour}`: The Fairphone-specific Android build flavour, e.g. `gms` |
| (Fairphone OS) or `sibon` (Fairphone Open). |
| """ |
| |
| def resolve(self, *, sdk: int, flavour: str) -> AndroidApp: |
| """Resolve the app template into a Viser App.""" |
| if sdk >= 24: |
| sdk = 24 |
| else: |
| sdk = 19 |
| return AndroidApp( |
| self.package, self.apk_filename_template.format(sdk=sdk, flavour=flavour) |
| ) |
| |
| |
| @dataclasses.dataclass(frozen=True) |
| class ViserSuite: |
| """A SmartViser viSer app suite. |
| |
| Example: |
| |
| >>> device = DeviceUnderTest(...) |
| >>> suite = ViserSuite(...) |
| >>> suite.deploy(device) |
| |
| """ |
| |
| VISER_PROXY_APP_TEMPLATE = ViserProxyApp( |
| "com.lunarlabs.panda.proxy", |
| "com.lunarlabs.panda.proxy-latest-sdk{sdk}-{flavour}.apk", |
| ) |
| ANDROID_APPS = [ |
| AndroidApp("com.smartviser.demogame", "com.smartviser.demogame-latest.apk"), |
| AndroidApp("com.lunarlabs.panda", "com.lunarlabs.panda-latest.apk"), |
| ] |
| |
| prebuilts_path: pathlib.Path |
| scenarios_path: pathlib.Path |
| scenarios_data_path: pathlib.Path |
| target_path: pathlib.Path = pathlib.Path("/sdcard/Viser") |
| |
| @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" |
| |
| 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.VISER_PROXY_APP_TEMPLATE.resolve( |
| sdk=device.sdk, flavour=device.os_flavour |
| ) |
| ] + self.ANDROID_APPS |
| |
| def deploy(self, device: DeviceUnderTest) -> None: |
| """Deploy the suite on a device. |
| |
| Copy the test scenarios and their data, and install the |
| different apps composing 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) |
| |
| 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_path, self.target_scenarios_data_path) |
| |
| def install_suite(self, device: DeviceUnderTest) -> None: |
| """Install the suite apps on a 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) |
| |
| |
| def deploy(): |
| serials = [] |
| suite = ViserSuite( |
| pathlib.Path("../../vendor/smartviser/viser/prebuilts/apk"), |
| pathlib.Path("../scenarios"), |
| pathlib.Path("../scenarios-data"), |
| ) |
| |
| 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 |
| suite.deploy(device) |
| |
| with device.launch("com.lunarlabs.panda"): |
| 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() |