1#!/usr/bin/env python3
2import argparse
3import json
4import os
5import re
6from typing import Dict, List, Optional, Set, TypedDict
7
8import yaml
9from sh import git  # type: ignore
10from yaml.loader import SafeLoader
11
12# A list of Gerrit users (email addresses).
13#   Some OWNERS files have empty lists for 'owners' or 'reviewers', which
14#   results in a None type for the value.
15UsersList = Optional[List[str]]
16
17
18# A YAML node with an extra line number.
19class NumberedNode(TypedDict):
20    line_number: int
21
22
23class MatchEntry(TypedDict, total=False):
24    suffix: str
25    regex: str
26    partial_regex: str
27    exact: str
28    owners: UsersList
29    reviewers: UsersList
30
31
32# The root YAML node of an OWNERS file
33class OwnersData(NumberedNode, TypedDict, total=False):
34    owners: UsersList
35    reviewers: UsersList
36    matchers: List[MatchEntry]
37
38
39# A YAML loader that adds the start line number onto each node (for
40# later linting support)
41class YamlLoader(SafeLoader):
42    def construct_mapping(
43        self, node: yaml.nodes.Node, deep: bool = False
44    ) -> NumberedNode:
45        mapping: NumberedNode = super(YamlLoader, self).construct_mapping(
46            node, deep=deep
47        )  # type: ignore
48        mapping["line_number"] = node.start_mark.line + 1
49        return mapping
50
51    # Load a file and return the OwnersData.
52    @staticmethod
53    def load(file: str) -> OwnersData:
54        data: OwnersData
55        with open(file, "r") as f:
56            data = yaml.load(f, Loader=YamlLoader)
57        return data
58
59
60# Class to match commit information with OWNERS files.
61class CommitMatch:
62    def __init__(
63        self, args: argparse.Namespace, owners: Dict[str, OwnersData]
64    ):
65        files: Set[str] = set(
66            git.bake("-C", args.path)
67            .show(args.commit, pretty="", name_only=True, _tty_out=False)
68            .splitlines()
69        )
70
71        self.owners: Set[str] = set()
72        self.reviewers: Set[str] = set()
73
74        for f in files:
75            path = f
76
77            while True:
78                path = os.path.dirname(path)
79
80                if path not in owners:
81                    if not path:
82                        break
83                    continue
84
85                local_owners = owners[path]
86
87                self.owners = self.owners.union(
88                    local_owners.get("owners") or []
89                )
90                self.reviewers = self.reviewers.union(
91                    local_owners.get("reviewers") or []
92                )
93
94                rel_file = os.path.relpath(f, path)
95
96                for e in local_owners.get("matchers", None) or []:
97                    if "exact" in e:
98                        self.__exact(rel_file, e)
99                    elif "partial_regex" in e:
100                        self.__partial_regex(rel_file, e)
101                    elif "regex" in e:
102                        self.__regex(rel_file, e)
103                    elif "suffix" in e:
104                        self.__suffix(rel_file, e)
105
106                if not path:
107                    break
108
109        self.reviewers = self.reviewers.difference(self.owners)
110
111    def __add_entry(self, entry: MatchEntry) -> None:
112        self.owners = self.owners.union(entry.get("owners") or [])
113        self.reviewers = self.reviewers.union(entry.get("reviewers") or [])
114
115    def __exact(self, file: str, entry: MatchEntry) -> None:
116        if file == entry["exact"]:
117            self.__add_entry(entry)
118
119    def __partial_regex(self, file: str, entry: MatchEntry) -> None:
120        if re.search(entry["partial_regex"], file):
121            self.__add_entry(entry)
122
123    def __regex(self, file: str, entry: MatchEntry) -> None:
124        if re.fullmatch(entry["regex"], file):
125            self.__add_entry(entry)
126
127    def __suffix(self, file: str, entry: MatchEntry) -> None:
128        if os.path.splitext(file)[1] == entry["suffix"]:
129            self.__add_entry(entry)
130
131
132# The subcommand to get the reviewers.
133def subcmd_reviewers(
134    args: argparse.Namespace, data: Dict[str, OwnersData]
135) -> None:
136    matcher = CommitMatch(args, data)
137
138    # Print in `git push refs/for/branch%<reviewers>` format.
139    if args.push_args:
140        result = []
141        for o in sorted(matcher.owners):
142            # Gerrit uses 'r' for the required reviewers (owners).
143            result.append(f"r={o}")
144        for r in sorted(matcher.reviewers):
145            # Gerrit uses 'cc' for the optional reviewers.
146            result.append(f"cc={r}")
147        print(",".join(result))
148    # Print as Gerrit Add Reviewers POST format.
149    # https://gerrit.openbmc.org/Documentation/rest-api-changes.html#add-reviewer
150    else:
151
152        def review_js(reviewer: str, state: str) -> str:
153            return json.dumps(
154                {
155                    "reviewer": reviewer,
156                    "state": state,
157                    "notify": "NONE",
158                    "notify_details": {"TO": {"accounts": [reviewer]}},
159                }
160            )
161
162        for o in sorted(matcher.owners):
163            print(review_js(o, "REVIEWER"))
164        for r in sorted(matcher.reviewers):
165            print(review_js(r, "CC"))
166
167
168def main() -> None:
169    parser = argparse.ArgumentParser()
170    parser.add_argument(
171        "-p", "--path", default=".", help="Root path to analyse"
172    )
173    subparsers = parser.add_subparsers()
174
175    parser_reviewers = subparsers.add_parser(
176        "reviewers", help="Generate List of Reviewers"
177    )
178    parser_reviewers.add_argument(
179        "--push-args",
180        default=False,
181        action="store_true",
182        help="Format as git push options",
183    )
184    parser_reviewers.add_argument(
185        "--commit",
186        default="HEAD",
187        help="Commit(s) to match against",
188    )
189    parser_reviewers.set_defaults(func=subcmd_reviewers)
190
191    args = parser.parse_args()
192
193    owners_files = git.bake("-C", args.path)(
194        "ls-files", "OWNERS", "**/OWNERS"
195    ).splitlines()
196
197    files = {}
198    for f in owners_files:
199        file = YamlLoader.load(os.path.join(args.path, f))
200        dirpath = os.path.dirname(f)
201        files[dirpath] = file
202
203    args.func(args, files)
204
205
206if __name__ == "__main__":
207    main()
208