recipes/test_owners.resources/find_owners.py

#!/usr/bin/env python
# Copyright 2020 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Find test owners.

This script assumes that it's run in the root directory of a Fuchsia
checkout.
"""

import json
import os
import re
import sys


# OWNERS file lines that match this regex (a very crude email matcher) will be
# considered to represent owners.
OWNER_REGEX = re.compile(r"^\S+@\S+$")

# An OWNERS file can import another OWNERS file using a line of the form
# `include /path/to/other/OWNERS`.
INCLUDE_REGEX = re.compile(r"^include (\S+)$")

# OWNERS file lines that match this regex indicate Monorail bug components.
BUG_COMPONENT_REGEX = re.compile(r"^# COMPONENT: (\S+)$")


def main():
    args = sys.argv[1:]
    if len(args) != 2:
        print "Usage: %s <input path> <output path>" % sys.argv[0]
        sys.exit(1)

    input_path, output_path = args[0], args[1]
    output = find_owners(input_path)
    output.sort(key=lambda t: t["test_name"])
    with open(output_path, "w") as f:
        json.dump(output, f, indent=1, separators=(",", ": "))


def find_owners(test_manifest_path):
    checkout_dir = os.getcwd()
    with open(test_manifest_path) as f:
        test_manifest = json.load(f)

    result = []

    for test in unique_dicts(test_manifest):
        test_name, gn_label = test["test_name"], test["gn_label"]
        if not gn_label:
            continue
        owners, bug_components = find_test_owners(checkout_dir, gn_label)
        result.append(
            {
                "test_name": test_name,
                "owners": owners,
                "bug_components": bug_components,
            }
        )

    return result


def unique_dicts(tests):
    """Ensure that every entry in the test manifest is unique.

    This should theoretically always be the case, but bugs in various parts
    of the Fuchsia test database system might cause dupes, in which case we
    don't want them to clutter up the output file.
    """
    s = set(tuple(t.items()) for t in tests)
    return map(dict, s)


def find_test_owners(checkout_dir, gn_label):
    """Find a test's owners and bug components given its GN label."""
    # Strip off toolchain, target name, and leading slashes.
    rel_test_dir = gn_label.split(":")[0].strip("/")

    owners = []
    bug_components = []
    next_dir_to_check = os.path.join(checkout_dir, *rel_test_dir.split("/"))
    while True:
        owners_file = os.path.join(next_dir_to_check, "OWNERS")
        next_dir_to_check = os.path.dirname(next_dir_to_check)
        if not os.path.exists(owners_file):
            # Give up if we reach the checkout root before finding an OWNERS
            # file. We want to avoid falling back to global owners since global
            # owners are basically just for large-scale change reviews and
            # aren't responsible for arbitrary tests.
            if next_dir_to_check == checkout_dir:
                break
            continue
        owners, new_bug_components = parse_owners(owners_file)
        if not bug_components:
            bug_components = new_bug_components
        # If we find an OWNERS file with only bug components and no owners,
        # keep track of the bug components we found and keep searching for an
        # OWNERS file that actually contains owners.
        if not owners:
            continue
        break

    return owners, bug_components


# Used for memoizing OWNERS file reads.
_OWNERS_CACHE = {}


def parse_owners(owners_file):
    """Given an OWNERS file, return a list of owners and bug components.

    Ignores any per-file owners, only returning owner emails that are on
    their own lines.
    """
    if owners_file not in _OWNERS_CACHE:
        result = _parse_owners_once(owners_file)
        _OWNERS_CACHE[owners_file] = result
    return _OWNERS_CACHE[owners_file]


def _parse_owners_once(owners_file):
    owners = []
    bug_components = []
    with open(owners_file) as f:
        for line in f.readlines():
            line = line.strip()
            component_match = BUG_COMPONENT_REGEX.match(line)
            if component_match:
                bug_components.append(component_match.group(1))
                continue
            include_match = INCLUDE_REGEX.match(line)
            if include_match:
                included_file_parts = include_match.group(1).lstrip("/").split("/")
                # Included paths can be relative.
                if included_file_parts[0].startswith("."):
                    path = os.path.abspath(
                        os.path.join(os.path.dirname(owners_file), *included_file_parts)
                    )
                else:
                    path = os.path.join(os.getcwd(), *included_file_parts)
                included_owners, _ = parse_owners(path)
                owners.extend(included_owners)
            if OWNER_REGEX.match(line):
                owners.append(line)
    return owners, bug_components


if __name__ == "__main__":
    main()