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 75 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."""
+    assert '\n' not in name, 'Stream name must not contain a newline.'
     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)
+      assert '\n' not in log_name, 'Log name must not contain a newline.'
       ret = self._engine.LogStream(self, log_name)
       self._logs[log_name] = ret
       return ret
 
     def add_step_text(self, text):
-      pass
+      assert '\n' not in text, 'Step text must not contain a newline.'
 
     def add_step_summary_text(self, text):
-      pass
+      assert '\n' not in text, 'Step summary text must not contain a newline.'
 
     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
+      assert '\n' not in name, 'Link name must not contain a newline.'
+      assert isinstance(url, basestring), 'Link URL %s is not a string' % url
+      assert '\n' not in url, 'Link URL must not contain a newline.'

[iannucci, 2017/04/05 02:40:23]

Since these values are user-controlled (i.e. they vary at runtime), it would be
better not to use `assert` for them (since -O removes asserts in cpython:
`python -O -c 'assert False'` exits 0).

Would you mind switching the asserts in this file to a function (maybe
"check()"?) that raises ValueError instead?

[dnj, 2017/04/05 15:08:07]

Done.

 
     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
+      assert '\n' not in key, 'Property key must not contain a newline.'
+      assert '\n' not in value, 'Property value must not contain a newline.'
+      json.loads(value) # value must be a valud JSON object.

[iannucci, 2017/04/05 02:40:23]

nit: valud

[dnj, 2017/04/05 15:08:07]

Done.

 
     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.
+      assert '\n' not in spec, 'Spec must not contain a newline.'
+      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, 'Inividual 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')