DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (409f3966645a)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223
# 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/.

from __future__ import absolute_import

from telnetlib import Telnet
import datetime
import os
import shutil
import subprocess
import tempfile
import time

from mozdevice import ADBHost
from mozprocess import ProcessHandler

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


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