Quelle  emulator.py   Sprache: Python

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import datetime
import os
import shutil
import subprocess
import tempfile
import time

from mozdevice import ADBHost
from mozprocess import ProcessHandler

from ..errors import TimeoutException
from .base import Device
from .emulator_battery import EmulatorBattery
from .emulator_geo import EmulatorGeo
from .emulator_screen import EmulatorScreen

try:
    from telnetlib import Telnet
except ImportError:  # telnetlib was removed in Python 3.13
    from .telnetlib import Telnet


class ArchContext(object):
    def __init__(self, arch, context, binary=None, avd=None, extra_args=None):
        homedir = getattr(context, "homedir", "")
        kernel = os.path.join(homedir, "prebuilts", "qemu-kernel", "%s", "%s")
        sysdir = os.path.join(homedir, "out", "target", "product", "%s")
        self.extra_args = []
        self.binary = os.path.join(context.bindir or "", "emulator")
        if arch == "x86":
            self.binary = os.path.join(context.bindir or "", "emulator-x86")
            self.kernel = kernel % ("x86", "kernel-qemu")
            self.sysdir = sysdir % "generic_x86"
        elif avd:
            self.avd = avd
            self.extra_args = [
                "-show-kernel",
                "-debug",
                "init,console,gles,memcheck,adbserver,adbclient,adb,avd_config,socket",
            ]
        else:
            self.kernel = kernel % ("arm", "kernel-qemu-armv7")
            self.sysdir = sysdir % "generic"
            self.extra_args = ["-cpu", "cortex-a8"]

        if binary:
            self.binary = binary

        if extra_args:
            self.extra_args.extend(extra_args)


class SDCard(object):
    def __init__(self, emulator, size):
        self.emulator = emulator
        self.path = self.create_sdcard(size)

    def create_sdcard(self, sdcard_size):
        """
        Creates an sdcard partition in the emulator.

        :param sdcard_size: Size of partition to create, e.g '10MB'.
        """
        mksdcard = self.emulator.app_ctx.which("mksdcard")
        path = tempfile.mktemp(prefix="sdcard", dir=self.emulator.tmpdir)
        sdargs = [mksdcard, "-l", "mySdCard", sdcard_size, path]
        sd = subprocess.Popen(sdargs, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
        retcode = sd.wait()
        if retcode:
            raise Exception(
                "unable to create sdcard: exit code %d: %s"
                % (retcode, sd.stdout.read())
            )
        return path


class BaseEmulator(Device):
    port = None
    proc = None
    telnet = None

    def __init__(self, app_ctx, **kwargs):
        self.arch = ArchContext(
            kwargs.pop("arch", "arm"),
            app_ctx,
            binary=kwargs.pop("binary", None),
            avd=kwargs.pop("avd", None),
        )
        super(BaseEmulator, self).__init__(app_ctx, **kwargs)
        self.tmpdir = tempfile.mkdtemp()
        # These rely on telnet
        self.battery = EmulatorBattery(self)
        self.geo = EmulatorGeo(self)
        self.screen = EmulatorScreen(self)

    @property
    def args(self):
        """
        Arguments to pass into the emulator binary.
        """
        return [self.arch.binary]

    def start(self):
        """
        Starts a new emulator.
        """
        if self.proc:
            return

        original_devices = set(self._get_online_devices())

        # QEMU relies on atexit() to remove temporary files, which does not
        # work since mozprocess uses SIGKILL to kill the emulator process.
        # Use a customized temporary directory so we can clean it up.
        os.environ["ANDROID_TMP"] = self.tmpdir

        qemu_log = None
        qemu_proc_args = {}
        if self.logdir:
            # save output from qemu to logfile
            qemu_log = os.path.join(self.logdir, "qemu.log")
            if os.path.isfile(qemu_log):
                self._rotate_log(qemu_log)
            qemu_proc_args["logfile"] = qemu_log
        else:
            qemu_proc_args["processOutputLine"] = lambda line: None
        self.proc = ProcessHandler(self.args, **qemu_proc_args)
        self.proc.run()

        devices = set(self._get_online_devices())
        now = datetime.datetime.now()
        while (devices - original_devices) == set([]):
            time.sleep(1)
            # Sometimes it takes more than 60s to launch emulator, so we
            # increase timeout value to 180s. Please see bug 1143380.
            if datetime.datetime.now() - now > datetime.timedelta(seconds=180):
                raise TimeoutException("timed out waiting for emulator to start")
            devices = set(self._get_online_devices())
        devices = devices - original_devices
        self.serial = devices.pop()
        self.connect()

    def _get_online_devices(self):
        adbhost = ADBHost(adb=self.app_ctx.adb)
        return [
            d["device_serial"]
            for d in adbhost.devices()
            if d["state"] != "offline"
            if d["device_serial"].startswith("emulator")
        ]

    def connect(self):
        """
        Connects to a running device. If no serial was specified in the
        constructor, defaults to the first entry in `adb devices`.
        """
        if self.connected:
            return

        super(BaseEmulator, self).connect()
        self.port = int(self.serial[self.serial.rindex("-") + 1 :])

    def cleanup(self):
        """
        Cleans up and kills the emulator, if it was started by mozrunner.
        """
        super(BaseEmulator, self).cleanup()
        if self.proc:
            self.proc.kill()
            self.proc = None
            self.connected = False

        # Remove temporary files
        if os.path.isdir(self.tmpdir):
            shutil.rmtree(self.tmpdir)

    def _get_telnet_response(self, command=None):
        output = []
        assert self.telnet
        if command is not None:
            self.telnet.write("%s\n" % command)
        while True:
            line = self.telnet.read_until("\n")
            output.append(line.rstrip())
            if line.startswith("OK"):
                return output
            elif line.startswith("KO:"):
                raise Exception("bad telnet response: %s" % line)

    def _run_telnet(self, command):
        if not self.telnet:
            self.telnet = Telnet("localhost", self.port)
            self._get_telnet_response()
        return self._get_telnet_response(command)

    def __del__(self):
        if self.telnet:
            self.telnet.write("exit\n")
            self.telnet.read_all()


class EmulatorAVD(BaseEmulator):
    def __init__(self, app_ctx, binary, avd, port=5554, **kwargs):
        super(EmulatorAVD, self).__init__(app_ctx, binary=binary, avd=avd, **kwargs)
        self.port = port

    @property
    def args(self):
        """
        Arguments to pass into the emulator binary.
        """
        qemu_args = super(EmulatorAVD, self).args
        qemu_args.extend(["-avd", self.arch.avd, "-port", str(self.port)])
        qemu_args.extend(self.arch.extra_args)
        return qemu_args

    def start(self):
        if self.proc:
            return

        env = os.environ
        env["ANDROID_AVD_HOME"] = self.app_ctx.avd_home

        super(EmulatorAVD, self).start()