Assert validity of stream parameters.

recipe_engine/unittests/stream_test.py

 #!/usr/bin/env python
 # 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.
 
 import cStringIO
 import unittest
 
 import test_env
 
@@ skipping 143 matching lines @@
       with self.assertRaises(AssertionError):
         foo.set_step_status('SINGLE')
 
   def test_buildbot_status_constraint(self):
     with stream.StreamEngineInvariants() as engine:
       foo = engine.make_step_stream('foo')
       foo.set_step_status('FAILURE')
       with self.assertRaises(AssertionError):
         foo.set_step_status('SUCCESS')
 
+  def test_content_assertions(self):
+    with stream.StreamEngineInvariants() as engine:
+      with self.assertRaises(AssertionError):
+        engine.make_step_stream('foo\nbar')
+
+      # Test StepStream.
+      s = engine.make_step_stream('foo')
+      with self.assertRaises(AssertionError):
+        s.new_log_stream('foo\nbar')
+      ls = s.new_log_stream('foo')
+
+      with self.assertRaises(AssertionError):
+        s.write_line('foo\nbar')
+      s.write_line('foo')
+
+      with self.assertRaises(AssertionError):
+        s.add_step_text('foo\nbar')
+      s.add_step_text('foo')
+
+      with self.assertRaises(AssertionError):
+        s.add_step_summary_text('foo\nbar')
+      s.add_step_summary_text('foo')
+
+      with self.assertRaises(AssertionError):
+        s.add_step_link('foo\nbar', 'baz')
+      with self.assertRaises(AssertionError):
+        s.add_step_link('foo', 'bar\nbaz')
+      s.add_step_link('foo', 'bar')
+
+      with self.assertRaises(AssertionError):
+        s.set_build_property('foo\nbar', 'true')
+      with self.assertRaises(AssertionError):
+        s.set_build_property('foo', 'true\n')
+      with self.assertRaises(ValueError):
+        s.set_build_property('foo', 'NOT JSON')
+      s.set_build_property('foo', '"Is JSON"')
+
+      with self.assertRaises(AssertionError):
+        s.trigger('true\n')
+      with self.assertRaises(ValueError):
+        s.trigger('NOT JSON')
+      s.trigger('"Is JSON"')
+
+      # Test LogStream.
+      with self.assertRaises(AssertionError):
+        ls.write_line('foo\nbar')
+      ls.write_line('foo')
+
+      ls.close()
+      s.close()
+
+
 if __name__ == '__main__':
   unittest.main()