Source code
Revision control
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 argparse
import errno
import hashlib
import imp
import os
import six
import sys
from six.moves import configparser
from mozboot.util import get_state_dir
from mozlog.structured import commandline
import manifestdownload
from wptrunner import wptcommandline
manifest = None
def do_delayed_imports(wpt_dir):
global manifest
imp.load_source(
"localpaths", os.path.join(wpt_dir, "tests", "tools", "localpaths.py")
)
sys.path.insert(0, os.path.join(wpt_dir, "tools", "manifest"))
import manifest
def create_parser():
p = argparse.ArgumentParser()
p.add_argument(
"--rebuild", action="store_true", help="Rebuild manifest from scratch"
)
download_group = p.add_mutually_exclusive_group()
download_group.add_argument(
"--download",
dest="download",
action="store_true",
default=None,
help="Always download even if the local manifest is recent",
)
download_group.add_argument(
"--no-download",
dest="download",
action="store_false",
help="Don't try to download the manifest",
)
p.add_argument(
"--no-update",
action="store_false",
dest="update",
default=True,
help="Just download the manifest, don't update",
)
p.add_argument(
"--config",
action="store",
dest="config_path",
default=None,
help="Path to wptrunner config file",
)
p.add_argument(
"--rewrite-config",
action="store_true",
default=False,
help="Force the local configuration to be regenerated",
)
p.add_argument(
"--cache-root",
action="store",
default=os.path.join(get_state_dir(), "cache", "wpt"),
help="Path to use for the metadata cache",
)
commandline.add_logging_group(p)
return p
def ensure_kwargs(kwargs):
_kwargs = vars(create_parser().parse_args([]))
_kwargs.update(kwargs)
return _kwargs
def run(src_root, obj_root, logger=None, **kwargs):
kwargs = ensure_kwargs(kwargs)
if logger is None:
from wptrunner import wptlogging
logger = wptlogging.setup(kwargs, {"mach": sys.stdout})
src_wpt_dir = os.path.join(src_root, "testing", "web-platform")
do_delayed_imports(src_wpt_dir)
if not kwargs["config_path"]:
config_path = generate_config(
logger,
src_root,
src_wpt_dir,
os.path.join(obj_root, "_tests", "web-platform"),
kwargs["rewrite_config"],
)
else:
config_path = kwargs["config_path"]
if not os.path.exists(config_path):
logger.critical("Config file %s does not exist" % config_path)
return None
logger.debug("Using config path %s" % config_path)
test_paths = wptcommandline.get_test_paths(wptcommandline.config.read(config_path))
for paths in six.itervalues(test_paths):
if "manifest_path" not in paths:
paths["manifest_path"] = os.path.join(
paths["metadata_path"], "MANIFEST.json"
)
ensure_manifest_directories(logger, test_paths)
local_config = read_local_config(src_wpt_dir)
for section in ["manifest:upstream", "manifest:mozilla"]:
url_base = local_config.get(section, "url_base")
manifest_rel_path = os.path.join(
local_config.get(section, "metadata"), "MANIFEST.json"
)
test_paths[url_base]["manifest_rel_path"] = manifest_rel_path
if not kwargs["rebuild"] and kwargs["download"] is not False:
force_download = False if kwargs["download"] is None else True
manifestdownload.download_from_taskcluster(
logger, src_root, test_paths, force=force_download
)
else:
logger.debug("Skipping manifest download")
update = kwargs["update"] or kwargs["rebuild"]
manifests = load_and_update(
logger,
src_wpt_dir,
test_paths,
update=update,
rebuild=kwargs["rebuild"],
cache_root=kwargs["cache_root"],
)
return manifests
def ensure_manifest_directories(logger, test_paths):
for paths in six.itervalues(test_paths):
manifest_dir = os.path.dirname(paths["manifest_path"])
if not os.path.exists(manifest_dir):
logger.info("Creating directory %s" % manifest_dir)
# Even though we just checked the path doesn't exist, there's a chance
# of race condition with another process or thread having created it in
# between. This happens during tests.
try:
os.makedirs(manifest_dir)
except OSError as e:
if e.errno != errno.EEXIST:
raise
elif not os.path.isdir(manifest_dir):
raise IOError("Manifest directory is a file")
def read_local_config(wpt_dir):
src_config_path = os.path.join(wpt_dir, "wptrunner.ini")
parser = configparser.SafeConfigParser()
success = parser.read(src_config_path)
assert src_config_path in success
return parser
def generate_config(logger, repo_root, wpt_dir, dest_path, force_rewrite=False):
"""Generate the local wptrunner.ini file to use locally"""
if not os.path.exists(dest_path):
# Even though we just checked the path doesn't exist, there's a chance
# of race condition with another process or thread having created it in
# between. This happens during tests.
try:
os.makedirs(dest_path)
except OSError as e:
if e.errno != errno.EEXIST:
raise
dest_config_path = os.path.join(dest_path, "wptrunner.local.ini")
if not force_rewrite and os.path.exists(dest_config_path):
logger.debug("Config is up to date, not regenerating")
return dest_config_path
logger.info("Creating config file %s" % dest_config_path)
parser = read_local_config(wpt_dir)
for section in ["manifest:upstream", "manifest:mozilla"]:
meta_rel_path = parser.get(section, "metadata")
tests_rel_path = parser.get(section, "tests")
parser.set(
section, "manifest", os.path.join(dest_path, meta_rel_path, "MANIFEST.json")
)
parser.set(section, "metadata", os.path.join(wpt_dir, meta_rel_path))
parser.set(section, "tests", os.path.join(wpt_dir, tests_rel_path))
parser.set(
"paths",
"prefs",
os.path.abspath(os.path.join(wpt_dir, parser.get("paths", "prefs"))),
)
with open(dest_config_path, "wt") as config_file:
parser.write(config_file)
return dest_config_path
def load_and_update(
logger,
wpt_dir,
test_paths,
rebuild=False,
config_dir=None,
cache_root=None,
update=True,
):
rv = {}
wptdir_hash = hashlib.sha256(os.path.abspath(wpt_dir).encode()).hexdigest()
for url_base, paths in six.iteritems(test_paths):
manifest_path = paths["manifest_path"]
this_cache_root = os.path.join(
cache_root, wptdir_hash, os.path.dirname(paths["manifest_rel_path"])
)
m = manifest.manifest.load_and_update(
paths["tests_path"],
manifest_path,
url_base,
update=update,
rebuild=rebuild,
working_copy=True,
cache_root=this_cache_root,
)
path_data = {"url_base": url_base}
path_data.update(paths)
rv[m] = path_data
return rv
def log_error(logger, manifest_path, msg):
logger.lint_error(
path=manifest_path, message=msg, lineno=0, source="", linter="wpt-manifest"
)