grpc/tools/mkowners/mkowners.py

#!/usr/bin/env python3
# Copyright 2017 gRPC authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import argparse
import collections
import operator
import os
import re
import subprocess

#
# Find the root of the git tree
#

git_root = (
    subprocess.check_output(["git", "rev-parse", "--show-toplevel"])
    .decode("utf-8")
    .strip()
)

#
# Parse command line arguments
#

default_out = os.path.join(git_root, ".github", "CODEOWNERS")

argp = argparse.ArgumentParser("Generate .github/CODEOWNERS file")
argp.add_argument(
    "--out",
    "-o",
    type=str,
    default=default_out,
    help="Output file (default %s)" % default_out,
)
args = argp.parse_args()

#
# Walk git tree to locate all OWNERS files
#

owners_files = [
    os.path.join(root, "OWNERS")
    for root, dirs, files in os.walk(git_root)
    if "OWNERS" in files
]

#
# Parse owners files
#

Owners = collections.namedtuple("Owners", "parent directives dir")
Directive = collections.namedtuple("Directive", "who globs")


def parse_owners(filename):
    with open(filename) as f:
        src = f.read().splitlines()
    parent = True
    directives = []
    for line in src:
        line = line.strip()
        # line := directive | comment
        if not line:
            continue
        if line[0] == "#":
            continue
        # it's a directive
        directive = None
        if line == "set noparent":
            parent = False
        elif line == "*":
            directive = Directive(who="*", globs=[])
        elif " " in line:
            (who, globs) = line.split(" ", 1)
            globs_list = [glob for glob in globs.split(" ") if glob]
            directive = Directive(who=who, globs=globs_list)
        else:
            directive = Directive(who=line, globs=[])
        if directive:
            directives.append(directive)
    return Owners(
        parent=parent,
        directives=directives,
        dir=os.path.relpath(os.path.dirname(filename), git_root),
    )


owners_data = sorted(
    [parse_owners(filename) for filename in owners_files],
    key=operator.attrgetter("dir"),
)

#
# Modify owners so that parented OWNERS files point to the actual
# Owners tuple with their parent field
#

new_owners_data = []
for owners in owners_data:
    if owners.parent == True:
        best_parent = None
        best_parent_score = None
        for possible_parent in owners_data:
            if possible_parent is owners:
                continue
            rel = os.path.relpath(owners.dir, possible_parent.dir)
            # '..' ==> we had to walk up from possible_parent to get to owners
            #      ==> not a parent
            if ".." in rel:
                continue
            depth = len(rel.split(os.sep))
            if not best_parent or depth < best_parent_score:
                best_parent = possible_parent
                best_parent_score = depth
        if best_parent:
            owners = owners._replace(parent=best_parent.dir)
        else:
            owners = owners._replace(parent=None)
    new_owners_data.append(owners)
owners_data = new_owners_data

#
# In bottom to top order, process owners data structures to build up
# a CODEOWNERS file for GitHub
#


def full_dir(rules_dir, sub_path):
    return os.path.join(rules_dir, sub_path) if rules_dir != "." else sub_path


# glob using git
gg_cache = {}


def git_glob(glob):
    global gg_cache
    if glob in gg_cache:
        return gg_cache[glob]
    r = set(
        subprocess.check_output(
            ["git", "ls-files", os.path.join(git_root, glob)]
        )
        .decode("utf-8")
        .strip()
        .splitlines()
    )
    gg_cache[glob] = r
    return r


def expand_directives(root, directives):
    globs = collections.OrderedDict()
    # build a table of glob --> owners
    for directive in directives:
        for glob in directive.globs or ["**"]:
            if glob not in globs:
                globs[glob] = []
            if directive.who not in globs[glob]:
                globs[glob].append(directive.who)
    # expand owners for intersecting globs
    sorted_globs = sorted(
        list(globs.keys()),
        key=lambda g: len(git_glob(full_dir(root, g))),
        reverse=True,
    )
    out_globs = collections.OrderedDict()
    for glob_add in sorted_globs:
        who_add = globs[glob_add]
        pre_items = [i for i in list(out_globs.items())]
        out_globs[glob_add] = who_add.copy()
        for glob_have, who_have in pre_items:
            files_add = git_glob(full_dir(root, glob_add))
            files_have = git_glob(full_dir(root, glob_have))
            intersect = files_have.intersection(files_add)
            if intersect:
                for f in sorted(files_add):  # sorted to ensure merge stability
                    if f not in intersect:
                        out_globs[os.path.relpath(f, start=root)] = who_add
                for who in who_have:
                    if who not in out_globs[glob_add]:
                        out_globs[glob_add].append(who)
    return out_globs


def add_parent_to_globs(parent, globs, globs_dir):
    if not parent:
        return
    for owners in owners_data:
        if owners.dir == parent:
            owners_globs = expand_directives(owners.dir, owners.directives)
            for oglob, oglob_who in list(owners_globs.items()):
                for gglob, gglob_who in list(globs.items()):
                    files_parent = git_glob(full_dir(owners.dir, oglob))
                    files_child = git_glob(full_dir(globs_dir, gglob))
                    intersect = files_parent.intersection(files_child)
                    gglob_who_orig = gglob_who.copy()
                    if intersect:
                        for f in sorted(
                            files_child
                        ):  # sorted to ensure merge stability
                            if f not in intersect:
                                who = gglob_who_orig.copy()
                                globs[os.path.relpath(f, start=globs_dir)] = who
                        for who in oglob_who:
                            if who not in gglob_who:
                                gglob_who.append(who)
            add_parent_to_globs(owners.parent, globs, globs_dir)
            return
    assert False


todo = owners_data.copy()
done = set()
with open(args.out, "w") as out:
    out.write("# Auto-generated by the tools/mkowners/mkowners.py tool\n")
    out.write("# Uses OWNERS files in different modules throughout the\n")
    out.write("# repository as the source of truth for module ownership.\n")
    written_globs = []
    while todo:
        head, *todo = todo
        if head.parent and not head.parent in done:
            todo.append(head)
            continue
        globs = expand_directives(head.dir, head.directives)
        add_parent_to_globs(head.parent, globs, head.dir)
        for glob, owners in list(globs.items()):
            skip = False
            for glob1, owners1, dir1 in reversed(written_globs):
                files = git_glob(full_dir(head.dir, glob))
                files1 = git_glob(full_dir(dir1, glob1))
                intersect = files.intersection(files1)
                if files == intersect:
                    if sorted(owners) == sorted(owners1):
                        skip = True  # nothing new in this rule
                        break
                elif intersect:
                    # continuing would cause a semantic change since some files are
                    # affected differently by this rule and CODEOWNERS is order dependent
                    break
            if not skip:
                out.write(
                    "/%s %s\n" % (full_dir(head.dir, glob), " ".join(owners))
                )
                written_globs.append((glob, owners, head.dir))
        done.add(head.dir)