Source code

Revision control

Copy as Markdown

Other Tools

import argparse
import logging
import os
import re
import subprocess
import sys
from collections import OrderedDict
from ..manifest import manifest
from ..manifest.utils import git as get_git_cmd
except ValueError:
# if we're not within the tools package, the above is an import from above
# the top-level which raises ValueError, so reimport it with an absolute
# reference
# note we need both because depending on caller we may/may not have the
# paths set up correctly to handle both and mypy has no knowledge of our
# sys.path magic
from manifest import manifest # type: ignore
from manifest.utils import git as get_git_cmd # type: ignore
from typing import Any, Dict, Iterable, List, Optional, Pattern, Sequence, Set, Text, Tuple
DEFAULT_IGNORE_RULES = ("resources/testharness*", "resources/testdriver*")
here = os.path.dirname(__file__)
wpt_root = os.path.abspath(os.path.join(here, os.pardir, os.pardir))
logger = logging.getLogger()
def display_branch_point() -> None:
def branch_point() -> Optional[Text]:
git = get_git_cmd(wpt_root)
if git is None:
raise Exception("git not found")
if (os.environ.get("GITHUB_PULL_REQUEST", "false") == "false" and
os.environ.get("GITHUB_BRANCH") == "master"):
# For builds on the master branch just return the HEAD commit
return git("rev-parse", "HEAD")
elif os.environ.get("GITHUB_PULL_REQUEST", "false") != "false":
# This is a PR, so the base branch is in GITHUB_BRANCH
base_branch = os.environ.get("GITHUB_BRANCH")
assert base_branch, "GITHUB_BRANCH environment variable is defined"
branch_point: Optional[Text] = git("merge-base", "HEAD", base_branch)
# Otherwise we aren't on a PR, so we try to find commits that are only in the
# current branch c.f.
# parse HEAD into an object ref
head = git("rev-parse", "HEAD")
# get everything in refs/heads and refs/remotes that doesn't include HEAD
not_heads = [item for item in git("rev-parse", "--not", "--branches", "--remotes").split("\n")
if item and item != "^%s" % head]
# get all commits on HEAD but not reachable from anything in not_heads
cmd = ["git", "rev-list", "--topo-order", "--parents", "--stdin", "HEAD"]
proc = subprocess.Popen(cmd,
commits_bytes, _ = proc.communicate(b"\n".join(item.encode("ascii") for item in not_heads))
if proc.returncode != 0:
raise subprocess.CalledProcessError(proc.returncode,
commit_parents: Dict[Text, List[Text]] = OrderedDict()
commits = commits_bytes.decode("ascii")
if commits:
for line in commits.split("\n"):
line_commits = line.split(" ")
commit_parents[line_commits[0]] = line_commits[1:]
branch_point = None
# if there are any commits, take the first parent that is not in commits
for commit, parents in commit_parents.items():
for parent in parents:
if parent not in commit_parents:
branch_point = parent
if branch_point:
# if we had any commits, we should now have a branch point
assert branch_point or not commit_parents
# The above heuristic will fail in the following cases:
# - The current branch has fallen behind the remote version
# - Changes on the current branch were rebased and therefore do not exist on any
# other branch. This will result in the selection of a commit that is earlier
# in the history than desired (as determined by calculating the later of the
# branch point and the merge base)
# In either case, fall back to using the merge base as the branch point.
merge_base = git("merge-base", "HEAD", "origin/master")
if (branch_point is None or
(branch_point != merge_base and
not git("log", "--oneline", f"{merge_base}..{branch_point}").strip())):
logger.debug("Using merge-base as the branch point")
branch_point = merge_base
logger.debug("Using first commit on another branch as the branch point")
logger.debug("Branch point from master: %s" % branch_point)
if branch_point:
branch_point = branch_point.strip()
return branch_point
def compile_ignore_rule(rule: Text) -> Pattern[Text]:
rule = rule.replace(os.path.sep, "/")
parts = rule.split("/")
re_parts = []
for part in parts:
if part.endswith("**"):
re_parts.append(re.escape(part[:-2]) + ".*")
elif part.endswith("*"):
re_parts.append(re.escape(part[:-1]) + "[^/]*")
return re.compile("^%s$" % "/".join(re_parts))
def repo_files_changed(revish: Text, include_uncommitted: bool = False, include_new: bool = False) -> Set[Text]:
git = get_git_cmd(wpt_root)
if git is None:
raise Exception("git not found")
if "..." in revish:
raise Exception(f"... not supported when finding files changed (revish: {revish!r}")
if ".." in revish:
# ".." isn't treated as a range for git-diff; what we want is
# everything reachable from B but not A, and git diff A...B
# gives us that (via the merge-base)
revish = revish.replace("..", "...")
files_list = git("diff", "--no-renames", "--name-only", "-z", revish).split("\0")
assert not files_list[-1], f"final item should be empty, got: {files_list[-1]!r}"
files = set(files_list[:-1])
if include_uncommitted:
entries = git("status", "-z").split("\0")
assert not entries[-1]
entries = entries[:-1]
for item in entries:
status, path = item.split(" ", 1)
if status == "??" and not include_new:
if not os.path.isdir(path):
for dirpath, dirnames, filenames in os.walk(path):
for filename in filenames:
files.add(os.path.join(dirpath, filename))
return files
def exclude_ignored(files: Iterable[Text], ignore_rules: Optional[Sequence[Text]]) -> Tuple[List[Text], List[Text]]:
if ignore_rules is None:
compiled_ignore_rules = [compile_ignore_rule(item) for item in set(ignore_rules)]
changed = []
ignored = []
for item in sorted(files):
fullpath = os.path.join(wpt_root, item)
rule_path = item.replace(os.path.sep, "/")
for rule in compiled_ignore_rules:
if rule.match(rule_path):
return changed, ignored
def files_changed(revish: Text,
ignore_rules: Optional[Sequence[Text]] = None,
include_uncommitted: bool = False,
include_new: bool = False
) -> Tuple[List[Text], List[Text]]:
"""Find files changed in certain revisions.
The function passes `revish` directly to `git diff`, so `revish` can have a
variety of forms; see `git diff --help` for details. Files in the diff that
are matched by `ignore_rules` are excluded.
files = repo_files_changed(revish,
if not files:
return [], []
return exclude_ignored(files, ignore_rules)
def _in_repo_root(full_path: Text) -> bool:
rel_path = os.path.relpath(full_path, wpt_root)
path_components = rel_path.split(os.sep)
return len(path_components) < 2
def load_manifest(manifest_path: Optional[Text] = None, manifest_update: bool = True) -> manifest.Manifest:
if manifest_path is None:
manifest_path = os.path.join(wpt_root, "MANIFEST.json")
return manifest.load_and_update(wpt_root, manifest_path, "/",
def affected_testfiles(files_changed: Iterable[Text],
skip_dirs: Optional[Set[Text]] = None,
manifest_path: Optional[Text] = None,
manifest_update: bool = True
) -> Tuple[Set[Text], Set[Text]]:
"""Determine and return list of test files that reference changed files."""
if skip_dirs is None:
skip_dirs = {"conformance-checkers", "docs", "tools"}
affected_testfiles = set()
# Exclude files that are in the repo root, because
# they are not part of any test.
files_changed = [f for f in files_changed if not _in_repo_root(f)]
nontests_changed = set(files_changed)
wpt_manifest = load_manifest(manifest_path, manifest_update)
test_types = ["crashtest", "print-reftest", "reftest", "testharness", "wdspec"]
support_files = {os.path.join(wpt_root, path)
for _, path, _ in wpt_manifest.itertypes("support")}
wdspec_test_files = {os.path.join(wpt_root, path)
for _, path, _ in wpt_manifest.itertypes("wdspec")}
test_files = {os.path.join(wpt_root, path)
for _, path, _ in wpt_manifest.itertypes(*test_types)}
interface_dir = os.path.join(wpt_root, 'interfaces')
interfaces_files = {os.path.join(wpt_root, 'interfaces', filename)
for filename in os.listdir(interface_dir)}
interfaces_changed = interfaces_files.intersection(nontests_changed)
nontests_changed = nontests_changed.intersection(support_files)
tests_changed = {item for item in files_changed if item in test_files}
nontest_changed_paths = set()
rewrites: Dict[Text, Text] = {"/resources/webidl2/lib/webidl2.js": "/resources/WebIDLParser.js"}
for full_path in nontests_changed:
rel_path = os.path.relpath(full_path, wpt_root)
path_components = rel_path.split(os.sep)
top_level_subdir = path_components[0]
if top_level_subdir in skip_dirs:
repo_path = "/" + os.path.relpath(full_path, wpt_root).replace(os.path.sep, "/")
if repo_path in rewrites:
repo_path = rewrites[repo_path]
full_path = os.path.join(wpt_root, repo_path[1:].replace("/", os.path.sep))
nontest_changed_paths.add((full_path, repo_path))
interfaces_changed_names = [os.path.splitext(os.path.basename(interface))[0]
for interface in interfaces_changed]
def affected_by_wdspec(test: Text) -> bool:
affected = False
if test in wdspec_test_files:
for support_full_path, _ in nontest_changed_paths:
# parent of support file or of "support" directory
parent = os.path.dirname(support_full_path)
if os.path.basename(parent) == "support":
parent = os.path.dirname(parent)
relpath = os.path.relpath(test, parent)
if not relpath.startswith(os.pardir):
# testfile is in subtree of support file
affected = True
return affected
def affected_by_interfaces(file_contents: Text) -> bool:
if len(interfaces_changed_names) > 0:
if 'idlharness.js' in file_contents:
for interface in interfaces_changed_names:
regex = '[\'"]' + interface + '(\\.idl)?[\'"]'
if, file_contents):
return True
return False
for root, dirs, fnames in os.walk(wpt_root):
# Walk top_level_subdir looking for test files containing either the
# relative filepath or absolute filepath to the changed files.
if root == wpt_root:
for dir_name in skip_dirs:
for fname in fnames:
test_full_path = os.path.join(root, fname)
# Skip any file that's not a test file.
if test_full_path not in test_files:
if affected_by_wdspec(test_full_path):
with open(test_full_path, "rb") as fh:
raw_file_contents: bytes =
if raw_file_contents.startswith(b"\xfe\xff"):
file_contents: Text = raw_file_contents.decode("utf-16be", "replace")
elif raw_file_contents.startswith(b"\xff\xfe"):
file_contents = raw_file_contents.decode("utf-16le", "replace")
file_contents = raw_file_contents.decode("utf8", "replace")
for full_path, repo_path in nontest_changed_paths:
rel_path = os.path.relpath(full_path, root).replace(os.path.sep, "/")
if rel_path in file_contents or repo_path in file_contents or affected_by_interfaces(file_contents):
return tests_changed, affected_testfiles
def get_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser()
parser.add_argument("revish", default=None, help="Commits to consider. Defaults to the "
"commits on the current branch", nargs="?")
parser.add_argument("--ignore-rule", action="append",
help="Override the rules for paths to exclude from lists of changes. "
"Rules are paths relative to the test root, with * before a separator "
"or the end matching anything other than a path separator and ** in that "
"position matching anything. This flag can be used multiple times for "
"multiple rules. Specifying this flag overrides the default: " +
parser.add_argument("--modified", action="store_true",
help="Include files under version control that have been "
"modified or staged")
parser.add_argument("--new", action="store_true",
help="Include files in the worktree that are not in version control")
parser.add_argument("--show-type", action="store_true",
help="Print the test type along with each affected test")
parser.add_argument("--null", action="store_true",
help="Separate items with a null byte")
return parser
def get_parser_affected() -> argparse.ArgumentParser:
parser = get_parser()
help="Directory that will contain MANIFEST.json")
return parser
def get_revish(**kwargs: Any) -> Text:
revish = kwargs.get("revish")
if revish is None:
revish = "%s..HEAD" % branch_point()
return revish.strip()
def run_changed_files(**kwargs: Any) -> None:
revish = get_revish(**kwargs)
changed, _ = files_changed(revish,
separator = "\0" if kwargs["null"] else "\n"
for item in sorted(changed):
line = os.path.relpath(item, wpt_root) + separator
def run_tests_affected(**kwargs: Any) -> None:
revish = get_revish(**kwargs)
changed, _ = files_changed(revish,
manifest_path = os.path.join(kwargs["metadata_root"], "MANIFEST.json")
tests_changed, dependents = affected_testfiles(
{"conformance-checkers", "docs", "tools"},
message = "{path}"
if kwargs["show_type"]:
wpt_manifest = load_manifest(manifest_path)
message = "{path}\t{item_type}"
message += "\0" if kwargs["null"] else "\n"
for item in sorted(tests_changed | dependents):
results = {
"path": os.path.relpath(item, wpt_root)
if kwargs["show_type"]:
item_types = {i.item_type for i in wpt_manifest.iterpath(results["path"])}
if len(item_types) != 1:
item_types = {" ".join(item_types)}
results["item_type"] = item_types.pop()