Assert validity of stream parameters.

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.
@@ skipping 28 matching lines @@
   Args:
     stream (StreamEngine.Stream): The stream to output to.
     it (iterable): An iterable that yields strings to write.
   """
   for text in it:
     lines = (text.split('\n') if text else ('',))
     for line in lines:
       stream.write_line(line)
 
 
+def _check(cond, msg, **kwargs):
+  """Runtime assertion used for parameter correctness."""
+  if not cond:
+    raise ValueError('Failed assertion: %s %r' % (msg, kwargs))
+
+
 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."""
       for actual_line in string.splitlines() or ['']: # preserve empty lines
         self.write_line(actual_line)
 
@@ skipping 26 matching lines @@
       raise NotImplementedError()
 
     def set_build_property(self, key, value):
       raise NotImplementedError()
 
     def trigger(self, trigger_spec):
       raise NotImplementedError()
 
   def make_step_stream(self, name, **kwargs):
     """Shorthand for creating a step stream from a step configuration dict."""
+    _check('\n' not in name, 'Stream name must not contain a newline.',
+           name=name)
     kwargs['name'] = name
     return self.new_step_stream(recipe_api.StepConfig.create(**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
@@ skipping 201 matching lines @@
   class StepStream(StreamEngine.StepStream):
     def __init__(self, engine, step_name):
       super(StreamEngineInvariants.StepStream, self).__init__()
       self._engine = engine
       self._step_name = step_name
       self._open = True
       self._logs = {}
       self._status = 'SUCCESS'
 
     def write_line(self, line):
-      assert '\n' not in line
+      assert '\n' not in line, 'Individual line must not contain a newline.'
       assert self._open
 
     def close(self):
       assert self._open
       for log_name, log in self._logs.iteritems():
         assert not log._open, 'Log %s still open when closing step %s' % (
           log_name, self._step_name)
       self._open = False
 
     def new_log_stream(self, log_name):
       assert self._open
       assert log_name not in self._logs, 'Log %s already exists in step %s' % (
           log_name, self._step_name)
+
+      _check('\n' not in log_name, 'Log name must not contain a newline.',
+             log_name=log_name)
       ret = self._engine.LogStream(self, log_name)
       self._logs[log_name] = ret
       return ret
 
     def add_step_text(self, text):
-      pass
+      _check('\n' not in text, 'Step text must not contain a newline.',
+             text=text)
 
     def add_step_summary_text(self, text):
-      pass
+      _check('\n' not in text, 'Step summary text must not contain a newline.',
+             text=text)
 
     def add_step_link(self, name, url):
-      assert isinstance(name, basestring), 'Link name %s is not a string' % name
-      assert isinstance(url, basestring), 'Link url %s is not a string' % url
+      _check(isinstance(name, basestring), 'Link name is not a string',
+             name=name)
+      _check('\n' not in name, 'Link name must not contain a newline.',
+             name=name)
+      _check(isinstance(url, basestring),
+             'Link URL is not a string', url=url)
+      _check('\n' not in url, 'Link URL must not contain a newline.',
+             url=url)
 
     def set_step_status(self, status):
       assert status in ('SUCCESS', 'WARNING', 'FAILURE', 'EXCEPTION')
       if status == 'SUCCESS':
         # A constraint imposed by the annotations implementation
         assert self._status == 'SUCCESS', (
           'Cannot set successful status after status is %s' % self._status)
       self._status = status
 
     def set_build_property(self, key, value):
-      pass
+      _check('\n' not in key, 'Property key must not contain a newline.',
+             key=key)
+      _check('\n' not in value, 'Property value must not contain a newline.',
+             value=value)
+      json.loads(value) # value must be a valid JSON object.
 
     def trigger(self, spec):
-      assert '\n' not in spec # Spec must fit on one line.
-      json.loads(spec) # Spec must be a valid json object.
+      _check('\n' not in spec, 'Spec must not contain a newline.',
+             spec=spec)
+      json.loads(spec) # Spec must be a valid JSON object.
 
   class LogStream(StreamEngine.Stream):
     def __init__(self, step_stream, log_name):
       self._step_stream = step_stream
       self._log_name = log_name
       self._open = True
 
     def write_line(self, line):
-      assert '\n' not in line
+      assert '\n' not in line, 'Individual line must not contain a newline.'
       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_config):
     assert step_config.name not in self._streams, (
@@ skipping 152 matching lines @@
   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')