Formally define step config, pass to stream.

recipe_engine/stream.py

 # Copyright 2015 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Abstract stream interface for representing recipe runs.
 
 We need to create streams for steps (and substeps) and also LOG_LINE steps.
 LogDog will implement LOG_LINE steps as real logs (i.e. uniformly), but
 annotations will implement them differently from normal logs, so we need
 a way to distinguish.
 
 StreamEngine will coordinate the multiplexing of streams.  In the case of
 annotations, this involves keeping track of the STEP_CURSOR and setting it
 accordingly, as well as filtering @@@ lines.
 
 Stream is a virtual well-behaved stream (associated with an Engine) which you
 can just write to without worrying.
 """
 
 import json
 import time
 
 from . import env
+from . import recipe_api
 
 
 class StreamEngine(object):
   class Stream(object):
     def write_line(self, line):
       raise NotImplementedError()
 
     def write_split(self, string):
       """Write a string (which may contain newlines) to the stream.  It will
       be terminated by a newline."""
@@ skipping 27 matching lines @@
 
     def set_step_status(self, status):
       raise NotImplementedError()
 
     def set_build_property(self, key, value):
       raise NotImplementedError()
 
     def trigger(self, trigger_spec):
       raise NotImplementedError()
 
-  def new_step_stream(self, step_name, allow_subannotations=False,
-                      nest_level=None):
+  def make_step_stream(self, name, **kwargs):
+    """Shorthand for creating a step stream from a step configuration dict."""
+    kwargs['name'] = name
+    return self.new_step_stream(recipe_api._make_step_config(**kwargs))
+
+  def new_step_stream(self, step_config):
     """Creates a new StepStream in this engine.
 
     The step will be considered started at the moment this method is called.
 
     TODO(luqui): allow_subannotations is a bit of a hack, whether to allow
     annotations that this step emits through to the annotator (True), or
     guard them by prefixing them with ! (False).  The proper way to do this
     is to implement an annotations parser that converts to StreamEngine calls;
     i.e. parse -> re-emit.
 
     Args:
-      nest_level (int): The nest level of the step. None/0 are top-level.
+      step_config (recipe_api.StepConfig): The step configuration.
     """
-    return self._new_step_stream(step_name, allow_subannotations, nest_level)
-
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
-    """ABC overridable function for "new_step_stream" with no defaults."""
     raise NotImplementedError()
 
   def open(self):
     pass
 
   def close(self):
     pass
 
   def __enter__(self):
     self.open()
@@ skipping 39 matching lines @@
           self._stream_b.new_log_stream(log_name))
 
     add_step_text = _void_product('add_step_text')
     add_step_summary_text = _void_product('add_step_summary_text')
     add_step_link = _void_product('add_step_link')
     reset_subannotation_state = _void_product('reset_subannotation_state')
     set_step_status = _void_product('set_step_status')
     set_build_property = _void_product('set_build_property')
     trigger = _void_product('trigger')
 
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
+  def new_step_stream(self, step_config):
     return self.StepStream(
-        self._engine_a._new_step_stream(
-            step_name, allow_subannotations, nest_level),
-        self._engine_b._new_step_stream(
-            step_name, allow_subannotations, nest_level))
+        self._engine_a.new_step_stream(step_config),
+        self._engine_b.new_step_stream(step_config))
 
   def open(self):
     self._engine_a.open()
     self._engine_b.open()
 
   def close(self):
     self._engine_a.close()
     self._engine_b.close()
 
 
 def _noop(*args, **kwargs):
   pass
 
 class NoopStreamEngine(StreamEngine):
   class Stream(StreamEngine.Stream):
     write_line = _noop
     close = _noop
 
   class StepStream(Stream):
     def new_log_stream(self, log_name):
       return NoopStreamEngine.Stream()
     add_step_text = _noop
     add_step_summary_text = _noop
     add_step_link = _noop
     reset_subannotation_state = _noop
     set_step_status = _noop
     set_build_property = _noop
     trigger = _noop
 
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
+  def new_step_stream(self, step_config):
     return self.StepStream()
 
 
 class StreamEngineInvariants(StreamEngine):
   """Checks that the users are using a StreamEngine hygenically.
 
   Multiply with actually functional StreamEngines so you don't have to check
   these all over the place.
   """
   def __init__(self):
@@ skipping 60 matching lines @@
     def write_line(self, line):
       assert '\n' not in line
       assert self._step_stream._open
       assert self._open
 
     def close(self):
       assert self._step_stream._open
       assert self._open
       self._open = False
 
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
-    assert step_name not in self._streams, 'Step %s already exists' % step_name
-    self._streams.add(step_name)
-    return self.StepStream(self, step_name)
+  def new_step_stream(self, step_config):
+    assert step_config.name not in self._streams, (
+        'Step %s already exists' % step_config.name)
+    self._streams.add(step_config.name)
+    return self.StepStream(self, step_config.name)
 
 
 class AnnotatorStreamEngine(StreamEngine):
   def __init__(self, outstream, emit_timestamps=False, time_fn=None):
     self._current_step = None
     self._opened = set()
     self._outstream = outstream
     self.emit_timestamps = emit_timestamps
     self.time_fn = time_fn or time.time
 
@@ skipping 108 matching lines @@
       self.basic_write(line + '\n')
 
     # HACK(luqui): If the subannotator script changes the active step, we need
     # a way to get back to the real step that spawned the script.  The right
     # way to do that is to parse the annotation stream and re-emit it.  But for
     # now we just provide this method.
     def reset_subannotation_state(self):
       self._engine._current_step = None
 
 
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
-    return self._create_step_stream(step_name, self._outstream,
-                                    allow_subannotations, nest_level)
+  def new_step_stream(self, step_config):
+    self.output_root_annotation('SEED_STEP', step_config.name)
+    return self._create_step_stream(step_config, self._outstream)
 
-  def _create_step_stream(self, step_name, outstream, allow_subannotations,
-                          nest_level):
-    self.output_root_annotation('SEED_STEP', step_name)
-    if allow_subannotations:
-      stream = self.AllowSubannotationsStepStream(self, outstream, step_name)
+  def _create_step_stream(self, step_config, outstream):
+    if step_config.allow_subannotations:
+      stream = self.AllowSubannotationsStepStream(self, outstream,
+                                                  step_config.name)
     else:
-      stream = self.StepStream(self, outstream, step_name)
+      stream = self.StepStream(self, outstream, step_config.name)
 
-    if nest_level > 0:
+    if step_config.nest_level > 0:
       # Emit our current nest level, if we are nested.
-      stream.output_annotation('STEP_NEST_LEVEL', str(nest_level))
+      stream.output_annotation('STEP_NEST_LEVEL', str(step_config.nest_level))
     return stream
 
 
 def encode_str(s):
   """Tries to encode a string into a python str type.
 
   Currently buildbot only supports ascii. If we have an error decoding the
   string (which means it might not be valid ascii), we decode the string with
   the 'replace' error mode, which replaces invalid characters with a suitable
   replacement character.
   """
   try:
     return str(s)
   except UnicodeEncodeError:
     return s.encode('utf-8', 'replace')
   except UnicodeDecodeError:
     return s.decode('utf-8', 'replace')