Source code
Revision control
Copy as Markdown
Other Tools
#!/usr/bin/env python
# ***** BEGIN LICENSE BLOCK *****
# 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,
# ***** END LICENSE BLOCK *****
import datetime
import functools
import glob
import os
import posixpath
import re
import signal
import subprocess
import tempfile
import time
from threading import Timer
import six
from mozharness.base.script import PostScriptAction, PreScriptAction
from mozharness.mozilla.automation import EXIT_STATUS_DICT, TBPL_RETRY
def ensure_dir(dir):
"""Ensures the given directory exists"""
if dir and not os.path.exists(dir):
try:
os.makedirs(dir)
except OSError as error:
if error.errno != errno.EEXIST:
raise
class AndroidMixin(object):
"""
Mixin class used by Android test scripts.
"""
def __init__(self, **kwargs):
self._adb_path = None
self._device = None
self.app_name = None
self.device_name = os.environ.get("DEVICE_NAME", None)
self.device_serial = os.environ.get("DEVICE_SERIAL", None)
self.device_ip = os.environ.get("DEVICE_IP", None)
self.logcat_proc = None
self.logcat_file = None
self.use_gles3 = False
self.use_root = True
self.xre_path = None
super(AndroidMixin, self).__init__(**kwargs)
@property
def adb_path(self):
"""Get the path to the adb executable."""
self.activate_virtualenv()
if not self._adb_path:
self._adb_path = self.query_exe("adb")
return self._adb_path
@property
def device(self):
if not self._device:
# We must access the adb_path property to activate the
# virtualenv before importing mozdevice in order to
# import the mozdevice installed into the virtualenv and
# not any system-wide installation of mozdevice.
adb = self.adb_path
import mozdevice
self._device = mozdevice.ADBDeviceFactory(
adb=adb, device=self.device_serial, use_root=self.use_root
)
return self._device
@property
def is_android(self):
c = self.config
installer_url = c.get("installer_url", None)
return (
self.device_serial is not None
or self.is_emulator
or (
installer_url is not None
and (installer_url.endswith(".apk") or installer_url.endswith(".aab"))
)
)
@property
def is_emulator(self):
c = self.config
return True if c.get("emulator_avd_name") else False
def _get_repo_url(self, path):
"""
Return a url for a file (typically a tooltool manifest) in this hg repo
and using this revision (or mozilla-central/default if repo/rev cannot
be determined).
:param path specifies the directory path to the file of interest.
"""
if "GECKO_HEAD_REPOSITORY" in os.environ and "GECKO_HEAD_REV" in os.environ:
# probably taskcluster
repo = os.environ["GECKO_HEAD_REPOSITORY"]
revision = os.environ["GECKO_HEAD_REV"]
else:
# something unexpected!
revision = "default"
self.warning(
"Unable to find repo/revision for manifest; "
"using mozilla-central/default"
)
url = "%s/raw-file/%s/%s" % (repo, revision, path)
return url
def _launch_emulator(self):
env = self.query_env()
# Write a default ddms.cfg to avoid unwanted prompts
avd_home_dir = self.abs_dirs["abs_avds_dir"]
DDMS_FILE = os.path.join(avd_home_dir, "ddms.cfg")
with open(DDMS_FILE, "w") as f:
f.write("pingOptIn=false\npingId=0\n")
self.info("wrote dummy %s" % DDMS_FILE)
# Delete emulator auth file, so it doesn't prompt
AUTH_FILE = os.path.join(
os.path.expanduser("~"), ".emulator_console_auth_token"
)
if os.path.exists(AUTH_FILE):
try:
os.remove(AUTH_FILE)
self.info("deleted %s" % AUTH_FILE)
except Exception:
self.warning("failed to remove %s" % AUTH_FILE)
env["ANDROID_EMULATOR_HOME"] = avd_home_dir
avd_path = os.path.join(avd_home_dir, "avd")
if os.path.exists(avd_path):
env["ANDROID_AVD_HOME"] = avd_path
self.info("Found avds at %s" % avd_path)
else:
self.warning("AVDs missing? Not found at %s" % avd_path)
if "deprecated_sdk_path" in self.config:
sdk_path = os.path.abspath(os.path.join(avd_home_dir, ".."))
else:
sdk_path = self.abs_dirs["abs_sdk_dir"]
if os.path.exists(sdk_path):
env["ANDROID_SDK_HOME"] = sdk_path
env["ANDROID_SDK_ROOT"] = sdk_path
self.info("Found sdk at %s" % sdk_path)
else:
self.warning("Android sdk missing? Not found at %s" % sdk_path)
avd_config_path = os.path.join(
avd_path, "%s.ini" % self.config["emulator_avd_name"]
)
avd_folder = os.path.join(avd_path, "%s.avd" % self.config["emulator_avd_name"])
if os.path.isfile(avd_config_path):
# The ini file points to the absolute path to the emulator folder,
# which might be different, so we need to update it.
old_config = ""
with open(avd_config_path, "r") as config_file:
old_config = config_file.readlines()
self.info("Old Config: %s" % old_config)
with open(avd_config_path, "w") as config_file:
for line in old_config:
if line.startswith("path="):
config_file.write("path=%s\n" % avd_folder)
self.info("Updating path from: %s" % line)
else:
config_file.write("%s\n" % line)
else:
self.warning("Could not find config path at %s" % avd_config_path)
# enable EGL 3.0 in advancedFeatures.ini
AF_FILE = os.path.join(avd_home_dir, "advancedFeatures.ini")
with open(AF_FILE, "w") as f:
if self.use_gles3:
f.write("GLESDynamicVersion=on\n")
else:
f.write("GLESDynamicVersion=off\n")
# extra diagnostics for kvm acceleration
emu = self.config.get("emulator_process_name")
if os.path.exists("/dev/kvm") and emu and "x86" in emu:
try:
self.run_command(["ls", "-l", "/dev/kvm"])
self.run_command(["kvm-ok"])
self.run_command(["emulator", "-accel-check"], env=env)
except Exception as e:
self.warning("Extra kvm diagnostics failed: %s" % str(e))
self.info("emulator env: %s" % str(env))
command = ["emulator", "-avd", self.config["emulator_avd_name"]]
if "emulator_extra_args" in self.config:
command += self.config["emulator_extra_args"]
dir = self.query_abs_dirs()["abs_blob_upload_dir"]
tmp_file = tempfile.NamedTemporaryFile(
mode="w", prefix="emulator-", suffix=".log", dir=dir, delete=False
)
self.info("Launching the emulator with: %s" % " ".join(command))
self.info("Writing log to %s" % tmp_file.name)
proc = subprocess.Popen(
command, stdout=tmp_file, stderr=tmp_file, env=env, bufsize=0
)
return proc
def _verify_emulator(self):
boot_ok = self._retry(
30,
10,
self.is_boot_completed,
"Verify Android boot completed",
max_time=330,
)
if not boot_ok:
self.warning("Unable to verify Android boot completion")
return False
return True
def _verify_emulator_and_restart_on_fail(self):
emulator_ok = self._verify_emulator()
if not emulator_ok:
self.device_screenshot("screenshot-emulator-start")
self.kill_processes(self.config["emulator_process_name"])
subprocess.check_call(["ps", "-ef"])
# remove emulator tmp files
for dir in glob.glob("/tmp/android-*"):
self.rmtree(dir)
time.sleep(5)
self.emulator_proc = self._launch_emulator()
return emulator_ok
def _retry(self, max_attempts, interval, func, description, max_time=0):
"""
Execute func until it returns True, up to max_attempts times, waiting for
interval seconds between each attempt. description is logged on each attempt.
If max_time is specified, no further attempts will be made once max_time
seconds have elapsed; this provides some protection for the case where
the run-time for func is long or highly variable.
"""
status = False
attempts = 0
if max_time > 0:
end_time = datetime.datetime.now() + datetime.timedelta(seconds=max_time)
else:
end_time = None
while attempts < max_attempts and not status:
if (end_time is not None) and (datetime.datetime.now() > end_time):
self.info(
"Maximum retry run-time of %d seconds exceeded; "
"remaining attempts abandoned" % max_time
)
break
if attempts != 0:
self.info("Sleeping %d seconds" % interval)
time.sleep(interval)
attempts += 1
self.info(
">> %s: Attempt #%d of %d" % (description, attempts, max_attempts)
)
status = func()
return status
def dump_perf_info(self):
"""
Dump some host and android device performance-related information
to an artifact file, to help understand task performance.
"""
dir = self.query_abs_dirs()["abs_blob_upload_dir"]
perf_path = os.path.join(dir, "android-performance.log")
with open(perf_path, "w") as f:
f.write("\n\nHost cpufreq/scaling_governor:\n")
cpus = glob.glob("/sys/devices/system/cpu/cpu*/cpufreq/scaling_governor")
for cpu in cpus:
out = subprocess.check_output(["cat", cpu], universal_newlines=True)
f.write("%s: %s" % (cpu, out))
f.write("\n\nHost /proc/cpuinfo:\n")
out = subprocess.check_output(
["cat", "/proc/cpuinfo"], universal_newlines=True
)
f.write(out)
f.write("\n\nHost /proc/meminfo:\n")
out = subprocess.check_output(
["cat", "/proc/meminfo"], universal_newlines=True
)
f.write(out)
f.write("\n\nHost process list:\n")
out = subprocess.check_output(["ps", "-ef"], universal_newlines=True)
f.write(out)
f.write("\n\nDevice /proc/cpuinfo:\n")
cmd = "cat /proc/cpuinfo"
out = self.shell_output(cmd)
f.write(out)
cpuinfo = out
f.write("\n\nDevice /proc/meminfo:\n")
cmd = "cat /proc/meminfo"
out = self.shell_output(cmd)
f.write(out)
f.write("\n\nDevice process list:\n")
cmd = "ps"
out = self.shell_output(cmd)
f.write(out)
# Search android cpuinfo for "BogoMIPS"; if found and < (minimum), retry
# this task, in hopes of getting a higher-powered environment.
# (Carry on silently if BogoMIPS is not found -- this may vary by
# Android implementation -- no big deal.)
# low bogomips can be a good predictor of that condition.
bogomips_minimum = int(self.config.get("bogomips_minimum") or 0)
for line in cpuinfo.split("\n"):
m = re.match(r"BogoMIPS.*: (\d*)", line, re.IGNORECASE)
if m:
bogomips = int(m.group(1))
if bogomips_minimum > 0 and bogomips < bogomips_minimum:
self.fatal(
"INFRA-ERROR: insufficient Android bogomips (%d < %d)"
% (bogomips, bogomips_minimum),
EXIT_STATUS_DICT[TBPL_RETRY],
)
self.info("Found Android bogomips: %d" % bogomips)
break
def logcat_path(self):
logcat_filename = "logcat-%s.log" % self.device_serial
return os.path.join(
self.query_abs_dirs()["abs_blob_upload_dir"], logcat_filename
)
def logcat_start(self):
"""
Start recording logcat. Writes logcat to the upload directory.
"""
# Start logcat for the device. The adb process runs until the
# corresponding device is stopped. Output is written directly to
# the blobber upload directory so that it is uploaded automatically
# at the end of the job.
self.logcat_file = open(self.logcat_path(), "w")
logcat_cmd = [
self.adb_path,
"-s",
self.device_serial,
"logcat",
"-v",
"threadtime",
"Trace:S",
"StrictMode:S",
"ExchangeService:S",
]
self.info(" ".join(logcat_cmd))
self.logcat_proc = subprocess.Popen(
logcat_cmd, stdout=self.logcat_file, stdin=subprocess.PIPE
)
def logcat_stop(self):
"""
Stop logcat process started by logcat_start.
"""
if self.logcat_proc:
self.info("Killing logcat pid %d." % self.logcat_proc.pid)
self.logcat_proc.kill()
self.logcat_file.close()
def _install_android_app_retry(self, app_path, replace):
import mozdevice
try:
if app_path.endswith(".aab"):
self.device.install_app_bundle(
self.query_abs_dirs()["abs_bundletool_path"], app_path, timeout=120
)
self.device.run_as_package = self.query_package_name()
else:
self.device.run_as_package = self.device.install_app(
app_path, replace=replace, timeout=120
)
return True
except (
mozdevice.ADBError,
mozdevice.ADBProcessError,
mozdevice.ADBTimeoutError,
) as e:
self.info(
"Failed to install %s on %s: %s %s"
% (app_path, self.device_name, type(e).__name__, e)
)
return False
def install_android_app(self, app_path, replace=False):
"""
Install the specified app.
"""
app_installed = self._retry(
5,
10,
functools.partial(self._install_android_app_retry, app_path, replace),
"Install app",
)
if not app_installed:
self.fatal(
"INFRA-ERROR: Failed to install %s" % os.path.basename(app_path),
EXIT_STATUS_DICT[TBPL_RETRY],
)
def uninstall_android_app(self):
"""
Uninstall the app associated with the configured app, if it is
installed.
"""
import mozdevice
try:
package_name = self.query_package_name()
self.device.uninstall_app(package_name)
except (
mozdevice.ADBError,
mozdevice.ADBProcessError,
mozdevice.ADBTimeoutError,
) as e:
self.info(
"Failed to uninstall %s from %s: %s %s"
% (package_name, self.device_name, type(e).__name__, e)
)
self.fatal(
"INFRA-ERROR: %s Failed to uninstall %s"
% (type(e).__name__, package_name),
EXIT_STATUS_DICT[TBPL_RETRY],
)
def is_boot_completed(self):
import mozdevice
try:
return self.device.is_device_ready(timeout=30)
except (ValueError, mozdevice.ADBError, mozdevice.ADBTimeoutError):
pass
return False
def shell_output(self, cmd, enable_run_as=False):
import mozdevice
try:
return self.device.shell_output(
cmd, timeout=30, enable_run_as=enable_run_as
)
except mozdevice.ADBTimeoutError as e:
self.info(
"Failed to run shell command %s from %s: %s %s"
% (cmd, self.device_name, type(e).__name__, e)
)
self.fatal(
"INFRA-ERROR: %s Failed to run shell command %s"
% (type(e).__name__, cmd),
EXIT_STATUS_DICT[TBPL_RETRY],
)
def device_screenshot(self, prefix):
"""
On emulator, save a screenshot of the entire screen to the upload directory;
otherwise, save a screenshot of the device to the upload directory.
:param prefix specifies a filename prefix for the screenshot
"""
from mozscreenshot import dump_device_screen, dump_screen
reset_dir = False
if not os.environ.get("MOZ_UPLOAD_DIR", None):
dirs = self.query_abs_dirs()
os.environ["MOZ_UPLOAD_DIR"] = dirs["abs_blob_upload_dir"]
reset_dir = True
if self.is_emulator:
if self.xre_path:
dump_screen(self.xre_path, self, prefix=prefix)
else:
self.info("Not saving screenshot: no XRE configured")
else:
dump_device_screen(self.device, self, prefix=prefix)
if reset_dir:
del os.environ["MOZ_UPLOAD_DIR"]
def query_package_name(self):
if self.app_name is None:
# For convenience, assume geckoview.test/geckoview_example when install
# target looks like geckoview.
if "androidTest" in self.installer_path:
self.app_name = "org.mozilla.geckoview.test"
elif "test_runner" in self.installer_path:
self.app_name = "org.mozilla.geckoview.test_runner"
elif "geckoview" in self.installer_path:
self.app_name = "org.mozilla.geckoview_example"
if self.app_name is None:
# Find appname from package-name.txt - assumes download-and-extract
# has completed successfully.
# The app/package name will typically be org.mozilla.fennec,
# but org.mozilla.firefox for release builds, and there may be
# other variations. 'aapt dump badging <apk>' could be used as an
# alternative to package-name.txt, but introduces a dependency
# on aapt, found currently in the Android SDK build-tools component.
app_dir = self.abs_dirs["abs_work_dir"]
self.app_path = os.path.join(app_dir, self.installer_path)
unzip = self.query_exe("unzip")
package_path = os.path.join(app_dir, "package-name.txt")
unzip_cmd = [unzip, "-q", "-o", self.app_path]
self.run_command(unzip_cmd, cwd=app_dir, halt_on_failure=True)
self.app_name = str(
self.read_from_file(package_path, verbose=True)
).rstrip()
return self.app_name
def kill_processes(self, process_name):
self.info("Killing every process called %s" % process_name)
process_name = six.ensure_binary(process_name)
out = subprocess.check_output(["ps", "-A"])
for line in out.splitlines():
if process_name in line:
pid = int(line.split(None, 1)[0])
self.info("Killing pid %d." % pid)
os.kill(pid, signal.SIGKILL)
def delete_ANRs(self):
remote_dir = self.device.stack_trace_dir
try:
if not self.device.is_dir(remote_dir):
self.device.mkdir(remote_dir)
self.info("%s created" % remote_dir)
return
self.device.chmod(remote_dir, recursive=True)
for trace_file in self.device.ls(remote_dir, recursive=True):
trace_path = posixpath.join(remote_dir, trace_file)
if self.device.is_file(trace_path):
self.device.rm(trace_path)
self.info("%s deleted" % trace_path)
except Exception as e:
self.info(
"failed to delete %s: %s %s" % (remote_dir, type(e).__name__, str(e))
)
def check_for_ANRs(self):
"""
Copy ANR (stack trace) files from device to upload directory.
"""
dirs = self.query_abs_dirs()
remote_dir = self.device.stack_trace_dir
try:
if not self.device.is_dir(remote_dir):
self.info("%s not found; ANR check skipped" % remote_dir)
return
self.device.chmod(remote_dir, recursive=True)
self.device.pull(remote_dir, dirs["abs_blob_upload_dir"])
self.delete_ANRs()
except Exception as e:
self.info(
"failed to pull %s: %s %s" % (remote_dir, type(e).__name__, str(e))
)
def delete_tombstones(self):
remote_dir = "/data/tombstones"
try:
if not self.device.is_dir(remote_dir):
self.device.mkdir(remote_dir)
self.info("%s created" % remote_dir)
return
self.device.chmod(remote_dir, recursive=True)
for trace_file in self.device.ls(remote_dir, recursive=True):
trace_path = posixpath.join(remote_dir, trace_file)
if self.device.is_file(trace_path):
self.device.rm(trace_path)
self.info("%s deleted" % trace_path)
except Exception as e:
self.info(
"failed to delete %s: %s %s" % (remote_dir, type(e).__name__, str(e))
)
def check_for_tombstones(self):
"""
Copy tombstone files from device to upload directory.
"""
dirs = self.query_abs_dirs()
remote_dir = "/data/tombstones"
try:
if not self.device.is_dir(remote_dir):
self.info("%s not found; tombstone check skipped" % remote_dir)
return
self.device.chmod(remote_dir, recursive=True)
self.device.pull(remote_dir, dirs["abs_blob_upload_dir"])
self.delete_tombstones()
except Exception as e:
self.info(
"failed to pull %s: %s %s" % (remote_dir, type(e).__name__, str(e))
)
# Script actions
def start_emulator(self):
"""
Starts the emulator
"""
if not self.is_emulator:
return
dirs = self.query_abs_dirs()
ensure_dir(dirs["abs_work_dir"])
ensure_dir(dirs["abs_blob_upload_dir"])
if not os.path.isfile(self.adb_path):
self.fatal("The adb binary '%s' is not a valid file!" % self.adb_path)
self.kill_processes("xpcshell")
self.emulator_proc = self._launch_emulator()
def verify_device(self):
"""
Check to see if the emulator can be contacted via adb.
If any communication attempt fails, kill the emulator, re-launch, and re-check.
"""
if not self.is_android:
return
if self.is_emulator:
max_restarts = 5
emulator_ok = self._retry(
max_restarts,
10,
self._verify_emulator_and_restart_on_fail,
"Check emulator",
)
if not emulator_ok:
self.fatal(
"INFRA-ERROR: Unable to start emulator after %d attempts"
% max_restarts,
EXIT_STATUS_DICT[TBPL_RETRY],
)
self.mkdir_p(self.query_abs_dirs()["abs_blob_upload_dir"])
self.dump_perf_info()
self.logcat_start()
self.delete_ANRs()
self.delete_tombstones()
self.info("verify_device complete")
@PreScriptAction("run-tests")
def timed_screenshots(self, action, success=None):
"""
If configured, start screenshot timers.
"""
if not self.is_android:
return
def take_screenshot(seconds):
self.device_screenshot("screenshot-%ss-" % str(seconds))
self.info("timed (%ss) screenshot complete" % str(seconds))
self.timers = []
for seconds in self.config.get("screenshot_times", []):
self.info("screenshot requested %s seconds from now" % str(seconds))
t = Timer(int(seconds), take_screenshot, [seconds])
t.start()
self.timers.append(t)
@PostScriptAction("run-tests")
def stop_device(self, action, success=None):
"""
Stop logcat and kill the emulator, if necessary.
"""
if not self.is_android:
return
for t in self.timers:
t.cancel()
if self.worst_status != TBPL_RETRY:
self.check_for_ANRs()
self.check_for_tombstones()
else:
self.info("ANR and tombstone checks skipped due to TBPL_RETRY")
self.logcat_stop()
if self.is_emulator:
self.kill_processes(self.config["emulator_process_name"])