recipes/test_owners.resources/find_owners.py

#!/usr/bin/env python3
# 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 functools
import json
import os
import re
import sys
import tempfile
import unittest


# 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+)$")

# METADATA.textproto lines that match this regex indicate Issue Tracker bug
# components. Note that this definition is too tight: a textproto can split the
# component and its definition across multiple lines.
# TODO(olivernewman): Use the actual proto schema to parse METADATA.textproto
# files instead of using regexes.
METADATA_COMPONENT_ID_REGEX = re.compile(
    r".*component_id:\s*(?P<component_id>\d+).*",
    flags=re.MULTILINE | re.DOTALL | re.IGNORECASE,
)


def main():
    args = sys.argv[1:]
    if len(args) != 2:
        print(f"Usage: {sys.argv[0]} <input path> <output path>")
        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, component_id = find_test_owners(checkout_dir, gn_label)
        result.append(
            {
                "test_name": test_name,
                "owners": owners,
                "issue_tracker_component_id": component_id,
                "gn_label": gn_label,
            }
        )

    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 list(map(dict, s))


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

    owners = []
    component_id = None
    current_dir = os.path.join(checkout_dir, *rel_test_dir.split("/"), "BUILD.gn")

    # 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
    # for large-scale change reviews and aren't responsible for arbitrary tests.
    while (current_dir := os.path.dirname(current_dir)) != checkout_dir and not owners:
        # If no component is defined, look for the component in the METADATA.textproto.
        # Only one component is allowed in metadata files.
        # If no component is found, fall back to OWNERS later.
        metadata_file = os.path.join(current_dir, "METADATA.textproto")
        if os.path.exists(metadata_file) and not component_id:
            component_id = load_metadata(metadata_file)
        owners_file = os.path.join(current_dir, "OWNERS")
        if os.path.exists(owners_file):
            owners = load_owners(owners_file)

    return owners, component_id


@functools.lru_cache(maxsize=None)
def load_owners(owners_file):
    """Given an OWNERS file, return a list of owner emails.

    Ignores any per-file owners, only returning owner emails that are on
    their own lines.
    """
    owners = []
    with open(owners_file) as f:
        for line in f.readlines():
            line = line.strip()
            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)
                try:
                    included_owners = load_owners(path)
                except FileNotFoundError:
                    # Ignore include statements that reference nonexistent
                    # files. Ideally such invalid includes would be prevented by
                    # presubmit checks, but until that's possible the
                    # test-owners builder shouldn't fall over when such a
                    # breakage is introduced.
                    pass
                else:
                    owners.extend(included_owners)
            if OWNER_REGEX.match(line):
                owners.append(line)
    return owners


def load_metadata(metadata_file):
    """Given a METADATA.textproto file, return a bug component ID."""
    with open(metadata_file) as f:  # pylint: disable=unspecified-encoding
        line = f.read().strip()
        if match := METADATA_COMPONENT_ID_REGEX.match(line):
            return int(match.group("component_id"))
    return None


# Run tests with recipes/test_owners.resources/run_resource_tests.sh
class TestLoadOwners(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory("find_owners_tests")
        os.chdir(self.tempdir.name)

    def tearDown(self):
        self.tempdir.cleanup()

    def _write_file(self, path, lines):
        os.makedirs(os.path.dirname(path), exist_ok=True)
        with open(path, "w") as f:
            f.write("\n".join(lines))

    def test_basic_file(self):
        self._write_file(
            os.path.join(self.tempdir.name, "foo", "OWNERS"),
            [
                "foo@example.com",
                "",
                "# This is a comment",
                "bar@example.com",
            ],
        )
        owners, component_id = find_test_owners(self.tempdir.name, "//foo:foo")
        self.assertEqual(owners, ["foo@example.com", "bar@example.com"])
        self.assertEqual(component_id, None)

    def test_include(self):
        self._write_file(
            os.path.join(self.tempdir.name, "foo", "OWNERS"),
            [
                "include /foo/bar/OWNERS",
                "bar@example.com",
            ],
        )
        self._write_file(
            os.path.join(self.tempdir.name, "foo", "bar", "OWNERS"),
            [
                "included@example.com",
            ],
        )
        owners, component_id = find_test_owners(self.tempdir.name, "//foo:foo")
        self.assertEqual(owners, ["included@example.com", "bar@example.com"])
        self.assertEqual(component_id, None)

    def test_invalid_include(self):
        self._write_file(
            os.path.join(self.tempdir.name, "foo", "OWNERS"),
            [
                "include /does-not-exist/OWNERS",
                "bar@example.com",
            ],
        )
        owners, component_id = find_test_owners(self.tempdir.name, "//foo:foo")
        self.assertEqual(owners, ["bar@example.com"])
        self.assertEqual(component_id, None)

    def test_owners_alongside_build_file(self):
        gn_label = "//foo/bar($toolchain)"
        self._write_file(
            # OWNERS file is in the same directory as the BUILD.gn file that
            # declared the test.
            os.path.join(self.tempdir.name, "foo", "bar", "OWNERS"),
            ["foo@example.com"],
        )
        owners, component_id = find_test_owners(self.tempdir.name, gn_label)
        self.assertEqual(owners, ["foo@example.com"])
        self.assertEqual(component_id, None)

    def test_metadata_file_in_parent_dir(self):
        gn_label = "//foo/bar:bar($toolchain)"
        self._write_file(
            os.path.join(self.tempdir.name, "foo", "OWNERS"),
            ["foo@example.com"],
        )
        self._write_file(
            os.path.join(self.tempdir.name, "foo", "METADATA.textproto"),
            # Intentionally use wacky whitespace.
            [
                """
                trackers: {
                    issue_tracker: {
                        component_id:
                          12345
                    }
                }
"""
            ],
        )
        owners, component_id = find_test_owners(self.tempdir.name, gn_label)
        self.assertEqual(owners, ["foo@example.com"])
        self.assertEqual(component_id, 12345)


if __name__ == "__main__":
    main()