revng-revng/libexec/revng/revng-check-conventions

#!/usr/bin/env python3

#
# This file is distributed under the MIT License. See LICENSE.md for details.
#

import abc
import argparse
import asyncio
import os
import re
import sys
import time
from asyncio import create_subprocess_exec
from collections import defaultdict
from dataclasses import dataclass
from itertools import groupby
from pathlib import Path
from string import Template
from subprocess import PIPE, STDOUT, check_output
from subprocess import run as process_run
from textwrap import indent
from typing import Dict, Iterable, List, Mapping, NoReturn, Optional, Protocol, Sequence, Set
from typing import Sized, TypeVar, Union

import yaml

T = TypeVar("T")
RuleResult = Mapping[str, Sequence[str]]
LineResult = Mapping[Path, Sequence[int]]
FileResult = Sequence[Path]
term_colors = {
    "reset": "\033[0;0m",
    "red": "\033[1;31m",
    "green": "\033[1;32m",
    "blue": "\033[1;34m",
    "purple": "\033[1;35m",
    "cyan": "\033[1;36m",
}


def colored(text, color: str) -> str:
    return f"{term_colors[color]}{text}{term_colors['reset']}"


term_colon = colored(":", "cyan")


class Matcher(Protocol):
    def __call__(self, path: Path, line: str) -> List[re.Match]:
        """Abstraction that's used by passes to check if line matches"""
        ...


class FileFilter(Protocol):
    def __call__(self, path: Path, content: str) -> bool:
        """Used in filters to determine if a file belongs to a tag
        path is the path of the file and content is the first 512 bytes of it
        """
        ...


@dataclass
class ProcessResult:
    output: str
    returncode: int


@dataclass
class PassResult:
    output: str
    ok: bool

    def verify(self):
        return self.output.strip() == "" and self.ok

    @staticmethod
    def _check_processes(*args: ProcessResult) -> bool:
        combined_stdout = "\n".join(p.output for p in args)
        return all(p.returncode == 0 for p in args) and combined_stdout.strip() == ""

    @classmethod
    def from_process(cls, *processes: ProcessResult) -> "PassResult":
        return PassResult("\n".join(p.output for p in processes), cls._check_processes(*processes))

    @staticmethod
    def combine(*pass_results: "PassResult") -> "PassResult":
        return PassResult(
            "\n".join(p.output for p in pass_results), all(p.ok for p in pass_results)
        )

    @staticmethod
    def make_from(
        *,
        files: Mapping[str, FileResult] = {},
        lines: Mapping[str, LineResult] = {},
        rules: Iterable[RuleResult] = (),
    ) -> "PassResult":
        output = ""
        output += "".join(generate_error_files(m, f) for m, f in files.items())
        output += "".join(generate_error_lines(m, l) for m, l in lines.items())
        output += "".join(generate_rule_results(r) for r in rules)
        return PassResult(output, all_empty(*files.values(), *lines.values(), *rules))


class Pass(abc.ABC):
    """Abstraction that describes the operations on a set of files
    When called it will be given a dictionary of <tag>:list<files> and a
    dictionary of arguments (e.g. dry_run=1), in general the pass can operate on
    more than one tag, but in most cases it's better to override the `process` method
    that will give a list of files given the tag attribute"""

    def __init__(self, name: str, tag: str):
        self.name = name
        self.tag = tag

    async def run(self, files: Dict[str, Dict[Path, str]], args: Dict[str, str]) -> PassResult:
        file_list = files[self.tag]
        if len(file_list.keys()) == 0:
            return PassResult("", True)
        return await self.process(files[self.tag], args)

    @abc.abstractmethod
    async def process(self, files: Dict[Path, str], args: Dict[str, str]) -> PassResult:
        ...


def log(string: str):
    sys.stdout.write(f"{string}\n")


def fail(reason: str, returncode: int = 1) -> NoReturn:
    sys.stderr.write(f"{reason}\n")
    sys.stderr.flush()
    sys.exit(returncode)


def template_iterable(iterable: Iterable[str], args: Dict[str, str]) -> List[str]:
    return [Template(element).substitute(args) for element in iterable]


async def run_with_capture(cmd: Sequence[Union[str, Path]]) -> ProcessResult:
    proc = await create_subprocess_exec(cmd[0], *cmd[1:], stdout=PIPE, stderr=STDOUT)
    returncode = await proc.wait()
    assert proc.stdout is not None
    stdout = await proc.stdout.read()
    return ProcessResult(stdout.decode("utf-8"), returncode)


def read_file(filename, size: Optional[int] = None):
    with open(filename) as f:
        return f.read(size)


def chunk(values: Sequence[T], number: int = 1) -> List[List[T]]:
    res: List[List[T]] = [[] for _ in range(0, number)]
    for i in range(0, len(values)):
        res[i % number].append(values[i])
    return [r for r in res if len(r) > 0]


def all_empty(*args: Sized) -> bool:
    return all(len(a) == 0 for a in args)


def deserialize_config_item(dictionary: dict):
    assert "type" in dictionary
    type_ = dictionary["type"]
    assert type_ in globals(), f"Type {type_} not defined"
    return globals()[type_](**{k: v for k, v in dictionary.items() if k != "type"})


def deserialize_matchers(matcher_list: List[dict]) -> Dict[str, Matcher]:
    return {f"{item['name']}:": deserialize_config_item(item["matcher"]) for item in matcher_list}


def highlight_line(line: str, matches: List[re.Match]) -> str:
    result = ""
    is_red = False
    for index, char in enumerate(line):
        if any(m.start() <= index < m.end() for m in matches):
            result += (term_colors["red"] if not is_red else "") + char
            is_red = True
        else:
            result += (term_colors["reset"] if is_red else "") + char
            is_red = False
    return result + (term_colors["reset"] if is_red else "")


def line_matcher(files: Dict[Path, str], matchers: Mapping[str, Matcher]) -> RuleResult:
    result: Dict[str, List[str]] = defaultdict(list)
    for path, content in files.items():
        for lineno, line in enumerate(content.splitlines()):
            for msg, matcher in matchers.items():
                if matches := matcher(path, line):
                    result[msg].append(
                        f"{colored(path, 'purple')}{term_colon}{colored(lineno+1, 'green')}"
                        f"{term_colon} {highlight_line(line, matches)}"
                    )
    return result


def regexes_matcher(regexes: Iterable[str]) -> Matcher:
    compiled_regexes = [re.compile(r) for r in regexes]

    def matcher(path: Path, line: str):
        result = []
        for regex in compiled_regexes:
            if match := regex.search(line):
                result.append(match)
        return result

    return matcher


def regexes_matcher_prefix(prefix: str, regexes: Iterable[str]) -> Matcher:
    return regexes_matcher((prefix + r for r in regexes))


def any_startswith(string: str, prefixes: Iterable[str]) -> bool:
    return any(string.startswith(p) for p in prefixes)


def join_items(items: Iterable[str], join: str = "\n", indent_: str = "  ") -> str:
    return indent("".join(f"{i}{join}" for i in items), indent_)


def generate_error_files(msg: str, files: FileResult) -> str:
    if len(files) == 0:
        return ""
    return f"{msg}\n" + join_items((colored(f, "purple") for f in files))


def generate_error_lines(msg: str, files: LineResult) -> str:
    if len(files) == 0:
        return ""
    return f"{msg}\n" + join_items(
        (
            f"{colored(path, 'purple')}{term_colon} {'line' if len(lines) == 1 else 'lines'} "
            + join_items((colored(line, "green") for line in lines), ", ", "")
            for path, lines in files.items()
        ),
    )


def generate_rule_results(result: RuleResult) -> str:
    return "".join(f"{rule}\n" + join_items(items) for rule, items in result.items())


class SingleCommandPass(Pass):
    def __init__(
        self,
        name: str,
        tag: str,
        command: Iterable[str],
        dry_command: Optional[Iterable[str]] = None,
    ):
        super().__init__(name, tag)
        self.command = command
        self.dry_command = dry_command if dry_command is not None else command

    async def process(self, files: Dict[Path, str], args) -> PassResult:
        cmd_args = template_iterable(self.dry_command if args["dry_run"] else self.command, args)
        process = await run_with_capture((*cmd_args, *map(str, files)))
        return PassResult.from_process(process)


class ParallelCommandPass(Pass):
    def __init__(
        self,
        name: str,
        tag: str,
        command: Iterable[str],
        dry_command: Optional[Iterable[str]] = None,
    ):
        super().__init__(name, tag)
        self.command = command
        self.dry_command = dry_command if dry_command is not None else command

    async def process(self, files: Dict[Path, str], args) -> PassResult:
        cpu_count = os.cpu_count()
        chunks = chunk(list(files), cpu_count if cpu_count else 1)
        cmd_args = template_iterable(self.dry_command if args["dry_run"] else self.command, args)
        processes = [run_with_capture((*cmd_args, *files)) for files in chunks]
        process_results = await asyncio.gather(*processes)
        return PassResult.from_process(*process_results)


class MyPyPass(SingleCommandPass):
    def __init__(self, command: List[str]):
        super().__init__("mypy", "python", command)

    async def process(self, files: Dict[Path, str], args) -> PassResult:
        script_files: List[Path] = []
        normal_files: List[Path] = []
        for filepath in files:
            if str(filepath).startswith("python/scripts"):
                script_files.append(filepath)
            else:
                normal_files.append(filepath)

        runs = []

        if normal_files:
            runs.append(await super().process({f: "" for f in normal_files}, args))

        for script in script_files:
            runs.append(await super().process({script: ""}, args))
        return PassResult.combine(*runs)


class LicenseCheckPass(Pass):
    def __init__(
        self,
        ignore_suffixes: List[str],
        ignore_pathnames: List[str],
        ignore_stems: List[str],
    ):
        super().__init__("license-check", "all")
        self.ignore_suffixes = ignore_suffixes
        self.ignore_pathnames = ignore_pathnames
        self.ignore_stems = ignore_stems

    def skip_test(self, path: Path) -> bool:
        return (
            path.suffix in self.ignore_suffixes
            or path.name in self.ignore_pathnames
            or path.stem in self.ignore_stems
        )

    @staticmethod
    def check_license(content: str) -> bool:
        return any("See LICENSE.md for details" in line for line in content.splitlines()[:10])

    async def process(self, files: Dict[Path, str], args: Dict[str, str]) -> PassResult:
        missing_license = []
        for path, content in files.items():
            if self.skip_test(path):
                continue
            if not self.check_license(content):
                missing_license.append(path)
        return PassResult.make_from(
            files={"There are files without a license header:": missing_license}
        )


class BashPass(SingleCommandPass):
    def __init__(self):
        super().__init__("bash-check", "bash", ("shellcheck",))

    @staticmethod
    def check_set(content: str) -> bool:
        return any(
            "# rcc-ignore: bash-set-flags" in line or "set -euo pipefail" in line
            for line in content.splitlines()[:10]
        )

    async def process(self, files: Dict[Path, str], args: Dict[str, str]) -> PassResult:
        shellcheck = await super().process(files, args)
        noset_files = [path for path, content in files.items() if not self.check_set(content)]
        output = generate_error_files(
            "There are script files without 'set -euo pipefail':", noset_files
        )
        return PassResult.combine(shellcheck, PassResult(output, len(noset_files) == 0))


class WhitespaceCheckPass(Pass):
    def __init__(self):
        super().__init__("whitespace-check", "all")

    async def process(self, files: Dict[Path, str], args) -> PassResult:
        trailing_whitespace = defaultdict(list)
        tabs = defaultdict(list)
        no_end_newline = []
        for path, content in files.items():
            lines = content.splitlines()
            for index, line in enumerate(lines):
                if line.endswith(" "):
                    trailing_whitespace[path].append(index + 1)
                if "\t" in line:
                    tabs[path].append(index + 1)
            if content == "" or content[-1] != "\n":
                no_end_newline.append(path)

        return PassResult.make_from(
            files={"Files that don't end in a newline:": no_end_newline},
            lines={
                "Whitespace at the end of line:": trailing_whitespace,
                "Tabs present:": tabs,
            },
        )


class CPPCheckPass(Pass):
    def __init__(
        self,
        support_files: List[str],
        ignore_cpp: Dict[str, List[str]],
        matchers: Dict[str, List[dict]],
    ):
        super().__init__("cpp-checks", "c")
        self.support_files = support_files
        self.ignore_cpp = ignore_cpp
        self.non_support_matchers = deserialize_matchers(matchers["non_support"])
        self.header_matchers = deserialize_matchers(matchers["header"])
        self.cpp_matchers = deserialize_matchers(matchers["cpp"])
        self.cpp_object_matchers = deserialize_matchers(matchers["cpp_object"])
        self.generic_matchers = deserialize_matchers(matchers["generic"])
        self.generic_matchers["Parenthesis at the end of line:"] = self.parethesis_matcher

    @staticmethod
    def parethesis_matcher(path: Path, line: str) -> List[re.Match]:
        if re.search(r"R\"LLVM.*\($", line):
            return []
        main_match = re.search(r"\($", line)
        return [main_match] if main_match else []

    @staticmethod
    def include_matcher(cpp_files: Dict[Path, str]) -> Matcher:
        def include_matcher_inner(path: Path, line: str) -> List[re.Match]:
            if path in cpp_files:
                match = re.search(r"^\s*#include <.*\.h(|pp)>", line)
            else:
                match = re.search(r"^\s*#include <.*\.hpp>", line)
            return [match] if match else []

        return include_matcher_inner

    @staticmethod
    def initrevng_matcher(cpp_files: Dict[Path, str]) -> List[Path]:
        return [
            file
            for file, contents in cpp_files.items()
            if re.search(r"int main\([^)]+\) {", contents)
            and ("// rcc-ignore: initrevng" not in contents and "InitRevng" not in contents)
        ]

    async def process(self, files: Dict[Path, str], args) -> PassResult:
        non_support_files = {
            path: content for path, content in files.items() if str(path) not in self.support_files
        }

        cpp_files = {
            path: content
            for path, content in files.items()
            if not any_startswith(str(path), self.ignore_cpp["paths"])
            and str(path) not in self.ignore_cpp["files"]
            and path.suffix != ".c"
        }

        header_files = {path: content for path, content in cpp_files.items() if path.suffix == ".h"}
        cpp_object_files = {
            path: content for path, content in cpp_files.items() if path.suffix != ".h"
        }

        include_matcher = {
            "Includes should never use <..> except for C++ "
            "standard includes:": self.include_matcher(cpp_files)
        }

        generic_result = line_matcher(files, self.generic_matchers)
        non_support_result = line_matcher(non_support_files, self.non_support_matchers)
        cpp_include_result = line_matcher(files, include_matcher)
        cpp_result = line_matcher(cpp_files, self.cpp_matchers)
        cpp_object_result = line_matcher(cpp_object_files, self.cpp_object_matchers)
        header_result = line_matcher(header_files, self.header_matchers)
        main_no_init = self.initrevng_matcher(cpp_object_files)

        headers_without_pragma_once = [
            file
            for file, content in header_files.items()
            if content.splitlines()[0] != "#pragma once"
        ]

        return PassResult.make_from(
            files={
                "Header does not start with #pragma once": headers_without_pragma_once,
                "Main executable does not have 'InitRevng'": main_no_init,
            },
            rules=(
                generic_result,
                non_support_result,
                cpp_result,
                cpp_include_result,
                header_result,
                cpp_object_result,
            ),
        )


def suffix_or_shebang(suffixes: Sequence[str] = (), shebang: str = "") -> FileFilter:
    def matcher(path: Path, content: str) -> bool:
        if len(suffixes) > 0 and path.suffix.startswith(".") and path.suffix[1:] in suffixes:
            return True
        if shebang != "" and content != "":
            first_line = content.splitlines()[0]
            if re.search(r"#!.*" + shebang, first_line):
                return True
        return False

    return matcher


def cmake_filter() -> FileFilter:
    return lambda path, content: "cmake" in path.name.lower()


def all_filter() -> FileFilter:
    return lambda path, content: True


def get_config(filename: str) -> Path:
    filepath = Path(__file__).parent / "../../share/revng" / filename
    if filepath.is_file():
        return filepath
    else:
        fail(f"Could not find config file {filename}")  # noqa: R503


def read_rcc_config():
    config_path = get_config("rcc-config.yml")
    with open(config_path) as config_file:
        return yaml.safe_load(config_file)


def filter_files(
    file_content_list: Dict[Path, str], filters: Dict[str, FileFilter]
) -> Dict[str, Dict[Path, str]]:
    return {
        tag: {
            filename: content
            for filename, content in file_content_list.items()
            if matcher(filename, content)
        }
        for tag, matcher in filters.items()
    }


def git_ls_files_wrapper(*args: str) -> List[str]:
    file_list = check_output(["git", "ls-files", "-z", *args], text=True)
    deleted_list = check_output(["git", "ls-files", "-z", "--deleted"], text=True)
    return [f for f in file_list.split("\x00") if f != "" and f not in deleted_list.split("\x00")]


def git_diff_wrapper(*args: str) -> List[str]:
    file_list = check_output(
        ["git", "diff", "-z", "--diff-filter=AMRC", "--name-status", *args], text=True
    )
    split_file_list = [c for c in file_list.split("\x00") if c != ""]
    result = []
    while len(split_file_list) > 0:
        op = split_file_list.pop(0)[0]
        if op in ["A", "M"]:
            result.append(split_file_list.pop(0))
        elif op in ["R", "C"]:
            split_file_list.pop(0)
            result.append(split_file_list.pop(0))
        else:
            fail(f"Invalid git operation: {op}")
    return result


def get_file_list(args) -> List[Path]:
    if len(args.FILES) > 0:
        files = args.FILES
        for file in (f for f in files if not os.path.isfile(f)):
            fail(f"File {file} does not exist")
    elif args.HEAD:
        files = git_diff_wrapper("HEAD^..HEAD")
    elif args.commit_range:
        files = git_diff_wrapper(args.commit_range)
    elif args.staged:
        files = git_diff_wrapper("--cached")
    elif args.unstaged:
        files = git_ls_files_wrapper("-m", "--exclude-standard")
    else:
        files = git_ls_files_wrapper()
    return [Path(f) for f in files]


def read_clang_style(args) -> str:
    if not args.use_local_clang_format_file:
        style = get_config("clang-format-style-file").read_text()
    else:
        style = "file"
    if args.print_clang_format_config:
        process_run(["clang-format", "--dry-run", f"-style={style}", "--dump-config"])
        sys.exit(0)
    return style


async def inflate_files(initial_dict: Dict[str, Dict[Path, str]]) -> Dict[str, Dict[Path, str]]:
    paths: Set[Path] = set()
    for pathset in initial_dict.values():
        paths.update(pathset.keys())
    contents = {path: read_file(path) for path in paths}
    return {
        tag: {path: contents[path] for path in paths.keys()} for tag, paths in initial_dict.items()
    }


@dataclass
class PassRun:
    name: str
    result: PassResult
    time: float


async def run_passes(files: Dict[str, Dict[Path, str]], args: Dict[str, str], config: dict):
    async def pass_wrapper(pass_: Pass, *args) -> PassRun:
        start = time.time()
        result = await pass_.run(*args)
        return PassRun(pass_.name, result, time.time() - start)

    def combine_passes(*passes: Pass):
        async def combined_pass(*args):
            return [await pass_wrapper(pass_, *args) for pass_ in passes]

        return combined_pass

    write_passes = [deserialize_config_item(pass_) for pass_ in config["write_passes"]]
    read_passes = [deserialize_config_item(pass_) for pass_ in config["read_passes"]]
    combined_rw_passes = [
        combine_passes(*passes) for _, passes in groupby(write_passes, lambda p: p.tag)
    ]

    runs: List[PassRun] = []
    start = time.time()
    rw_pass_runs = [p(files, args) for p in combined_rw_passes]
    for pass_group in await asyncio.gather(*rw_pass_runs):
        runs.extend(pass_group)
    files = await inflate_files(files)
    ro_pass_runs = [pass_wrapper(p, files, args) for p in read_passes]
    runs.extend(await asyncio.gather(*ro_pass_runs))

    for run in runs:
        if not run.result.verify():
            log(f"Pass {colored(run.name, 'blue')} failed")
            if run.result.output.strip() != "":
                log(indent(run.result.output.strip(), "  "))

    if args["print_timings"]:
        log(f"{colored('Timing Information:', 'blue')}")
        max_len = max(len(run.name) for run in runs) + 2
        for run in runs:
            log(f"{f'{run.name}:' : <{max_len}}{run.time:.4f}s")
        log("---")
        log(f"Total Time: {time.time() - start:.4f}s")

    sys.exit(0 if all(r.result.verify() for r in runs) else 1)


def parse_args():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawTextHelpFormatter,
        description="""Checks that the files specified in FILES... respect the rev.ng coding
conventions, and prints all the violations.

If no FILES are specified then all the files tracked by git will be checked.""",
        epilog="""RETURN VALUES
On success exit code is 0.
On failure, i.e. if there is at least one file that is not respecting the
coding conventions, exit code is 1.""",
    )
    parser.add_argument(
        "--force-format",
        action="store_true",
        help="""Run linters on the specified files before checking.
WARNING: using this option will overwrite your files, make sure to backup
important stuff.""",
    )

    parser.add_argument(
        "--print-timings",
        action="store_true",
        help="After execution, print timing information for each pass",
    )

    parser.add_argument(
        "--use-local-clang-format-file",
        action="store_true",
        help="""Instead of using the repository config, let clang-format search in its default
locations for a config file.
See clang-format documentation for more details.""",
    )

    parser.add_argument(
        "--print-clang-format-config",
        action="store_true",
        help="""Print the clang-format configuration to stdout and then exit, ignoring other
arguments except for --use-local-clang-format-file.""",
    )

    parser_file_selection = parser.add_mutually_exclusive_group()
    parser_file_selection.add_argument(
        "--commit-range", metavar="RANGE", help="Check files changed in the specified commit range"
    )
    parser_file_selection.add_argument(
        "--HEAD", action="store_true", help="Alias for '--commit-range HEAD^..HEAD'"
    )
    parser_file_selection.add_argument(
        "--staged", action="store_true", help="Only check staged files"
    )
    parser_file_selection.add_argument(
        "--unstaged", action="store_true", help="Only check unstaged files"
    )
    parser.add_argument(
        "FILES",
        nargs="*",
        help="List of files to check against the rev.ng coding conventions.",
    )

    return parser.parse_args()


async def amain():
    args = parse_args()
    clang_style = read_clang_style(args)
    file_list = get_file_list(args)
    files = {filename: read_file(filename, 512) for filename in file_list}
    config = read_rcc_config()
    filters = {k: deserialize_config_item(v) for k, v in config["matchers"].items()}
    filtered_files = filter_files(files, filters)
    new_args = {
        "dry_run": not args.force_format,
        "print_timings": args.print_timings,
        "clang_style": clang_style,
        "isort_config": str(get_config("isort.cfg")),
        "flake8_config": str(get_config("flake8-config")),
        "prettier_config": str(get_config("prettierrc.yml")),
    }
    await run_passes(filtered_files, new_args, config)


if __name__ == "__main__":
    asyncio.run(amain())