Revert of Track step nesting in StreamEngine.

recipe_engine/stream.py

Index: recipe_engine/stream.py
diff --git a/recipe_engine/stream.py b/recipe_engine/stream.py
index f916c80e01e7e2936e8f9a7c37f5113ee3334043..4aeb9a6fcb68eddd741d7fc1c8f11b99bf07420d 100644
--- a/recipe_engine/stream.py
+++ b/recipe_engine/stream.py
@@ -62,14 +62,16 @@
     def set_step_status(self, status):
       raise NotImplementedError()
 
+    def set_nest_level(self, nest_level):
+      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 new_step_stream(self, step_name, allow_subannotations=False):
     """Creates a new StepStream in this engine.
 
     The step will be considered started at the moment this method is called.
@@ -79,14 +81,8 @@
     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.
     """
-    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):
@@ -108,13 +104,11 @@
 # had them formalized...).
 class ProductStreamEngine(StreamEngine):
   def __init__(self, engine_a, engine_b):
-    assert engine_a and engine_b
     self._engine_a = engine_a
     self._engine_b = engine_b
 
   class Stream(StreamEngine.Stream):
     def __init__(self, stream_a, stream_b):
-      assert stream_a and stream_b
       self._stream_a = stream_a
       self._stream_b = stream_b
 
@@ -143,15 +137,14 @@
     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_nest_level = _void_product('set_nest_level')
     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_name, allow_subannotations=False):
     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_name, allow_subannotations),
+        self._engine_b.new_step_stream(step_name, allow_subannotations))
 
   def open(self):
     self._engine_a.open()
@@ -178,10 +171,11 @@
     add_step_link = _noop
     reset_subannotation_state = _noop
     set_step_status = _noop
+    set_nest_level = _noop
     set_build_property = _noop
     trigger = _noop
 
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
+  def new_step_stream(self, step_name, allow_subannotations=False):
     return self.StepStream()
 
 
@@ -239,6 +233,9 @@
           'Cannot set successful status after status is %s' % self._status)
       self._status = status
 
+    def set_nest_level(self, nest_level):
+      assert isinstance(nest_level, int)
+
     def set_build_property(self, key, value):
       pass
 
@@ -262,10 +259,78 @@
       assert self._open
       self._open = False
 
-  def _new_step_stream(self, step_name, allow_subannotations, nest_level):
+  def new_step_stream(self, step_name, allow_subannotations=False):
     assert step_name not in self._streams, 'Step %s already exists' % step_name
     self._streams.add(step_name)
     return self.StepStream(self, step_name)
+
+
+class AnnotationStepStream(StreamEngine.StepStream):
+  def __init__(self, emit_timestamps=False, time_fn=None):
+    self.emit_timestamps = emit_timestamps
+    self.time_fn = time_fn or time.time
+
+  def basic_write(self, line):
+    raise NotImplementedError()
+
+  def output_annotation(self, *args):
+    self.basic_write(
+        '@@@' + '@'.join(map(encode_str, args)) + '@@@\n')
+
+  def write_line(self, line):
+    if line.startswith('@@@'):
+      self.basic_write('!' + line + '\n')
+    else:
+      self.basic_write(line + '\n')
+
+  def close(self):
+    if self.emit_timestamps:
+      self.output_annotation('CURRENT_TIMESTAMP', self.time_fn())
+    self.output_annotation('STEP_CLOSED')
+
+  def new_log_stream(self, log_name):
+    return self.StepLogStream(self, log_name)
+
+  def add_step_text(self, text):
+    self.output_annotation('STEP_TEXT', text)
+
+  def add_step_summary_text(self, text):
+    self.output_annotation('STEP_SUMMARY_TEXT', text)
+
+  def add_step_link(self, name, url):
+    self.output_annotation('STEP_LINK', name, url)
+
+  def set_step_status(self, status):
+    if status == 'SUCCESS':
+      pass
+    elif status == 'WARNING':
+      self.output_annotation('STEP_WARNINGS')
+    elif status == 'FAILURE':
+      self.output_annotation('STEP_FAILURE')
+    elif status == 'EXCEPTION':
+      self.output_annotation('STEP_EXCEPTION')
+    else:
+      raise Exception('Impossible status %s' % status)
+
+  def set_nest_level(self, nest_level):
+    self.output_annotation('STEP_NEST_LEVEL', str(nest_level))
+
+  def set_build_property(self, key, value):
+    self.output_annotation('SET_BUILD_PROPERTY', key, value)
+
+  def trigger(self, spec):
+    self.output_annotation('STEP_TRIGGER', spec)
+
+  class StepLogStream(StreamEngine.Stream):
+    def __init__(self, step_stream, log_name):
+      self._step_stream = step_stream
+      self._log_name = log_name.replace('/', '/')
+
+    def write_line(self, line):
+      self._step_stream.output_annotation('STEP_LOG_LINE', self._log_name, line)
+
+    def close(self):
+      self._step_stream.output_annotation('STEP_LOG_END', self._log_name)
 
 
 class AnnotatorStreamEngine(StreamEngine):
@@ -278,109 +343,47 @@
 
   def open(self):
     super(AnnotatorStreamEngine, self).open()
-    self.output_current_time()
-    self.output_root_annotation('HONOR_ZERO_RETURN_CODE')
+    if self.emit_timestamps:
+      self.output_current_time()
+    self.output_annotation('HONOR_ZERO_RETURN_CODE')
 
   def close(self):
     super(AnnotatorStreamEngine, self).close()
-    self.output_current_time()
-
-  def output_current_time(self, step=None):
+    if self.emit_timestamps:
+      self.output_current_time()
+
+  def output_current_time(self):
     """Prints CURRENT_TIMESTAMP annotation with current time."""
-    if step:
-      self._step_cursor(step)
-    if self.emit_timestamps:
-      self.output_root_annotation('CURRENT_TIMESTAMP', self.time_fn())
-
-  @staticmethod
-  def write_annotation(outstream, *args):
+    self.output_annotation('CURRENT_TIMESTAMP', self.time_fn())
+
+  def output_annotation(self, *args):
     # Flush the stream before & after engine annotations, because they can
     # change which step we are talking about and this matters to buildbot.
-    outstream.flush()
-    outstream.write(
+    self._outstream.flush()
+    self._outstream.write(
         '@@@' + '@'.join(map(encode_str, args)) + '@@@\n')
-    outstream.flush()
-
-  def output_root_annotation(self, *args):
-    self.write_annotation(self._outstream, *args)
-
-  def _step_cursor(self, step_name):
-    if self._current_step != step_name:
-      self.output_root_annotation('STEP_CURSOR', step_name)
-      self._current_step = step_name
-    if step_name not in self._opened:
-      self.output_current_time()
-      self.output_root_annotation('STEP_STARTED')
-      self._opened.add(step_name)
-
-  class StepStream(StreamEngine.StepStream):
-    def __init__(self, engine, outstream, step_name):
-      super(StreamEngine.StepStream, self).__init__()
-
+    self._outstream.flush()
+
+  def _step_cursor(self, name):
+    if self._current_step != name:
+      self.output_annotation('STEP_CURSOR', name)
+      self._current_step = name
+    if name not in self._opened:
+      if self.emit_timestamps:
+        self.output_current_time()
+      self.output_annotation('STEP_STARTED')
+      self._opened.add(name)
+
+  class StepStream(AnnotationStepStream):
+    def __init__(self, engine, step_name):
+      AnnotationStepStream.__init__(
+          self, emit_timestamps=engine.emit_timestamps, time_fn=engine.time_fn)
       self._engine = engine
-      self._outstream = outstream
       self._step_name = step_name
 
     def basic_write(self, line):
       self._engine._step_cursor(self._step_name)
-      self._outstream.write(line)
-
-    def close(self):
-      self._engine.output_current_time(step=self._step_name)
-      self.output_annotation('STEP_CLOSED')
-
-    def output_annotation(self, *args):
-      self._engine._step_cursor(self._step_name)
-      self._engine.write_annotation(self._outstream, *args)
-
-    def write_line(self, line):
-      if line.startswith('@@@'):
-        self.basic_write('!' + line + '\n')
-      else:
-        self.basic_write(line + '\n')
-
-    def new_log_stream(self, log_name):
-      return self._engine.StepLogStream(self, log_name)
-
-    def add_step_text(self, text):
-      self.output_annotation('STEP_TEXT', text)
-
-    def add_step_summary_text(self, text):
-      self.output_annotation('STEP_SUMMARY_TEXT', text)
-
-    def add_step_link(self, name, url):
-      self.output_annotation('STEP_LINK', name, url)
-
-    def set_step_status(self, status):
-      if status == 'SUCCESS':
-        pass
-      elif status == 'WARNING':
-        self.output_annotation('STEP_WARNINGS')
-      elif status == 'FAILURE':
-        self.output_annotation('STEP_FAILURE')
-      elif status == 'EXCEPTION':
-        self.output_annotation('STEP_EXCEPTION')
-      else:
-        raise Exception('Impossible status %s' % status)
-
-    def set_build_property(self, key, value):
-      self.output_annotation('SET_BUILD_PROPERTY', key, value)
-
-    def trigger(self, spec):
-      self.output_annotation('STEP_TRIGGER', spec)
-
-
-  class StepLogStream(StreamEngine.Stream):
-    def __init__(self, step_stream, log_name):
-      self._step_stream = step_stream
-      self._log_name = log_name.replace('/', '/')
-
-    def write_line(self, line):
-      self._step_stream.output_annotation('STEP_LOG_LINE', self._log_name, line)
-
-    def close(self):
-      self._step_stream.output_annotation('STEP_LOG_END', self._log_name)
-
+      self._engine._outstream.write(line)
 
   class AllowSubannotationsStepStream(StepStream):
     def write_line(self, line):
@@ -393,23 +396,25 @@
     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 _create_step_stream(self, step_name, outstream, allow_subannotations,
-                          nest_level):
-    self.output_root_annotation('SEED_STEP', step_name)
+  def new_step_stream(self, step_name, allow_subannotations=False):
+    self.output_annotation('SEED_STEP', step_name)
     if allow_subannotations:
-      stream = self.AllowSubannotationsStepStream(self, outstream, step_name)
+      return self.AllowSubannotationsStepStream(self, step_name)
     else:
-      stream = self.StepStream(self, outstream, step_name)
-
-    if nest_level > 0:
-      # Emit our current nest level, if we are nested.
-      stream.output_annotation('STEP_NEST_LEVEL', str(nest_level))
-    return stream
+      return self.StepStream(self, step_name)
+
+
+class BareAnnotationStepStream(AnnotationStepStream):
+  """A StepStream that is not tied to any engine, and emits assuming it has the
+  cursor.
+
+  This is used for capturing the annotations in the engine.
+  """
+  def __init__(self, outstream):
+    self._outstream = outstream
+
+  def basic_write(self, line):
+    self._outstream.write(encode_str(line))
 
 
 def encode_str(s):