implement per-file OWNERS support

owners.py

 # Copyright (c) 2012 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.
 
 """A database of OWNERS files."""

[M-A Ruel, 2012/10/12 15:19:04]

One thing I'd really like is a formal description of the DSL 
(domain-specific-language) here.

Explain the various statements allowed in the file format and their exact
effects.

Side-effect by the placement of directories as they are found in the checkout up
to |root|.

[Dirk Pranke, 2012/10/16 22:42:41]

Okay, I'll write up something and see what you think.

 
 import collections
 import re
 
 
 # If this is present by itself on a line, this means that everyone can review.
 EVERYONE = '*'
 
 
 # Recognizes 'X@Y' email addresses. Very simplistic.
@@ skipping 13 matching lines @@
   def __init__(self, path, lineno, msg):
     super(SyntaxErrorInOwnersFile, self).__init__((path, lineno, msg))
     self.path = path
     self.lineno = lineno
     self.msg = msg
 
   def __str__(self):
     return "%s:%d syntax error: %s" % (self.path, self.lineno, self.msg)
 
 
 class Database(object):

[M-A Ruel, 2012/10/12 15:19:04]

I think the code would be easier to read if it was split functionally into:
- One Owners class presenting a parsed set of OWNERS files in a tree.
- One class deciding based on Owners() instance if the list of files along the
list of reviewers, should be accepted or not.

Technically, the second is a function*, here it's a member function, but I think
it'd be a bit simpler to read. Overall, it's a large refactor so I'm not asking
you to do it, just sayin'.

* But the "function" becomes more complex and stateful now that a glob is
required.

[Dirk Pranke, 2012/10/16 22:42:41]

I'll think about this, but definitely not do it in this patch.

   """A database of OWNERS files for a repository.
 
   This class allows you to find a suggested set of reviewers for a list
   of changed files, and see if a list of changed files is covered by a
   list of reviewers."""
 
-  def __init__(self, root, fopen, os_path):
+  def __init__(self, root, fopen, os_path, glob):
     """Args:
       root: the path to the root of the Repository
       open: function callback to open a text file for reading
       os_path: module/object callback with fields for 'abspath', 'dirname',
           'exists', and 'join'
+      glob: function callback to list entries in a directory match a glob (i.e., glob.glob)

[M-A Ruel, 2012/10/12 15:19:04]

80 cols

[Dirk Pranke, 2012/10/16 22:42:41]

Done.

     """
     self.root = root
     self.fopen = fopen
     self.os_path = os_path
+    self.glob = glob
 
     # Pick a default email regexp to use; callers can override as desired.
     self.email_regexp = re.compile(BASIC_EMAIL_REGEXP)
 
     # Mapping of owners to the paths they own.
     self.owned_by = {EVERYONE: set()}
 
     # Mapping of paths to authorized owners.
     self.owners_for = {}
 
     # Set of paths that stop us from looking above them for owners.
     # (This is implicitly true for the root directory).
     self.stop_looking = set([''])
 
   def reviewers_for(self, files):
     """Returns a suggested set of reviewers that will cover the files.
 
     files is a sequence of paths relative to (and under) self.root."""
     self._check_paths(files)
     self._load_data_needed_for(files)
     return self._covering_set_of_owners_for(files)
 
+  # TODO(dpranke): rename to objects_not_covered_by
   def directories_not_covered_by(self, files, reviewers):
     """Returns the set of directories that are not owned by a reviewer.
 
     Determines which of the given files are not owned by at least one of the
     reviewers, then returns a set containing the applicable enclosing
     directories, i.e. the ones upward from the files that have OWNERS files.
 
     Args:
         files is a sequence of paths relative to (and under) self.root.
         reviewers is a sequence of strings matching self.email_regexp.
     """
     self._check_paths(files)
     self._check_reviewers(reviewers)
     self._load_data_needed_for(files)
 
-    dirs = set([self.os_path.dirname(f) for f in files])
-    covered_dirs = self._dirs_covered_by(reviewers)
-    uncovered_dirs = [self._enclosing_dir_with_owners(d) for d in dirs
-                      if not self._is_dir_covered_by(d, covered_dirs)]
+    objs = set()
+    for f in files:
+      if f in self.owners_for:
+        objs.add(f)
+      else:
+        objs.add(self.os_path.dirname(f))
 
-    return set(uncovered_dirs)
+    covered_objs = self._objs_covered_by(reviewers)
+    uncovered_objs = [self._enclosing_obj_with_owners(o) for o in objs
+                      if not self._is_obj_covered_by(o, covered_objs)]
+
+    return set(uncovered_objs)
+
+  objects_not_covered_by = directories_not_covered_by

[M-A Ruel, 2012/10/12 15:19:04]

Why not rename right away? I don't see any advantage in doing it in two steps
(?)

[Dirk Pranke, 2012/10/16 22:42:41]

I thought it would make the patch easier to review.

 
   def _check_paths(self, files):
     def _is_under(f, pfx):
       return self.os_path.abspath(self.os_path.join(pfx, f)).startswith(pfx)
     _assert_is_collection(files)
     assert all(_is_under(f, self.os_path.abspath(self.root)) for f in files)
 
   def _check_reviewers(self, reviewers):
     _assert_is_collection(reviewers)
     assert all(self.email_regexp.match(r) for r in reviewers)
 
+  # TODO(dpranke): Rename to _objs_covered_by and update_callers
   def _dirs_covered_by(self, reviewers):
     dirs = self.owned_by[EVERYONE]
     for r in reviewers:
       dirs = dirs | self.owned_by.get(r, set())
     return dirs
 
+  _objs_covered_by = _dirs_covered_by
+
   def _stop_looking(self, dirname):
     return dirname in self.stop_looking
 
+  # TODO(dpranke): Rename to _is_dir_covered_by and update callers.
   def _is_dir_covered_by(self, dirname, covered_dirs):
     while not dirname in covered_dirs and not self._stop_looking(dirname):
       dirname = self.os_path.dirname(dirname)
     return dirname in covered_dirs
 
+  _is_obj_covered_by = _is_dir_covered_by
+
+  # TODO(dpranke): Rename to _enclosing_obj_with_owners and update callers.
   def _enclosing_dir_with_owners(self, directory):
     """Returns the innermost enclosing directory that has an OWNERS file."""
     dirpath = directory
     while not dirpath in self.owners_for:
       if self._stop_looking(dirpath):
         break
       dirpath = self.os_path.dirname(dirpath)
     return dirpath
 
+  _enclosing_obj_with_owners = _enclosing_dir_with_owners
+
   def _load_data_needed_for(self, files):
     for f in files:
       dirpath = self.os_path.dirname(f)
       while not dirpath in self.owners_for:
         self._read_owners_in_dir(dirpath)
         if self._stop_looking(dirpath):
           break
         dirpath = self.os_path.dirname(dirpath)
 
   def _read_owners_in_dir(self, dirpath):
     owners_path = self.os_path.join(self.root, dirpath, 'OWNERS')
     if not self.os_path.exists(owners_path):
       return
 
     lineno = 0
     for line in self.fopen(owners_path):
       lineno += 1
       line = line.strip()
       if line.startswith('#') or line == '':
         continue
       if line == 'set noparent':
         self.stop_looking.add(dirpath)
         continue
+
+      m = re.match("per-file (.+)=(.+)", line)
+      if m:
+        glob_string = m.group(1)
+        directive = m.group(2)
+        full_glob_string = self.os_path.join(self.root, dirpath, glob_string)

[M-A Ruel, 2012/10/12 15:19:04]

You should assert not self.os_path.isabs(glob_string)

[Dirk Pranke, 2012/10/16 22:42:41]

Good point. Actually it's probably better to ensure that we don't have os.sep in
the string at all, to eliminate possible relpaths, too.

+        baselines = self.glob(full_glob_string)
+        for baseline in [self.os_path.relpath(self.root, b) for b in baselines]:

[M-A Ruel, 2012/10/12 15:19:04]

for baseline in (self.os_path.relpath(self.root, b) for b in baselines):

[Dirk Pranke, 2012/10/16 22:42:41]

I tend to prefer the comprehension notation to the generator notation except
when inside function calls, but ok.

+          self._add_entry(baseline, directive, "per-file line",
+                          owners_path, lineno)
+        continue
+
       if line.startswith('set '):
         raise SyntaxErrorInOwnersFile(owners_path, lineno,
             'unknown option: "%s"' % line[4:].strip())
-      if self.email_regexp.match(line) or line == EVERYONE:
-        self.owned_by.setdefault(line, set()).add(dirpath)
-        self.owners_for.setdefault(dirpath, set()).add(line)
-        continue
+
+      self._add_entry(dirpath, line, "line", owners_path, lineno)
+
+  def _add_entry(self, path, directive, line_type, owners_path, lineno):
+    if directive == "set noparent":
+      self.stop_looking.add(path)
+    elif self.email_regexp.match(directive) or directive == EVERYONE:
+      self.owned_by.setdefault(directive, set()).add(path)
+      self.owners_for.setdefault(path, set()).add(directive)
+    else:
       raise SyntaxErrorInOwnersFile(owners_path, lineno,
-            ('line is not a comment, a "set" directive, '
-             'or an email address: "%s"' % line))
+          ('%s is not a "set" directive, '

[M-A Ruel, 2012/10/12 15:19:04]

Technically, you'd also want to say it's not an EVERYONE(*) directive either.

[Dirk Pranke, 2012/10/16 22:42:41]

Done.

+           'or an email address: "%s"' % (line_type, directive)))
+
 
   def _covering_set_of_owners_for(self, files):
     # Get the set of directories from the files.
     dirs = set()
     for f in files:
       dirs.add(self.os_path.dirname(f))
 
     owned_dirs = {}
     dir_owners = {}
 
@@ skipping 36 matching lines @@
         break
 
       final_owners.add(max_owner)
 
       # Remove all directories owned by the current owner from the remaining
       # list.
       for dirname in owned_dirs[max_owner]:
         dirs.discard(dirname)
 
     return final_owners