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
""" Drives an android device.
"""
import os
import posixpath
import tempfile
import contextlib
import time
import logging
import attr
from arsenic.services import Geckodriver, free_port, subprocess_based_service
from mozdevice import ADBDeviceFactory, ADBError
from condprof.util import write_yml_file, logger, DEFAULT_PREFS, BaseEnv
class AndroidDevice:
def __init__(
self,
app_name,
marionette_port=2828,
verbose=False,
remote_test_root="/sdcard/test_root/",
):
self.app_name = app_name
# XXX make that an option
if "fenix" in app_name:
self.activity = "org.mozilla.fenix.IntentReceiverActivity"
else:
self.activity = "org.mozilla.geckoview_example.GeckoViewActivity"
self.verbose = verbose
self.device = None
self.marionette_port = marionette_port
self.profile = None
self.remote_profile = None
self.log_file = None
self.remote_test_root = remote_test_root
self._adb_fh = None
def _set_adb_logger(self, log_file):
self.log_file = log_file
if self.log_file is None:
return
logger.info("Setting ADB log file to %s" % self.log_file)
adb_logger = logging.getLogger("adb")
adb_logger.setLevel(logging.DEBUG)
self._adb_fh = logging.FileHandler(self.log_file)
self._adb_fh.setLevel(logging.DEBUG)
adb_logger.addHandler(self._adb_fh)
def _unset_adb_logger(self):
if self._adb_fh is None:
return
logging.getLogger("adb").removeHandler(self._adb_fh)
self._adb_fh = None
def clear_logcat(self, timeout=None, buffers=[]):
if not self.device:
return
self.device.clear_logcat(timeout, buffers)
def get_logcat(self):
if not self.device:
return None
# we don't want to have ADBCommand dump the command
# in the debug stream so we reduce its verbosity here
# temporarely
old_verbose = self.device._verbose
self.device._verbose = False
try:
return self.device.get_logcat()
finally:
self.device._verbose = old_verbose
def prepare(self, profile, logfile):
self._set_adb_logger(logfile)
try:
# See android_emulator_pgo.py run_tests for more
# details on why test_root must be /sdcard/test_root
# for android pgo due to Android 4.3.
self.device = ADBDeviceFactory(
verbose=self.verbose, logger_name="adb", test_root=self.remote_test_root
)
except Exception:
logger.error("Cannot initialize device")
raise
device = self.device
self.profile = profile
# checking that the app is installed
if not device.is_app_installed(self.app_name):
raise Exception("%s is not installed" % self.app_name)
# debug flag
logger.info("Setting %s as the debug app on the phone" % self.app_name)
device.shell(
"am set-debug-app --persistent %s" % self.app_name,
stdout_callback=logger.info,
)
# creating the profile on the device
logger.info("Creating the profile on the device")
remote_profile = posixpath.join(self.remote_test_root, "profile")
logger.info("The profile on the phone will be at %s" % remote_profile)
device.rm(remote_profile, force=True, recursive=True)
logger.info("Pushing %s on the phone" % self.profile)
device.push(profile, remote_profile)
device.chmod(remote_profile, recursive=True)
self.profile = profile
self.remote_profile = remote_profile
# creating the yml file
yml_data = {
"args": ["-marionette", "-profile", self.remote_profile],
"prefs": DEFAULT_PREFS,
"env": {"LOG_VERBOSE": 1, "R_LOG_LEVEL": 6, "MOZ_LOG": ""},
}
yml_name = "%s-geckoview-config.yaml" % self.app_name
yml_on_host = posixpath.join(tempfile.mkdtemp(), yml_name)
write_yml_file(yml_on_host, yml_data)
tmp_on_device = posixpath.join("/data", "local", "tmp")
if not device.exists(tmp_on_device):
raise IOError("%s does not exists on the device" % tmp_on_device)
yml_on_device = posixpath.join(tmp_on_device, yml_name)
try:
device.rm(yml_on_device, force=True, recursive=True)
device.push(yml_on_host, yml_on_device)
device.chmod(yml_on_device, recursive=True)
except Exception:
logger.info("could not create the yaml file on device. Permission issue?")
raise
# command line 'extra' args not used with geckoview apps; instead we use
# an on-device config.yml file
intent = "android.intent.action.VIEW"
device.stop_application(self.app_name)
device.launch_application(
self.app_name, self.activity, intent, extras=None, url="about:blank"
)
if not device.process_exist(self.app_name):
raise Exception("Could not start %s" % self.app_name)
logger.info("Creating socket forwarding on port %d" % self.marionette_port)
device.forward(
local="tcp:%d" % self.marionette_port,
remote="tcp:%d" % self.marionette_port,
)
# we don't have a clean way for now to check that GV or Fenix
# is ready to handle our tests. So here we just wait 30s
logger.info("Sleeping for 30s")
time.sleep(30)
def stop_browser(self):
logger.info("Stopping %s" % self.app_name)
try:
self.device.stop_application(self.app_name)
except ADBError:
logger.info("Could not stop the application using force-stop")
time.sleep(5)
if self.device.process_exist(self.app_name):
logger.info("%s still running, trying SIGKILL" % self.app_name)
num_tries = 0
while self.device.process_exist(self.app_name) and num_tries < 5:
try:
self.device.pkill(self.app_name)
except ADBError:
pass
num_tries += 1
time.sleep(1)
logger.info("%s stopped" % self.app_name)
def collect_profile(self):
logger.info("Collecting profile from %s" % self.remote_profile)
self.device.pull(self.remote_profile, self.profile)
def close(self):
self._unset_adb_logger()
if self.device is None:
return
try:
self.device.remove_forwards("tcp:%d" % self.marionette_port)
except ADBError:
logger.warning("Could not remove forward port")
# XXX redundant, remove
@contextlib.contextmanager
def device(
app_name, marionette_port=2828, verbose=True, remote_test_root="/sdcard/test_root/"
):
device_ = AndroidDevice(
app_name, marionette_port, verbose, remote_test_root=remote_test_root
)
try:
yield device_
finally:
device_.close()
@attr.s
class AndroidGeckodriver(Geckodriver):
async def start(self):
port = free_port()
await self._check_version()
logger.info("Running Webdriver on port %d" % port)
logger.info("Running Marionette on port 2828")
pargs = [
self.binary,
"--log",
"trace",
"--port",
str(port),
"--marionette-port",
"2828",
]
logger.info("Connecting on Android device")
pargs.append("--connect-existing")
return await subprocess_based_service(
)
class AndroidEnv(BaseEnv):
@contextlib.contextmanager
def get_device(self, *args, **kw):
with device(self.firefox, *args, **kw) as d:
self.device = d
yield self.device
def get_target_platform(self):
app = self.firefox.split("org.mozilla.")[-1]
if self.device_name is None:
return app
return "%s-%s" % (self.device_name, app)
def dump_logs(self):
logger.info("Dumping Android logs")
try:
logcat = self.device.get_logcat()
if logcat:
# local path, not using posixpath
logfile = os.path.join(self.archive, "logcat.log")
logger.info("Writing logcat at %s" % logfile)
with open(logfile, "wb") as f:
for line in logcat:
f.write(line.encode("utf8", errors="replace") + b"\n")
else:
logger.info("logcat came back empty")
except Exception:
logger.error("Could not extract the logcat", exc_info=True)
@contextlib.contextmanager
def get_browser(self):
yield
def get_browser_args(self, headless, prefs=None):
# XXX merge with DesktopEnv.get_browser_args
options = ["--allow-downgrade"]
if headless:
options.append("-headless")
if prefs is None:
prefs = {}
return {"moz:firefoxOptions": {"args": options, "prefs": prefs}}
def prepare(self, logfile):
self.device.prepare(self.profile, logfile)
def get_browser_version(self):
return self.target_platform + "-XXXneedtograbversion"
def get_geckodriver(self, log_file):
return AndroidGeckodriver(binary=self.geckodriver, log_file=log_file)
def check_session(self, session):
async def fake_close(*args):
pass
session.close = fake_close
def collect_profile(self):
self.device.collect_profile()
def stop_browser(self):
self.device.stop_browser()