Source code
Revision control
Copy as Markdown
Other Tools
# 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
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:
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:
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().__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
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().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().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().__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().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().start()