watchlists.py

#!/usr/bin/env python3
# Copyright (c) 2011 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Watchlists

Watchlists is a mechanism that allow a developer (a "watcher") to watch over
portions of code that they are interested in. A "watcher" will be cc-ed to
changes that modify that portion of code, thereby giving them an opportunity
to make comments on chromium-review.googlesource.com even before the change is
committed.
Refer:
https://chromium.googlesource.com/chromium/src/+/HEAD/docs/infra/watchlists.md

When invoked directly from the base of a repository, this script lists out
the watchers for files given on the command line. This is useful to verify
changes to WATCHLISTS files.
"""

from __future__ import print_function

import logging
import os
import re
import sys


class Watchlists(object):
    """Manage Watchlists.

  This class provides mechanism to load watchlists for a repo and identify
  watchers.
  Usage:
    wl = Watchlists("/path/to/repo/root")
    watchers = wl.GetWatchersForPaths(["/path/to/file1",
                                       "/path/to/file2",])
  """

    _RULES = "WATCHLISTS"
    _RULES_FILENAME = _RULES
    _repo_root = None
    _defns = {}  # Definitions
    _path_regexps = {}  # Name -> Regular expression mapping
    _watchlists = {}  # name to email mapping

    def __init__(self, repo_root):
        self._repo_root = repo_root
        self._LoadWatchlistRules()

    def _GetRulesFilePath(self):
        """Returns path to WATCHLISTS file."""
        return os.path.join(self._repo_root, self._RULES_FILENAME)

    def _HasWatchlistsFile(self):
        """Determine if watchlists are available for this repo."""
        return os.path.exists(self._GetRulesFilePath())

    def _ContentsOfWatchlistsFile(self):
        """Read the WATCHLISTS file and return its contents."""
        try:
            watchlists_file = open(self._GetRulesFilePath())
            contents = watchlists_file.read()
            watchlists_file.close()
            return contents
        except IOError as e:
            logging.error("Cannot read %s: %s" % (self._GetRulesFilePath(), e))
            return ''

    def _LoadWatchlistRules(self):
        """Load watchlists from WATCHLISTS file. Does nothing if not present."""
        if not self._HasWatchlistsFile():
            return

        contents = self._ContentsOfWatchlistsFile()
        watchlists_data = None
        try:
            watchlists_data = eval(contents, {'__builtins__': None}, None)
        except SyntaxError as e:
            logging.error("Cannot parse %s. %s" % (self._GetRulesFilePath(), e))
            return

        defns = watchlists_data.get("WATCHLIST_DEFINITIONS")
        if not defns:
            logging.error("WATCHLIST_DEFINITIONS not defined in %s" %
                          self._GetRulesFilePath())
            return
        watchlists = watchlists_data.get("WATCHLISTS")
        if not watchlists:
            logging.error("WATCHLISTS not defined in %s" %
                          self._GetRulesFilePath())
            return
        self._defns = defns
        self._watchlists = watchlists

        # Compile the regular expressions ahead of time to avoid creating them
        # on-the-fly multiple times per file.
        self._path_regexps = {}
        for name, rule in defns.items():
            filepath = rule.get('filepath')
            if not filepath:
                continue
            self._path_regexps[name] = re.compile(filepath)

        # Verify that all watchlist names are defined
        for name in watchlists:
            if name not in defns:
                logging.error("%s not defined in %s" %
                              (name, self._GetRulesFilePath()))

    def GetWatchersForPaths(self, paths):
        """Fetch the list of watchers for |paths|

    Args:
      paths: [path1, path2, ...]

    Returns:
      [u1@chromium.org, u2@gmail.com, ...]
    """
        watchers = set()  # A set, to avoid duplicates
        for path in paths:
            path = path.replace(os.sep, '/')
            for name, rule in self._path_regexps.items():
                if name not in self._watchlists:
                    continue
                if rule.search(path):
                    for watchlist in self._watchlists[name]:
                        watchers.add(watchlist)
        return sorted(watchers)


def main(argv):
    # Confirm that watchlists can be parsed and spew out the watchers
    if len(argv) < 2:
        print("Usage (from the base of repo):")
        print("  %s [file-1] [file-2] ...." % argv[0])
        return 1
    wl = Watchlists(os.getcwd())
    watchers = wl.GetWatchersForPaths(argv[1:])
    print(watchers)


if __name__ == '__main__':
    main(sys.argv)