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
"""
Set up a browser environment before running a test.
"""
import json
import os
import shutil
import tempfile
import mozfile
import mozinfo
import mozrunner
import six
from mozlog import get_proxy_logger
from mozprofile.profile import Profile
from talos import heavy, utils
from talos.gecko_profile import GeckoProfile
from talos.utils import TalosError, run_in_debug_mode
here = os.path.abspath(os.path.dirname(__file__))
LOG = get_proxy_logger()
class FFSetup(object):
"""
Initialize the browser environment before running a test.
This prepares:
- the environment vars for running the test in the browser,
available via the instance member *env*.
- the profile used to run the test, available via the
instance member *profile_dir*.
- Gecko profiling, available via the instance member *gecko_profile*
of type :class:`GeckoProfile` or None if not used.
Note that the browser will be run once with the profile, to ensure
this is basically working and negate any performance noise with the
real test run (installing the profile the first time takes time).
This class should be used as a context manager::
with FFSetup(browser_config, test_config) as setup:
# setup.env is initialized, and setup.profile_dir created
pass
# here the profile is removed
"""
def __init__(self, browser_config, test_config):
self.browser_config, self.test_config = browser_config, test_config
self._tmp_dir = tempfile.mkdtemp()
self.env = None
# The profile dir must be named 'profile' because of xperf analysis
# (in etlparser.py). TODO fix that ?
self.profile_dir = os.path.join(self._tmp_dir, "profile")
self.gecko_profile = None
self.debug_mode = run_in_debug_mode(browser_config)
@property
def profile_data_dir(self):
if "MOZ_DEVELOPER_REPO_DIR" in os.environ:
return os.path.join(
os.environ["MOZ_DEVELOPER_REPO_DIR"], "testing", "profiles"
)
return os.path.join(here, "profile_data")
def _init_env(self):
self.env = dict(os.environ)
for k, v in six.iteritems(self.browser_config["env"]):
self.env[k] = str(v)
self.env["MOZ_CRASHREPORTER_NO_REPORT"] = "1"
if self.browser_config["symbols_path"]:
self.env["MOZ_CRASHREPORTER"] = "1"
else:
self.env["MOZ_CRASHREPORTER_DISABLE"] = "1"
self.env["MOZ_DISABLE_NONLOCAL_CONNECTIONS"] = "1"
self.env["LD_LIBRARY_PATH"] = os.path.dirname(
self.browser_config["browser_path"]
)
def _init_profile(self):
extensions = self.browser_config["extensions"][:]
if self.test_config.get("extensions"):
extensions.extend(self.test_config["extensions"])
# downloading a profile instead of using the empty one
if self.test_config["profile"] is not None:
path = heavy.download_profile(self.test_config["profile"])
self.test_config["profile_path"] = path
profile_path = os.path.normpath(self.test_config["profile_path"])
LOG.info("Cloning profile located at %s" % profile_path)
def _feedback(directory, content):
# Called by shutil.copytree on each visited directory.
# Used here to display info.
#
# Returns the items that should be ignored by
# shutil.copytree when copying the tree, so always returns
# an empty list.
sub = directory.split(profile_path)[-1].lstrip("/")
if sub:
LOG.info("=> %s" % sub)
return []
profile = Profile.clone(
profile_path, self.profile_dir, ignore=_feedback, restore=False
)
# build pref interpolation context
webserver = self.browser_config["webserver"]
if "://" not in webserver:
webserver = "http://" + webserver
interpolation = {
"webserver": webserver,
}
# merge base profiles
with open(os.path.join(self.profile_data_dir, "profiles.json"), "r") as fh:
base_profiles = json.load(fh)["talos"]
for name in base_profiles:
path = os.path.join(self.profile_data_dir, name)
LOG.info("Merging profile: {}".format(path))
profile.merge(path, interpolation=interpolation)
# set test preferences
preferences = self.browser_config.get("preferences", {}).copy()
if self.test_config.get("preferences"):
test_prefs = dict(
[
(i, utils.parse_pref(j))
for i, j in self.test_config["preferences"].items()
]
)
preferences.update(test_prefs)
for name, value in preferences.items():
if type(value) is str:
value = utils.interpolate(value, **interpolation)
preferences[name] = value
profile.set_preferences(preferences)
# installing addons
LOG.info("Installing Add-ons:")
LOG.info(extensions)
profile.addons.install(extensions)
# installing webextensions
webextensions_to_install = []
webextensions_folder = self.test_config.get("webextensions_folder", None)
if isinstance(webextensions_folder, six.string_types):
folder = utils.interpolate(webextensions_folder)
for file in os.listdir(folder):
if file.endswith(".xpi"):
webextensions_to_install.append(os.path.join(folder, file))
webextensions = self.test_config.get("webextensions", None)
if isinstance(webextensions, six.string_types):
webextensions_to_install.append(webextensions)
if webextensions_to_install is not None:
LOG.info("Installing Webextensions:")
for webext in webextensions_to_install:
filename = utils.interpolate(webext)
if mozinfo.os == "win":
filename = filename.replace("/", "\\")
if not filename.endswith(".xpi"):
continue
if not os.path.exists(filename):
continue
LOG.info(filename)
profile.addons.install(filename)
def _run_profile(self):
runner_cls = mozrunner.runners.get(
mozinfo.info.get("appname", "firefox"), mozrunner.Runner
)
args = list(self.browser_config["extra_args"])
args.append(self.browser_config["init_url"])
runner = runner_cls(
profile=self.profile_dir,
binary=self.browser_config["browser_path"],
cmdargs=args,
env=self.env,
)
runner.start(outputTimeout=30)
proc = runner.process_handler
LOG.process_start(
proc.pid, "%s %s" % (self.browser_config["browser_path"], " ".join(args))
)
try:
exit_code = proc.wait()
except Exception:
proc.kill()
raise TalosError("Browser Failed to close properly during warmup")
LOG.process_exit(proc.pid, exit_code)
def _init_gecko_profile(self):
upload_dir = os.getenv("MOZ_UPLOAD_DIR")
if self.test_config.get("gecko_profile") and not upload_dir:
LOG.critical("Profiling ignored because MOZ_UPLOAD_DIR was not" " set")
if upload_dir and self.test_config.get("gecko_profile"):
self.gecko_profile = GeckoProfile(
upload_dir, self.browser_config, self.test_config
)
self.gecko_profile.update_env(self.env)
def clean(self):
try:
mozfile.remove(self._tmp_dir)
except Exception as e:
LOG.info("Exception while removing profile directory: %s" % self._tmp_dir)
LOG.info(e)
if self.gecko_profile:
self.gecko_profile.clean()
def collect_or_clean_ccov(self, clean=False):
# NOTE: Currently only supported when running in production
if not self.browser_config.get("develop", False):
# first see if we an find any ccov files at the ccov output dirs
if clean:
LOG.info("Cleaning ccov files before starting the talos test")
else:
LOG.info(
"Collecting ccov files that were generated during the talos test"
)
gcov_prefix = os.getenv("GCOV_PREFIX", None)
js_ccov_dir = os.getenv("JS_CODE_COVERAGE_OUTPUT_DIR", None)
gcda_archive_folder_name = "gcda-archive"
_gcda_files_found = []
for _ccov_env in [gcov_prefix, js_ccov_dir]:
if _ccov_env is not None:
# ccov output dir env vars exist; now search for gcda files to remove
_ccov_path = os.path.abspath(_ccov_env)
if os.path.exists(_ccov_path):
# now walk through and look for gcda files
LOG.info("Recursive search for gcda files in: %s" % _ccov_path)
for root, dirs, files in os.walk(_ccov_path):
for next_file in files:
if next_file.endswith(".gcda"):
# don't want to move or delete files in our 'gcda-archive'
if root.find(gcda_archive_folder_name) == -1:
_gcda_files_found.append(
os.path.join(root, next_file)
)
else:
LOG.info(
"The ccov env var path doesn't exist: %s" % str(_ccov_path)
)
# now clean or collect gcda files accordingly
if clean:
# remove ccov data
LOG.info(
"Found %d gcda files to clean. Deleting..."
% (len(_gcda_files_found))
)
for _gcda in _gcda_files_found:
try:
mozfile.remove(_gcda)
except Exception as e:
LOG.info("Exception while removing file: %s" % _gcda)
LOG.info(e)
LOG.info("Finished cleaning ccov gcda files")
else:
# copy gcda files to archive folder to be collected later
gcda_archive_top = os.path.join(
gcov_prefix, gcda_archive_folder_name, self.test_config["name"]
)
LOG.info(
"Found %d gcda files to collect. Moving to gcda archive %s"
% (len(_gcda_files_found), str(gcda_archive_top))
)
if not os.path.exists(gcda_archive_top):
try:
os.makedirs(gcda_archive_top)
except OSError:
LOG.critical(
"Unable to make gcda archive folder %s" % gcda_archive_top
)
for _gcda in _gcda_files_found:
# want to copy the existing directory strucutre but put it under archive-dir
# need to remove preceeding '/' from _gcda file name so can join the path
gcda_archive_file = os.path.join(
gcov_prefix,
gcda_archive_folder_name,
self.test_config["name"],
_gcda.strip(gcov_prefix + "//"),
)
gcda_archive_dest = os.path.dirname(gcda_archive_file)
# create archive folder, mirroring structure
if not os.path.exists(gcda_archive_dest):
try:
os.makedirs(gcda_archive_dest)
except OSError:
LOG.critical(
"Unable to make archive folder %s" % gcda_archive_dest
)
# now copy the file there
try:
shutil.copy(_gcda, gcda_archive_dest)
except Exception as e:
LOG.info(
"Error copying %s to %s"
% (str(_gcda), str(gcda_archive_dest))
)
LOG.info(e)
LOG.info(
"Finished collecting ccov gcda files. Copied to: %s"
% gcda_archive_top
)
def __enter__(self):
LOG.info("Initialising browser for %s test..." % self.test_config["name"])
self._init_env()
self._init_profile()
try:
if not self.debug_mode and not self.test_config["name"].startswith("damp"):
self._run_profile()
except BaseException:
self.clean()
raise
self._init_gecko_profile()
LOG.info("Browser initialized.")
LOG.info("Fission enabled: %s" % self.browser_config.get("fission", True))
# remove ccov files before actual tests start
if self.browser_config.get("code_coverage", False):
# if the Firefox build was instrumented for ccov, initializing the browser
# will have caused ccov to output some gcda files; in order to have valid
# ccov data for the talos test we want to remove these files before starting
# the actual talos test(s)
self.collect_or_clean_ccov(clean=True)
return self
def __exit__(self, type, value, tb):
self.clean()