Quelle  android.py   Sprache: Python

# mypy: allow-untyped-defs

import argparse
import os
import platform
import signal
import shutil
import subprocess
import threading

import requests
from .wpt import venv_dir

android_device = None

here = os.path.abspath(os.path.dirname(__file__))
wpt_root = os.path.abspath(os.path.join(here, os.pardir, os.pardir))

CMDLINE_TOOLS_VERSION_STRING = "12.0"
CMDLINE_TOOLS_VERSION = "11076708"

AVD_MANIFEST_X86_64 = {
    "emulator_package": "system-images;android-24;default;x86_64",
    "emulator_avd_name": "mozemulator-x86_64",
    "emulator_extra_args": [
        "-skip-adb-auth",
        "-verbose",
        "-show-kernel",
        "-ranchu",
        "-selinux", "permissive",
        "-memory", "3072",
        "-cores", "4",
        "-skin", "800x1280",
        "-gpu", "on",
        "-no-snapstorage",
        "-no-snapshot",
        "-no-window",
        "-no-accel",
        "-prop", "ro.test_harness=true"
    ],
    "emulator_extra_config": {
        "hw.keyboard": "yes",
        "hw.lcd.density": "320",
        "disk.dataPartition.size": "4000MB",
        "sdcard.size": "600M"
    }
}


def do_delayed_imports(paths):
    global android_device
    from mozrunner.devices import android_device

    android_device.TOOLTOOL_PATH = os.path.join(os.path.dirname(__file__),
                                                os.pardir,
                                                "third_party",
                                                "tooltool",
                                                "tooltool.py")
    android_device.EMULATOR_HOME_DIR = paths["emulator_home"]


def get_parser_install():
    parser = argparse.ArgumentParser()
    parser.add_argument("--path", dest="dest",
                        help="Root path to use for emulator tooling")
    parser.add_argument("--reinstall", action="store_true",
                        help="Force reinstall even if the emulator already exists")
    parser.add_argument("--prompt", action="store_true",
                        help="Enable confirmation prompts")
    parser.add_argument("--no-prompt", dest="prompt", action="store_false",
                        help="Skip confirmation prompts")
    return parser


def get_parser_start():
    parser = get_parser_install()
    parser.add_argument("--device-serial",
                        help="Device serial number for Android emulator, if not emulator-5554")
    return parser


def install_fixed_emulator_version(logger, paths):
    # Downgrade to a pinned emulator version
    # See https://developer.android.com/studio/emulator_archive for what we're doing here
    from xml.etree import ElementTree

    version = "32.1.15"
    urls = {"linux": "https://redirector.gvt1.com/edgedl/android/repository/emulator-linux_x64-10696886.zip"}

    os_name = platform.system().lower()
    if os_name not in urls:
        logger.error(f"Don't know how to install old emulator for {os_name}, using latest version")
        # For now try with the latest version if this fails
        return

    logger.info(f"Downgrading emulator to {version}")
    url = urls[os_name]

    emulator_path = os.path.join(paths["sdk"], "emulator")
    latest_emulator_path = os.path.join(paths["sdk"], "emulator_latest")
    if os.path.exists(latest_emulator_path):
        shutil.rmtree(latest_emulator_path)
    os.rename(emulator_path, latest_emulator_path)

    download_and_extract(url, paths["sdk"])
    package_path = os.path.join(emulator_path, "package.xml")
    shutil.copyfile(os.path.join(latest_emulator_path, "package.xml"),
                    package_path)

    with open(package_path) as f:
        tree = ElementTree.parse(f)
    node = tree.find("localPackage").find("revision")
    assert len(node) == 3
    parts = version.split(".")
    for version_part, node in zip(parts, node):
        node.text = version_part
    with open(package_path, "wb") as f:
        tree.write(f, encoding="utf8")


def get_paths(dest):
    os_name = platform.system().lower()

    if dest is None:
        # os.getcwd() doesn't include the venv path
        base_path = os.path.join(wpt_root, venv_dir(), "android")
    else:
        base_path = dest

    sdk_path = os.environ.get("ANDROID_SDK_HOME", os.path.join(base_path, f"android-sdk-{os_name}"))
    avd_path = os.environ.get("ANDROID_AVD_HOME", os.path.join(sdk_path, ".android", "avd"))
    return {
        "base": base_path,
        "sdk": sdk_path,
        "sdk_tools": os.path.join(sdk_path, "cmdline-tools", CMDLINE_TOOLS_VERSION_STRING),
        "avd": avd_path,
        "emulator_home": os.path.dirname(avd_path)
    }


def get_sdk_manager_path(paths):
    os_name = platform.system().lower()
    file_name = "sdkmanager"
    if os_name.startswith("win"):
        file_name += ".bat"
    return os.path.join(paths["sdk_tools"], "bin", file_name)


def get_avd_manager(paths):
    os_name = platform.system().lower()
    file_name = "avdmanager"
    if os_name.startswith("win"):
        file_name += ".bat"
    return os.path.join(paths["sdk_tools"], "bin", file_name)


def uninstall_sdk(paths):
    if os.path.exists(paths["sdk"]) and os.path.isdir(paths["sdk"]):
        shutil.rmtree(paths["sdk"])


def get_os_tag(logger):
    os_name = platform.system().lower()
    if os_name not in ["darwin", "linux", "windows"]:
        logger.critical("Unsupported platform %s" % os_name)
        raise NotImplementedError

    if os_name == "macosx":
        return "darwin"
    if os_name == "windows":
        return "win"
    return "linux"


def download_and_extract(url, path):
    if not os.path.exists(path):
        os.makedirs(path)
    temp_path = os.path.join(path, url.rsplit("/", 1)[1])
    try:
        with open(temp_path, "wb") as f:
            with requests.get(url, stream=True) as resp:
                shutil.copyfileobj(resp.raw, f)

        # Python's zipfile module doesn't seem to work here
        subprocess.check_call(["unzip", temp_path], cwd=path)
    finally:
        if os.path.exists(temp_path):
            os.unlink(temp_path)


def install_sdk(logger, paths):
    if os.path.isdir(paths["sdk_tools"]):
        logger.info("Using SDK installed at %s" % paths["sdk_tools"])
        return False

    if not os.path.exists(paths["sdk"]):
        os.makedirs(paths["sdk"])

    download_path = os.path.dirname(paths["sdk_tools"])

    url = f'https://dl.google.com/android/repository/commandlinetools-{get_os_tag(logger)}-{CMDLINE_TOOLS_VERSION}_latest.zip'
    logger.info("Getting SDK from %s" % url)

    download_and_extract(url, download_path)
    os.rename(os.path.join(download_path, "cmdline-tools"), paths["sdk_tools"])

    return True


def install_android_packages(logger, paths, packages, prompt=True):
    sdk_manager = get_sdk_manager_path(paths)
    if not os.path.exists(sdk_manager):
        raise OSError(f"Can't find sdkmanager at {sdk_manager}")

    # TODO: make this work non-internactively
    logger.info(f"Installing Android packages {' '.join(packages)}")
    cmd = [sdk_manager] + packages

    input_data = None if prompt else "\n".join(["y"] * 100).encode("UTF-8")
    subprocess.run(cmd, check=True, input=input_data)


def install_avd(logger, paths, prompt=True):
    avd_manager = get_avd_manager(paths)
    avd_manifest = AVD_MANIFEST_X86_64

    install_android_packages(logger, paths, [avd_manifest["emulator_package"]], prompt=prompt)

    cmd = [avd_manager,
           "--verbose",
           "create",
           "avd",
           "--force",
           "--name",
           avd_manifest["emulator_avd_name"],
           "--package",
           avd_manifest["emulator_package"]]
    input_data = None if prompt else b"no"
    subprocess.run(cmd, check=True, input=input_data)


def get_emulator(paths, device_serial=None):
    if android_device is None:
        do_delayed_imports(paths)

    substs = {"top_srcdir": wpt_root, "TARGET_CPU": "x86"}
    emulator = android_device.AndroidEmulator(substs=substs,
                                              device_serial=device_serial,
                                              verbose=True)
    emulator.emulator_path = os.path.join(paths["sdk"], "emulator", "emulator")
    return emulator


class Environ:
    def __init__(self, **kwargs):
        self.environ = None
        self.set_environ = kwargs

    def __enter__(self):
        self.environ = os.environ.copy()
        for key, value in self.set_environ.items():
            if value is None:
                if key in os.environ:
                    del os.environ[key]
            else:
                os.environ[key] = value

    def __exit__(self, *args, **kwargs):
        os.environ = self.environ


def android_environment(paths):
    return Environ(ANDROID_EMULATOR_HOME=paths["emulator_home"],
                   ANDROID_AVD_HOME=paths["avd"],
                   ANDROID_SDK_ROOT=paths["sdk"],
                   ANDROID_SDK_HOME=paths["sdk"])


def install(logger, dest=None, reinstall=False, prompt=True):
    paths = get_paths(dest)

    with android_environment(paths):

        if reinstall:
            uninstall_sdk(paths)

        new_install = install_sdk(logger, paths)

        if new_install:
            packages = ["platform-tools",
                        "build-tools;35.0.0",
                        "platforms;android-35",
                        "emulator"]

            install_android_packages(logger, paths, packages, prompt=prompt)

            install_avd(logger, paths, prompt=prompt)

            install_fixed_emulator_version(logger, paths)

        emulator = get_emulator(paths)
    return emulator


def cancel_start(thread_id):
    def cancel_func():
        raise signal.pthread_kill(thread_id, signal.SIGINT)
    return cancel_func


def start(logger, dest=None, reinstall=False, prompt=True, device_serial=None):
    paths = get_paths(dest)

    with android_environment(paths):
        install(logger, dest=dest, reinstall=reinstall, prompt=prompt)

        emulator = get_emulator(paths, device_serial=device_serial)

        if not emulator.check_avd():
            logger.critical("Android AVD not found, please run |wpt install-android-emulator|")
            raise OSError

        emulator.start()
        timer = threading.Timer(300, cancel_start(threading.get_ident()))
        timer.start()
        for i in range(10):
            logger.info(f"Wait for emulator to start attempt {i + 1}/10")
            try:
                emulator.wait_for_start()
            except Exception:
                import traceback
                logger.warning(f"""emulator.wait_for_start() failed:
{traceback.format_exc()}""")
            else:
                break
        timer.cancel()
    return emulator


def run_install(venv, **kwargs):
    try:
        import logging
        logging.basicConfig()
        logger = logging.getLogger()

        install(logger, **kwargs)
    except Exception:
        import traceback
        traceback.print_exc()
        import pdb
        pdb.post_mortem()


def run_start(venv, **kwargs):
    try:
        import logging
        logging.basicConfig()
        logger = logging.getLogger()

        start(logger, **kwargs)
    except Exception:
        import traceback
        traceback.print_exc()
        import pdb
        pdb.post_mortem()