import argparse import logging import os import re import subprocess import sys from collections import OrderedDict try: 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: print(branch_point()) 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) else: # Otherwise we aren't on a PR, so we try to find commits that are only in the # current branch c.f. # http://stackoverflow.com/questions/13460152/find-first-ancestor-commit-in-another-branch # 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, stdin=subprocess.PIPE, stdout=subprocess.PIPE, cwd=wpt_root) commits_bytes, _ = proc.communicate(b"\n".join(item.encode("ascii") for item in not_heads)) if proc.returncode != 0: raise subprocess.CalledProcessError(proc.returncode, cmd, commits_bytes) 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 break if branch_point: break # 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 else: 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]) + "[^/]*") else: re_parts.append(re.escape(part)) 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: continue else: if not os.path.isdir(path): files.add(path) else: 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: ignore_rules = DEFAULT_IGNORE_RULES 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): ignored.append(fullpath) break else: changed.append(fullpath) 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, include_uncommitted=include_uncommitted, include_new=include_new) 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, "/", update=manifest_update) 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: continue 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 break 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 re.search(regex, 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: dirs.remove(dir_name) 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: continue if affected_by_wdspec(test_full_path): affected_testfiles.add(test_full_path) continue with open(test_full_path, "rb") as fh: raw_file_contents: bytes = fh.read() 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") else: 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): affected_testfiles.add(test_full_path) continue 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: " + ", ".join(DEFAULT_IGNORE_RULES)) 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() parser.add_argument("--metadata", dest="metadata_root", action="store", default=wpt_root, 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, kwargs["ignore_rule"], include_uncommitted=kwargs["modified"], include_new=kwargs["new"]) separator = "\0" if kwargs["null"] else "\n" for item in sorted(changed): line = os.path.relpath(item, wpt_root) + separator sys.stdout.write(line) def run_tests_affected(**kwargs: Any) -> None: revish = get_revish(**kwargs) changed, _ = files_changed(revish, kwargs["ignore_rule"], include_uncommitted=kwargs["modified"], include_new=kwargs["new"]) manifest_path = os.path.join(kwargs["metadata_root"], "MANIFEST.json") tests_changed, dependents = affected_testfiles( changed, {"conformance-checkers", "docs", "tools"}, manifest_path=manifest_path ) 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() sys.stdout.write(message.format(**results))