Add CURRENT_TIMESTAMP annotation

scripts/common/unittests/annotator_test.py

 #!/usr/bin/env python
 # Copyright (c) 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Unit tests for classes in annotator.py."""
 
 import cStringIO
 import json
 import types
@@ skipping 12 matching lines @@
 
 class TestAnnotationStreams(unittest.TestCase):
   def setUp(self):
     self.buf = cStringIO.StringIO()
 
   def _getLines(self):
     """Return list of non-empty lines in output."""
     return [line for line in self.buf.getvalue().rstrip().split('\n') if line]
 
   def testBasicUsage(self):
-    stream = annotator.StructuredAnnotationStream(stream=self.buf)
+    stream = annotator.StructuredAnnotationStream(
+        stream=self.buf, time_fn=lambda: 123456)
     with stream.step('one') as _:
       pass
     with stream.step('two') as _:
       pass
 
     result = [
         '@@@SEED_STEP one@@@',
         '@@@STEP_CURSOR one@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_STARTED@@@',
         '@@@STEP_CURSOR one@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_CLOSED@@@',
         '@@@SEED_STEP two@@@',
         '@@@STEP_CURSOR two@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_STARTED@@@',
         '@@@STEP_CURSOR two@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_CLOSED@@@',
     ]
 
     self.assertEquals(result, self._getLines())
 
   def testStepAnnotations(self):
-    stream = annotator.StructuredAnnotationStream(stream=self.buf)
+    stream = annotator.StructuredAnnotationStream(
+        stream=self.buf, time_fn=lambda: 123456)
     with stream.step('one') as s:
       s.step_warnings()
       s.step_failure()
       s.step_exception()
       s.step_clear()
       s.step_summary_clear()
       s.step_text('hello')
       s.step_summary_text('hello!')
       s.step_log_line('mylog', 'test')
       s.step_log_end('mylog')
       s.step_log_line('myperflog', 'perf data')
       s.step_log_end_perf('myperflog', 'dashboardname')
       s.step_link('cool_link', 'https://cool.example.com/beano_gnarly')
       s.write_log_lines('full_log', ['line one', 'line two'])
       s.write_log_lines('full_perf_log', ['perf line one', 'perf line two'],
                         perf='full_perf')
 
     result = [
         '@@@SEED_STEP one@@@',
         '@@@STEP_CURSOR one@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_STARTED@@@',
         '@@@STEP_WARNINGS@@@',
         '@@@STEP_FAILURE@@@',
         '@@@STEP_EXCEPTION@@@',
         '@@@STEP_CLEAR@@@',
         '@@@STEP_SUMMARY_CLEAR@@@',
         '@@@STEP_TEXT@hello@@@',
         '@@@STEP_SUMMARY_TEXT@hello!@@@',
         '@@@STEP_LOG_LINE@mylog@test@@@',
         '@@@STEP_LOG_END@mylog@@@',
         '@@@STEP_LOG_LINE@myperflog@perf data@@@',
         '@@@STEP_LOG_END_PERF@myperflog@dashboardname@@@',
         '@@@STEP_LINK@cool_link@https://cool.example.com/beano_gnarly@@@',
         '@@@STEP_LOG_LINE@full_log@line one@@@',
         '@@@STEP_LOG_LINE@full_log@line two@@@',
         '@@@STEP_LOG_END@full_log@@@',
         '@@@STEP_LOG_LINE@full_perf_log@perf line one@@@',
         '@@@STEP_LOG_LINE@full_perf_log@perf line two@@@',
         '@@@STEP_LOG_END_PERF@full_perf_log@full_perf@@@',
         '@@@STEP_CURSOR one@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_CLOSED@@@',
     ]
 
     self.assertEquals(result, self._getLines())
 
   def testStepAnnotationsWrongParams(self):
     stream = annotator.StructuredAnnotationStream(stream=self.buf)
     with stream.step('one') as s:
       with self.assertRaisesRegexp(TypeError, r'1 argument \(2 given\)'):
         s.step_warnings('bar')
@@ skipping 33 matching lines @@
   def testDupLogs(self):
     stream = annotator.StructuredAnnotationStream(stream=self.buf)
 
     with stream.step('one') as s:
       lines = ['one', 'two']
       s.write_log_lines('mylog', lines)
       self.assertRaises(ValueError, s.write_log_lines, 'mylog', lines)
 
   def testStructured(self):
     stream = annotator.StructuredAnnotationStream(
-        stream=self.buf, flush_before=None)
+        stream=self.buf, flush_before=None, time_fn=lambda: 123456)
     step = annotator.StructuredAnnotationStep(
         annotation_stream=stream, stream=self.buf, flush_before=None)
     stream.step_cursor('one')
     step.step_started()
     stream.step_cursor('two')
     step.step_started()
     stream.step_cursor('one')
     step.step_closed()
     stream.step_cursor('two')
     step.step_closed()
 
     result = [
         '@@@STEP_CURSOR one@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_STARTED@@@',
         '@@@STEP_CURSOR two@@@',
+        '@@@CURRENT_TIMESTAMP@123456@@@',
         '@@@STEP_STARTED@@@',
         '@@@STEP_CURSOR one@@@',
         '@@@STEP_CLOSED@@@',
         '@@@STEP_CURSOR two@@@',
         '@@@STEP_CLOSED@@@',
     ]
 
     self.assertEquals(result, self._getLines())
 
 
@@ skipping 126 matching lines @@
         continue
       implemented.add(name)
       self.assertIsInstance(fn, types.FunctionType)
       expected_num_args = annotator.ALL_ANNOTATIONS[name]
       self.assertEqual(expected_num_args, fn.func_code.co_argcount - 1)
     self.assertSetEqual(required, implemented)
 
 
 if __name__ == '__main__':
   unittest.main()