Quelle  android.py

# 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-34;google_apis;x86_64",
    "emulator_avd_name": "mozemulator-android34-x86_64"
}


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"]
    android_device.AVD_DICT["x86_64"] = android_device.AvdInfo(
        "Android x86_64",
        "mozemulator-android34-x86_64",
        [
            "-skip-adb-auth",
            "-verbose",
            "-show-kernel",
            "-ranchu",
            "-selinux",
            "permissive",
            "-memory",
            "4096",
            "-cores",
            "4",
            "-prop",
            "ro.test_harness=true",
            "-no-snapstorage",
            "-no-snapshot",
            "-skin",
            "800x1280"
        ],
        True,
    )


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 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:
                for chunk in resp.iter_content(2**16):
                    f.write(chunk)
        if not os.path.exists(temp_path):
            raise ValueError(f"Failed to download {url}, output path doesn't exist")
        # 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": platform.uname().machine,
        "HOST_CPU_ARCH": platform.uname().machine
    }
    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;37.0.0",
                        "platforms;android-37.0",
                        "emulator"]

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

            install_avd(logger, paths, prompt=prompt)

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