Cross-repo recipe package system.

third_party/recipe_engine/third_party/annotator.py

-# Copyright (c) 2015 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.
-
-"""Contains the parsing system of the Chromium Buildbot Annotator."""
-
-import os
-import sys
-import traceback
-
-# These are maps of annotation key -> number of expected arguments.
-STEP_ANNOTATIONS = {
-    'SET_BUILD_PROPERTY': 2,
-    'STEP_CLEAR': 0,
-    'STEP_EXCEPTION': 0,
-    'STEP_FAILURE': 0,
-    'STEP_LINK': 2,
-    'STEP_LOG_END': 1,
-    'STEP_LOG_END_PERF': 2,
-    'STEP_LOG_LINE': 2,
-    'STEP_SUMMARY_CLEAR': 0,
-    'STEP_SUMMARY_TEXT': 1,
-    'STEP_TEXT': 1,
-    'STEP_TRIGGER': 1,
-    'STEP_WARNINGS': 0,
-    'STEP_NEST_LEVEL': 1,
-}
-
-CONTROL_ANNOTATIONS = {
-    'STEP_CLOSED': 0,
-    'STEP_STARTED': 0,
-}
-
-STREAM_ANNOTATIONS = {
-    'HALT_ON_FAILURE': 0,
-    'HONOR_ZERO_RETURN_CODE': 0,
-    'SEED_STEP': 1,
-    'SEED_STEP_TEXT': 2,
-    'STEP_CURSOR': 1,
-}
-
-DEPRECATED_ANNOTATIONS = {
-    'BUILD_STEP': 1,
-}
-
-ALL_ANNOTATIONS = {}
-ALL_ANNOTATIONS.update(STEP_ANNOTATIONS)
-ALL_ANNOTATIONS.update(CONTROL_ANNOTATIONS)
-ALL_ANNOTATIONS.update(STREAM_ANNOTATIONS)
-ALL_ANNOTATIONS.update(DEPRECATED_ANNOTATIONS)
-
-# This is a mapping of old_annotation_name -> new_annotation_name.
-# Theoretically all annotator scripts should use the new names, but it's hard
-# to tell due to the decentralized nature of the annotator.
-DEPRECATED_ALIASES = {
-    'BUILD_FAILED': 'STEP_FAILURE',
-    'BUILD_WARNINGS': 'STEP_WARNINGS',
-    'BUILD_EXCEPTION': 'STEP_EXCEPTION',
-    'link': 'STEP_LINK',
-}
-
-# A couple of the annotations have the format:
-#  @@@THING arg@@@
-# for reasons no one knows. We only need this case until all masters have been
-# restarted to pick up the new master-side parsing code.
-OLD_STYLE_ANNOTATIONS = set((
-  'SEED_STEP',
-  'STEP_CURSOR',
-))
-
-
-def emit(line, stream, flush_before=None):
-  if flush_before:
-    flush_before.flush()
-  print >> stream
-  # WinDOS can only handle 64kb of output to the console at a time, per process.
-  if sys.platform.startswith('win'):
-    lim = 2**15
-    while line:
-      to_print, line = line[:lim], line[lim:]
-      stream.write(to_print)
-    stream.write('\n')
-  else:
-    print >> stream, line
-  stream.flush()
-
-
-class MetaAnnotationPrinter(type):
-  def __new__(mcs, name, bases, dct):
-    annotation_map = dct.get('ANNOTATIONS')
-    if annotation_map:
-      for key, v in annotation_map.iteritems():
-        key = key.lower()
-        dct[key] = mcs.make_printer_fn(key, v)
-    return type.__new__(mcs, name, bases, dct)
-
-  @staticmethod
-  def make_printer_fn(name, n_args):
-    """Generates a method which emits an annotation to the log stream."""
-    upname = name.upper()
-    if upname in OLD_STYLE_ANNOTATIONS:
-      assert n_args >= 1
-      fmt = '@@@%s %%s%s@@@' % (upname, '@%s' * (n_args - 1))
-    else:
-      fmt = '@@@%s%s@@@' % (upname, '@%s' * n_args)
-
-    inner_args = n_args + 1  # self counts
-    infix = '1 argument' if inner_args == 1 else ('%d arguments' % inner_args)
-    err = '%s() takes %s (%%d given)' % (name, infix)
-
-    def printer(self, *args):
-      if len(args) != n_args:
-        raise TypeError(err % (len(args) + 1))
-      self.emit(fmt % args)
-    printer.__name__ = name
-    printer.__doc__ = """Emits an annotation for %s.""" % name.upper()
-
-    return printer
-
-
-class AnnotationPrinter(object):
-  """A derivable class which will inject annotation-printing methods into the
-  subclass.
-
-  A subclass should define a class variable ANNOTATIONS equal to a
-  dictionary of the form { '<ANNOTATION_NAME>': <# args> }. This class will
-  then inject methods whose names are the undercased version of your
-  annotation names, and which take the number of arguments specified in the
-  dictionary.
-
-  Example:
-    >>> my_annotations = { 'STEP_LOG_LINE': 2 }
-    >>> class MyObj(AnnotationPrinter):
-    ...   ANNOTATIONS = my_annotations
-    ...
-    >>> o = MyObj()
-    >>> o.step_log_line('logname', 'here is a line to put in the log')
-    @@@STEP_LOG_LINE@logname@here is a line to put in the log@@@
-    >>> o.step_log_line()
-    Traceback (most recent call last):
-      File "<stdin>", line 1, in <module>
-    TypeError: step_log_line() takes exactly 3 arguments (1 given)
-    >>> o.setp_log_line.__doc__
-    "Emits an annotation for STEP_LOG_LINE."
-    >>>
-  """
-  __metaclass__ = MetaAnnotationPrinter
-
-  def __init__(self, stream, flush_before):
-    self.stream = stream
-    self.flush_before = flush_before
-
-  def emit(self, line):
-    emit(line, self.stream, self.flush_before)
-
-
-class StepCommands(AnnotationPrinter):
-  """Class holding step commands. Intended to be subclassed."""
-  ANNOTATIONS = STEP_ANNOTATIONS
-
-  def __init__(self, stream, flush_before):
-    super(StepCommands, self).__init__(stream, flush_before)
-    self.emitted_logs = set()
-
-  def write_log_lines(self, logname, lines, perf=None):
-    if logname in self.emitted_logs:
-      raise ValueError('Log %s has been emitted multiple times.' % logname)
-    self.emitted_logs.add(logname)
-
-    logname = logname.replace('/', '&#x2f;')
-
-    for line in lines:
-      for actual_line in line.split('\n'):
-        self.step_log_line(logname, actual_line)
-
-    if perf:
-      self.step_log_end_perf(logname, perf)
-    else:
-      self.step_log_end(logname)
-
-
-class StepControlCommands(AnnotationPrinter):
-  """Subclass holding step control commands. Intended to be subclassed.
-
-  This is subclassed out so callers in StructuredAnnotationStep can't call
-  step_started() or step_closed().
-  """
-  ANNOTATIONS = CONTROL_ANNOTATIONS
-
-
-class StructuredAnnotationStep(StepCommands, StepControlCommands):
-  """Helper class to provide context for a step."""
-
-  def __init__(self, annotation_stream, *args, **kwargs):
-    self.annotation_stream = annotation_stream
-    super(StructuredAnnotationStep, self).__init__(*args, **kwargs)
-    self.control = StepControlCommands(self.stream, self.flush_before)
-    self.emitted_logs = set()
-
-
-  def __enter__(self):
-    return self.step_started()
-
-  def step_started(self):
-    self.control.step_started()
-    return self
-
-  def __exit__(self, exc_type, exc_value, tb):
-    self.annotation_stream.step_cursor(self.annotation_stream.current_step)
-    #TODO(martinis) combine this and step_ended
-    if exc_type:
-      self.step_exception_occured(exc_type, exc_value, tb)
-
-    self.control.step_closed()
-    self.annotation_stream.current_step = ''
-    return not exc_type
-
-  def step_exception_occured(self, exc_type, exc_value, tb):
-    trace = traceback.format_exception(exc_type, exc_value, tb)
-    trace_lines = ''.join(trace).split('\n')
-    self.write_log_lines('exception', filter(None, trace_lines))
-    self.step_exception()
-
-  def step_ended(self):
-    self.annotation_stream.step_cursor(self.annotation_stream.current_step)
-    self.control.step_closed()
-    self.annotation_stream.current_step = ''
-
-    return True
-
-
-class StructuredAnnotationStream(AnnotationPrinter):
-  """Provides an interface to handle an annotated build.
-
-  StructuredAnnotationStream handles most of the step setup and closure calls
-  for you. All you have to do is execute your code within the steps and set any
-  failures or warnings that come up. You may optionally provide a list of steps
-  to seed before execution.
-
-  Usage:
-
-  stream = StructuredAnnotationStream()
-  with stream.step('compile') as s:
-    # do something
-    if error:
-      s.step_failure()
-  with stream.step('test') as s:
-    # do something
-    if warnings:
-      s.step_warnings()
-  """
-  ANNOTATIONS = STREAM_ANNOTATIONS
-
-  def __init__(self, stream=sys.stdout,
-               flush_before=sys.stderr,
-               seed_steps=None):  # pylint: disable=W0613
-    super(StructuredAnnotationStream, self).__init__(stream=stream,
-                                                     flush_before=flush_before)
-    self.current_step = ''
-
-  def step(self, name):
-    """Provide a context with which to execute a step."""
-    if self.current_step:
-      raise Exception('Can\'t start step %s while in step %s.' % (
-          name, self.current_step))
-
-    self.seed_step(name)
-    self.step_cursor(name)
-    self.current_step = name
-    return StructuredAnnotationStep(self, stream=self.stream,
-                                    flush_before=self.flush_before)
-
-
-def MatchAnnotation(line, callback_implementor):
-  """Call back into |callback_implementor| if line contains an annotation.
-
-  Args:
-    line (str) - The line to analyze
-    callback_implementor (object) - An object which contains methods
-      corresponding to all of the annotations in the |ALL_ANNOTATIONS|
-      dictionary. For example, it should contain a method STEP_SUMMARY_TEXT
-      taking a single argument.
-
-  Parsing method:
-    * if line doesn't match /^@@@.*@@@$/, return without calling back
-    * Look for the first '@' or ' '
-  """
-  if not (line.startswith('@@@') and line.endswith('@@@') and len(line) > 6):
-    return
-  line = line[3:-3]
-
-  # look until the first @ or ' '
-  idx = min((x for x in (line.find('@'), line.find(' '), len(line)) if x > 0))
-  cmd_text = line[:idx]
-  cmd = DEPRECATED_ALIASES.get(cmd_text, cmd_text)
-
-  field_count = ALL_ANNOTATIONS.get(cmd)
-  if field_count is None:
-    raise Exception('Unrecognized annotator command "%s"' % cmd_text)
-
-  if field_count:
-    if idx == len(line):
-      raise Exception('Annotator command "%s" expects %d args, got 0.'
-                      % (cmd_text, field_count))
-
-    line = line[idx+1:]
-
-    args = line.split('@', field_count-1)
-    if len(args) != field_count:
-      raise Exception('Annotator command "%s" expects %d args, got %d.'
-                      % (cmd_text, field_count, len(args)))
-  else:
-    line = line[len(cmd_text):]
-    if line:
-      raise Exception('Annotator command "%s" expects no args, got cruft "%s".'
-                      % (cmd_text, line))
-    args = []
-
-  fn = getattr(callback_implementor, cmd, None)
-  if fn is None:
-    raise Exception('"%s" does not implement "%s"'
-                    % (callback_implementor, cmd))
-
-  fn(*args)
-
-
-def print_step(step, env, stream):
-  """Prints the step command and relevant metadata.
-
-  Intended to be similar to the information that Buildbot prints at the
-  beginning of each non-annotator step.
-  """
-  step_info_lines = []
-  step_info_lines.append(' '.join(step['cmd']))
-  step_info_lines.append('in dir %s:' % (step['cwd'] or os.getcwd()))
-  for key, value in sorted(step.items()):
-    if value is not None:
-      if callable(value):
-        # This prevents functions from showing up as:
-        #   '<function foo at 0x7f523ec7a410>'
-        # which is tricky to test.
-        value = value.__name__+'(...)'
-      step_info_lines.append(' %s: %s' % (key, value))
-  step_info_lines.append('full environment:')
-  for key, value in sorted(env.items()):
-    step_info_lines.append(' %s: %s' % (key, value))
-  step_info_lines.append('')
-  stream.emit('\n'.join(step_info_lines))