#!/usr/bin/env python3 # # Copyright 2023 Ettus Research, a National Instruments Brand # # SPDX-License-Identifier: GPL-3.0-or-later # """Changeset analyzer: Create list of required tests based on changes in a branch. This script reads changes made on a particular branch compared to another and computes a list of tests that need to be run to validate this branch. """ import argparse import os import pathlib import re import shutil import subprocess import sys import requests from ruamel import yaml def find_git_toplevel(path): """Find the git toplevel by calling "git rev-parse --show-toplevel".""" cmd_args = ["git", "-C", path, "rev-parse", "--show-toplevel"] proc = subprocess.run(cmd_args, capture_output=True) assert proc.returncode == 0, "given path is not a git directory" return proc.stdout.decode().strip() def parse_args(): """Parse and return args.""" parser = argparse.ArgumentParser( description=__doc__, ) parser.add_argument( "--source-branch", help="Source branch off diff. Defaults to the current branch." ) parser.add_argument( "--target-branch", default="master", help="Target branch off diff. Defaults to 'master'." ) parser.add_argument( "--repo-path", default=os.getcwd(), help="Path to the UHD git repository. Defaults to the current directory." "If a subdirectory of the repository is given, the path is automatically resolved" "to the toplevel of the repository.", ) parser.add_argument("--rule-file", help="Path to rules file.") parser.add_argument("--set-azdo-var", help="Generate output to set an AzDO variable") parser.add_argument("--list-tests", action="store_true", help="Show the generated test-list.") parser.add_argument( "--include-target", action="store_true", help="Include changes that originate from the target branch.", ) parser.add_argument( "--github-label-api-endpoint", help="Endpoint to use for GitHub API requests read back labels.", ) parser.add_argument("--github-token", help="GitHub token to use for API requests.") parser.add_argument( "--remove-test", nargs="*", help="Remove a test from the list of tests.", default=[] ) parser.add_argument( "--add-test", nargs="*", help="Add a test to the list of tests.", default=[] ) parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output") args = parser.parse_args() args.repo_path = find_git_toplevel(args.repo_path) return args def get_git_commit_range(target_branch, source_branch=None, include_target=False): """Return the commit range arguments for git diff.""" assert target_branch, "target_branch must be specified" separator = ".." if include_target else "..." if source_branch: return f"{target_branch}{separator}{source_branch}" else: return f"{target_branch}{separator}" def check_changeset_content(file, **kwargs): """Check changeset content for code and/or comments changes. For example, if a .cpp file only has a comment change, we can use this to remove it from the "changed files" list. This is useful when we only want to only include tests if the underlying code was actually changed, and not just a comment (which cannot change the outcome of a hardware test). The values of the include_content argument have the following meaning: - 'code': at least one line with code changed - 'code-only': all changed lines must be code changes - 'comments': at least one line with comments changed - 'comments-only': all changed lines must be comment changes - 'code-and-comments': either code or comments changes (always True) Arguments: file: List of files that have been changed. repo_path: Path to the UHD repository. git_cmd: Path to the git command. include_content: allowed values are: code, code-only, comments, comments-only, code-and-comments Returns: True if the file should be included, False otherwise. """ def identify_comment_line_cpp(line): """Identify if a line is a comment in a C++ file.""" line = line.strip() return line.startswith("//") or line.startswith("/*") or line.startswith("*") def identify_comment_line_py(line): """Identify if a line is a comment in a Python file.""" line = line.strip() return line.startswith("#") comment_identifer = { ".cpp": identify_comment_line_cpp, ".hpp": identify_comment_line_cpp, ".c": identify_comment_line_cpp, ".h": identify_comment_line_cpp, ".py": identify_comment_line_py, } include_content = kwargs.get("include_content", None) if include_content == "code-and-comments": # No need to check, result will always be true return True elif include_content in ["code", "code-only"]: invert = True elif include_content in ["comments", "comments-only"]: invert = False else: raise ValueError(f"Unsupported argument: include_content={include_content}") git_commit_range = get_git_commit_range( kwargs["target_branch"], kwargs.get("source_branch", None), kwargs.get("include_target", False), ) get_diff_args = [ shutil.which("git"), "-C", kwargs["repo_path"], "diff", "--no-color", "--unified=0", git_commit_range, "--", ] ext = os.path.splitext(file)[1] if ext not in comment_identifer: # If we have no rule to check, then always include the file. return True diff_lines = ( subprocess.check_output(get_diff_args + [file], encoding="utf-8").strip().split("\n")[4:] ) line_matches = [ comment_identifer[ext](line[1:]) ^ invert for line in diff_lines if line[0] in ("-", "+") ] if include_content.endswith("-only"): return all(line_matches) else: return any(line_matches) def get_changed_files(repo_path, target_branch, source_branch, include_target): """Return a list of paths in the UHD repository that have are different between two branches. Arguments: repo_path: Path to the UHD repository. target_branch: Branch to compare against (e.g., master). source_branch: Branch to compare from. Defaults to the current branch. include_target: Include changes that originate from the target branch. """ git_commit_range = get_git_commit_range(target_branch, source_branch, include_target) git_cmd = shutil.which("git") get_diff_args = [git_cmd, "diff", "--name-only", git_commit_range] files = subprocess.check_output(get_diff_args, cwd=repo_path, encoding="utf-8") return files.strip().split("\n") def load_rules(rule_file): """Return the rules as a Python list.""" with open(rule_file, "r", encoding="utf-8") as rfd: y = yaml.YAML(typ="safe", pure=True) return y.load(rfd) class RuleApplier: """Helper class to update an internal test list based on a set of rules.""" def __init__(self, rules, labels, initial_list=[], **kwargs): """Initialize. Arguments: rules: List of rules to apply. labels: List of labels relevant to the current changeset. initial_list: Initial list of tests to apply rules against. """ self.rules = rules self.labels = labels self.args = kwargs self.test_list = set(initial_list) def apply(self, filename, verbose=False): """Apply rules against a file.""" for rule in self.rules: if self._apply_rule(rule, filename, verbose): break def apply_labels(self): """Apply rules based only on labels.""" for rule in self.rules: if ( "label" in rule and rule["label"] in self.labels and ("re" not in rule and "name" not in rule) ): self.test_list.update(rule.get("add", [])) self.test_list.difference_update(rule.get("remove", [])) for remove_re in rule.get("remove_re", []): self.test_list.difference_update( set([test for test in self.test_list if re.search(remove_re, test)]) ) def _apply_rule(self, rule, filename, verbose=False): """Apply a single rule. Returns True if you can stop applying rules against this filename. """ # First: Check if this rule even applies to this file if ( ("re" in rule and not re.search(rule["re"], filename)) or ("name" in rule and rule["name"] != filename) or ("name" not in rule and "re" not in rule) ): return False if verbose: sys.stderr.write(f"Filename {filename} matches rule: {rule}\n") if "label" in rule and not rule["label"] in self.labels: if verbose: sys.stderr.write(f"Skipping file based on missing label: {rule['label']}\n") return False if "label" in rule and verbose: sys.stderr.write(f"Label {rule['label']} found\n") include_content = rule.get("include_content", "code") if not check_changeset_content(filename, **self.args, include_content=include_content): if verbose: sys.stderr.write( f"Skipping {filename} based on content rule: include_content='{include_content}'\n" ) return False if "add" in rule: self.test_list.update(rule["add"]) if "remove" in rule: self.test_list.difference_update(rule["remove"]) # If stop is specified as False, then we can still apply more rules. if "stop" in rule and not rule["stop"]: return False return True def remove_tests(self, test_patterns, verbose=False): """Remove tests from the test list.""" if verbose: sys.stderr.write(f"Removing tests based on patterns: {test_patterns}\n") sys.stderr.write(f"Before: {self.test_list}\n") for test_pattern in test_patterns: self.test_list = set( [test for test in self.test_list if not re.search(test_pattern, test)] ) if verbose: sys.stderr.write(f"After: {self.test_list}\n") def get_labels(api_endpoint, token): """Get a list of labels from the GitHub API.""" if not api_endpoint: return [] if token: headers = {"Authorization": f"Bearer {token}"} else: headers = {} data = requests.get(api_endpoint, headers=headers).json() try: labels = [label["name"] for label in data] except (KeyError, TypeError): print("WARNING: Could not get labels from API. Return data: ", data) return [] print("Labels:", ", ".join(labels)) return labels def main(): """Gogogo! Run script.""" args = parse_args() rule_file = args.rule_file if not rule_file: rule_file = os.path.join(pathlib.Path(__file__).parent.resolve(), "changeset_testlist.yaml") file_list = get_changed_files( args.repo_path, args.target_branch, args.source_branch, args.include_target, ) labels = get_labels(args.github_label_api_endpoint, args.github_token) rule_applier = RuleApplier( load_rules(rule_file), labels, initial_list=args.add_test, repo_path=args.repo_path, target_branch=args.target_branch, source_branch=args.source_branch, include_target=args.include_target, verbose=args.verbose, ) for filename in file_list: rule_applier.apply(filename, args.verbose) rule_applier.apply_labels() if args.remove_test: rule_applier.remove_tests(args.remove_test, args.verbose) if args.set_azdo_var: print( f"##vso[task.setvariable variable={args.set_azdo_var};isoutput=true]" + ";".join(rule_applier.test_list) ) if args.list_tests: print("Required tests:") print("---------------", end="") print("\n* ".join(sorted([""] + list(rule_applier.test_list)))) if __name__ == "__main__": sys.exit(main())