An interactive tool to help find owners covering current change list.

owners_finder.py

+# Copyright (c) 2013 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.
+
+"""Helps picking owner_to_files for reviewing."""

[Dirk Pranke, 2013/07/27 00:06:03]

"""Interactive tool for finding reviewers/owners for a change.""" ?

Also, there should be some documentation for this somewhere; perhaps here, or in
the help text for CMDowners, or on dev.chromium.org or ...

It would be good to have both user documentation and some internal documentation
over the structure of the algorithm. The latter could be done here or in a
docstring for OwnersFinder.

[Bei Zhang, 2013/07/30 05:59:17]

I would like to put it on dev.chromium.org. How do I do that?

On 2013/07/27 00:06:03, Dirk Pranke wrote:

+
+import os.path
+import copy

[Dirk Pranke, 2013/07/27 00:06:03]

You should probably pass these references to the __init__() rather than
importing them here. That'll make things easier to mock-out and test.

[Bei Zhang, 2013/07/30 05:59:17]

Done.

+import owners as owners_module
+
+
+def first(iterable):
+  for element in iterable:
+    return element
+
+
+class OwnersFinder(object):
+  COLOR_LINK = '\033[4m'
+  COLOR_BOLD = '\033[1;32m'
+  COLOR_GREY = '\033[0;37m'
+  COLOR_RESET = '\033[0m'
+
+  indentation = 0
+
+  def __init__(self, files, local_root,
+               fopen, os_path, glob,
+               email_postfix='@chromium.org',
+               disable_color=False):
+    self.email_postfix = email_postfix
+
+    if os.name == 'nt' or disable_color:
+      self.COLOR_LINK = ''
+      self.COLOR_BOLD = ''
+      self.COLOR_GREY = ''
+      self.COLOR_RESET = ''
+
+    db = OwnersFinder._read_from_database(files, local_root, fopen, os_path,
+                                          glob)
+
+    self.file_to_owners = {}
+    self.owners_score = {}

[Dirk Pranke, 2013/07/27 00:06:03]

Add some comments to say what the owners_score and owners_queue for ?

[Bei Zhang, 2013/07/30 05:59:17]

Done.

+    self._map_files_to_owners(files, db)
+    self.owner_to_files = OwnersFinder._map_owners_to_files(self.file_to_owners,
+                                                            db)
+    self.original_files_to_owners = copy.deepcopy(self.file_to_owners)
+    self.comments = db.comments
+    self.owners_queue = []
+    self.unreviewed_files = set()
+    self.reviewed_by = {}
+    self.selected_owners = set()
+    self.deselected_owners = set()
+    self.reset()
+
+  @staticmethod
+  def _read_from_database(files, local_root, fopen, os_path, glob):
+    db = owners_module.Database(local_root, fopen, os_path, glob)
+    db.load_data_needed_for(files)
+    return db

[Dirk Pranke, 2013/07/27 00:06:03]

Is there a reason not to just inline this into __init__() ?

[Bei Zhang, 2013/07/30 05:59:17]

Done.

+
+  def _map_files_to_owners(self, files, db):

[Dirk Pranke, 2013/07/27 00:06:03]

You should add some comments about what this routine is doing ... what is the
owners_score data structure for? How is the logic here different from the logic
in owners.py's lowest_cost_owner() ? Can we re-use some of the code ?

[Bei Zhang, 2013/07/30 05:59:17]

I extracted the scoring routing. I think it is possible to reuse the scoring in
Database but they work somehow differently. So I think I probably will do that
in another CL.

On 2013/07/27 00:06:03, Dirk Pranke wrote:

+    for file_name in files:
+      base_score = 1.0
+      owners_set = set()
+      if file_name in db.owners_for:
+        for owner in db.owners_for[file_name]:
+          self.owners_score.setdefault(owner, {})
+          self.owners_score[owner].setdefault(file_name, 0)
+          self.owners_score[owner][file_name] += max(
+              self.owners_score[owner][file_name],
+              base_score / len(db.owners_for[file_name]))
+        owners_set = db.owners_for[file_name]
+      else:
+        dir_name = file_name
+        while dir_name != '':
+          if dir_name in db.stop_looking:
+            break
+          dir_name = os.path.dirname(dir_name)
+          if dir_name in db.owners_for:
+            owners_set = owners_set | db.owners_for[dir_name]
+            for owner in db.owners_for[dir_name]:
+              self.owners_score.setdefault(owner, {})
+              self.owners_score[owner].setdefault(dir_name, 0)
+              self.owners_score[owner][dir_name] = max(
+                  self.owners_score[owner][dir_name],
+                  base_score / len(db.owners_for[dir_name]))
+            # If the file can be reviewed by a nearer owner, set a lower
+            # priority to a owner in the parent folders.
+            if db.owners_for[dir_name]:
+              base_score *= 0.1
+            if owners_module.EVERYONE in owners_set:
+              break
+
+      if len(owners_set) == 0:
+        raise Exception('File "%s" has no owner' % file_name)
+
+      # Eliminate files that EVERYONE can review
+      if owners_module.EVERYONE in owners_set:
+        continue
+      self.file_to_owners[file_name] = owners_set
+
+    for owner in self.owners_score:
+      self.owners_score[owner] = sum(self.owners_score[owner].values())
+
+  @staticmethod
+  def _map_owners_to_files(files_to_owners, db):
+    owner_to_files = {}
+    for owner_name in db.owned_by:
+      if owner_name == owners_module.EVERYONE:
+        continue
+      files_set = set()
+      for file_name in files_to_owners:
+        if owner_name in files_to_owners[file_name]:
+          files_set.add(file_name)
+      if len(files_set) > 0:
+        owner_to_files[owner_name] = files_set
+    return owner_to_files
+
+  def bold(self, text):

[Dirk Pranke, 2013/07/27 00:06:03]

It seems like the only public method is really run(). I would perhaps
restructure this file so that run() follows init() and then the other methods
follow, top-down (or in some similar order) so that a new reader can get a sense
of how things work by just reading linearly through the file.

[Bei Zhang, 2013/07/30 05:59:17]

Done.

+    return self.COLOR_BOLD + text + self.COLOR_RESET
+
+  def bold_name(self, name):
+    return (self.COLOR_BOLD +
+            name.replace(self.email_postfix, '') + self.COLOR_RESET)
+
+  def greyed(self, text):
+    return self.COLOR_GREY + text + self.COLOR_RESET
+
+  def indent(self):
+    self.indentation += 1
+
+  def unindent(self):
+    self.indentation -= 1
+
+  def print_indent(self):
+    return '  ' * self.indentation
+
+  def writeln(self, text=''):
+    print self.print_indent() + text
+
+  def reset(self):
+    self.file_to_owners = copy.deepcopy(self.original_files_to_owners)
+    self.unreviewed_files = set(self.file_to_owners.keys())
+    self.reviewed_by = {}
+    self.selected_owners = set()
+    self.deselected_owners = set()
+
+    # Initialize owners queue, sort it by the number of files
+    # each owns
+    self.owners_queue = list(sorted(self.owner_to_files.keys(),
+                                    key=lambda owner: self.owners_score[owner],
+                                    reverse=True))
+    self.find_mandatory_owners()
+
+  def select_owner(self, owner, findMandatoryOwners=True):
+    if owner in self.selected_owners:
+      return
+    if owner in self.deselected_owners:
+      return
+    if not (owner in self.owners_queue):
+      return
+    self.writeln('Selected: ' + owner)
+    self.owners_queue.remove(owner)
+    self.selected_owners.add(owner)
+    for file_name in filter(
+        lambda file_name: file_name in self.unreviewed_files,
+        self.owner_to_files[owner]):
+      self.unreviewed_files.remove(file_name)
+      self.reviewed_by[file_name] = owner
+    if findMandatoryOwners:
+      self.find_mandatory_owners()
+
+  def deselect_owner(self, owner, findMandatoryOwners=True):
+    if owner in self.selected_owners:
+      return
+    if owner in self.deselected_owners:
+      return
+    if not (owner in self.owners_queue):
+      return
+    self.writeln('Deselected: ' + owner)
+    self.owners_queue.remove(owner)
+    self.deselected_owners.add(owner)
+    for file_name in self.owner_to_files[owner] & self.unreviewed_files:
+      self.file_to_owners[file_name].remove(owner)
+    if findMandatoryOwners:
+      self.find_mandatory_owners()
+
+  def find_mandatory_owners(self):
+    continues = True
+    for owner in self.owners_queue:
+      if owner in self.selected_owners:
+        continue
+      if owner in self.deselected_owners:
+        continue
+      if len(self.owner_to_files[owner] & self.unreviewed_files) == 0:
+        self.deselect_owner(owner, False)
+
+    while continues:
+      continues = False
+      for file_name in filter(
+          lambda file_name: len(self.file_to_owners[file_name]) == 1,
+          self.unreviewed_files):
+        owner = first(self.file_to_owners[file_name])
+        self.select_owner(owner, False)
+        continues = True
+        break
+
+  def print_comments(self, owner):
+    if owner not in self.comments:
+      self.writeln(self.bold_name(owner))
+    else:
+      self.writeln(self.bold_name(owner) + ' is commented as:')
+      self.indent()
+      for path in self.comments[owner]:
+        if len(self.comments[owner][path]) > 0:
+          self.writeln(self.greyed(self.comments[owner][path]) +
+                       ' (at ' + self.bold(path or '<root>') + ')')
+        else:
+          self.writeln(self.greyed('[No comment] ') + ' (at ' +
+                       self.bold(path or '<root>') + ')')
+      self.unindent()
+
+  def print_file_info(self, file_name, except_owner=''):
+    if file_name not in self.unreviewed_files:
+      self.writeln(self.greyed(file_name +
+                               ' (by ' +
+                               self.bold_name(self.reviewed_by[file_name]) +
+                               ')'))
+    else:
+      if len(self.file_to_owners[file_name]) <= 3:
+        other_owners = []
+        for ow in self.file_to_owners[file_name]:
+          if ow != except_owner:
+            other_owners.append(self.bold_name(ow))
+        self.writeln(file_name +
+                     ' [' + (', '.join(other_owners)) + ']')
+      else:
+        self.writeln(file_name + ' [' +
+                     self.bold(str(len(self.file_to_owners[file_name]))) +
+                     ']')
+
+  def print_file_info_detailed(self, file_name):
+    self.writeln(file_name)
+    self.indent()
+    for ow in sorted(self.file_to_owners[file_name]):
+      if ow in self.deselected_owners:
+        self.writeln(self.bold_name(self.greyed(ow)))
+      elif ow in self.selected_owners:
+        self.writeln(self.bold_name(self.greyed(ow)))
+      else:
+        self.writeln(self.bold_name(ow))
+    self.unindent()
+
+  def print_owned_files_for(self, owner):
+    # Print owned files
+    self.print_comments(owner)
+    self.writeln(self.bold_name(owner) + ' owns ' +
+                 str(len(self.owner_to_files[owner])) + ' file(s):')
+    self.indent()
+    for file_name in sorted(self.owner_to_files[owner]):
+      self.print_file_info(file_name, owner)
+    self.unindent()
+    self.writeln()
+
+  def list_owners(self, owners_queue):
+    if (len(self.owner_to_files) - len(self.deselected_owners) -
+            len(self.selected_owners)) > 3:
+      for ow in owners_queue:
+        if ow not in self.deselected_owners and ow not in self.selected_owners:
+          self.print_comments(ow)
+    else:
+      for ow in owners_queue:
+        if ow not in self.deselected_owners and ow not in self.selected_owners:
+          self.writeln()
+          self.print_owned_files_for(ow)
+
+  def list_files(self):
+    self.indent()
+    if len(self.unreviewed_files) > 5:
+      for file_name in sorted(self.unreviewed_files):
+        self.print_file_info(file_name)
+    else:
+      for file_name in self.unreviewed_files:
+        self.print_file_info_detailed(file_name)
+    self.unindent()
+
+  def pick_owner(self, ow):
+    # Allowing to omit domain suffixes
+    if ow not in self.owner_to_files:
+      if ow + self.email_postfix in self.owner_to_files:
+        ow += self.email_postfix
+
+    if ow not in self.owner_to_files:
+      self.writeln('You cannot pick ' + self.bold_name(ow) + ' manually. ' +
+                   'It\'s an invalid name or not related to the change list.')
+      return False
+    elif ow in self.selected_owners:
+      self.writeln('You cannot pick ' + self.bold_name(ow) + ' manually. ' +
+                   'It\'s already selected.')
+      return False
+    elif ow in self.deselected_owners:
+      self.writeln('You cannot pick ' + self.bold_name(ow) + ' manually.' +
+                   'It\'s already unselected.')
+      return False
+
+    self.select_owner(ow)
+    return True
+
+  def print_result(self):
+    # Print results
+    self.writeln()
+    self.writeln()
+    self.writeln('** You selected these owners **')
+    self.writeln()
+    for owner in self.selected_owners:
+      self.writeln(self.bold_name(owner) + ':')
+      self.indent()
+      for file_name in sorted(self.owner_to_files[owner]):
+        self.writeln(file_name)
+      self.unindent()
+
+  def hr(self):
+    self.writeln('=====================')
+
+  def run(self):
+    self.reset()
+    while len(self.owners_queue) > 0 and len(self.unreviewed_files) > 0:
+      owner = self.owners_queue[0]

[Dirk Pranke, 2013/07/27 00:06:03]

should this be owner = self.owners_queue.pop(0) ? 

It's not clear to me that this is actually a queue since it doesn't look like
you pop things in the order they show up in the list. Does the code become
clearer if you actually remove the owners one at a time and deal with them
(re-adding if necessary when they're deferred)?

(I'm not actually sure if this is the right thing to do as I don't fully
understand the algorithm yet, but I think that's part of the problem as the name
suggests to me things work a particular way and I'm not sure if that's true).

[Bei Zhang, 2013/07/30 05:59:17]

There are several decisions have to be made according to the full list of the
owners_queue. So I defer the pop later. It is a queue because things are always
inserted at the tail of the queue, and the current owner is always get from the
head of the queue.

On 2013/07/27 00:06:03, Dirk Pranke wrote:

[Dirk Pranke, 2013/07/30 22:01:00]

I don't think I'm seeing the decisions you have in mind. Can you point them out
to me?

In particular, it seems like if you did a pop up front, you could get rid of the
'if .. continue' blocks on lines 329-330 and 334-336. It also seems like you
could simplify select_owner() and deselect_owner() quite a bit, and then the
remove()/pop() call would be in the same place as where you are iterating over
the list. All of this feels like it would be clearer to me (but perhaps it
doesn't work for reasons I'm missing ...).

+
+      if owner in self.selected_owners:
+        continue
+      if len(self.unreviewed_files) == 0:
+        self.writeln('Finished.\n\n')
+        break
+      if owner in self.deselected_owners:
+        # If this owner is already deselected.
+        continue
+      if not any((file_name in self.unreviewed_files)
+                 for file_name in self.owner_to_files[owner]):
+        self.deselect_owner(owner)
+        continue
+      self.hr()
+      self.writeln(
+          self.bold(str(len(self.unreviewed_files))) + ' file(s) left.')
+      self.print_owned_files_for(owner)

[Dirk Pranke, 2013/07/27 00:06:03]

I'd probably pull lines 341-344 into a separate method just to provide a little
better separation between the logic and the interface.

[Bei Zhang, 2013/07/30 05:59:17]

Done.

+
+      while True:
+        self.writeln('Add ' + self.bold_name(owner) + ' as your reviewer? ')
+        inp = raw_input(
+            '[yes/no/Defer/pick/files/owners/quit/restart]: ').lower()

[Dirk Pranke, 2013/07/27 00:06:03]

Calling raw_input() directly makes this hard to write unit tests ...

I'd split lines 347-349 out into a separate method that also validates and
canonicalizes the input. That should make the rest of this loop more concise.

[Bei Zhang, 2013/07/30 05:59:17]

Done.

+        if inp == 'y' or inp == 'yes':
+          self.select_owner(owner)
+          break
+        elif inp == 'n' or inp == 'no':
+          self.deselect_owner(owner)
+          break
+        elif inp == '' or inp == 'd' or inp == 'defer':
+          self.owners_queue.append(self.owners_queue.pop(0))
+          break
+        elif inp == 'f' or inp == 'files':
+          self.list_files()
+        elif inp == 'o' or inp == 'owners':
+          self.list_owners(self.owners_queue)
+        elif inp == 'p' or inp == 'pick':
+          self.pick_owner(raw_input('Pick an owner: '))
+          break
+        elif inp.startswith('p ') or inp.startswith('pick '):
+          self.pick_owner(inp.split(' ', 2)[1])
+          break
+        elif inp == 'r' or inp == 'restart':
+          self.reset()
+          break
+        elif inp == 'q' or inp == 'quit':
+          # Exit with error
+          return 1
+
+    self.print_result()
+    return 0