Annotation protobuf code can now dump final.

recipe_engine/unittests/stream_logdog_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 collections
 import contextlib
 import datetime
 import json
+import os
+import shutil
+import tempfile
 import threading
 import time
 import unittest
 import StringIO
 
 import test_env
 
 import libs.logdog.stream
 import libs.logdog.varint
 from google.protobuf import json_format as jsonpb
 from recipe_engine import recipe_api
 from recipe_engine import stream
 from recipe_engine import stream_logdog
 
 
 import annotations_pb2 as pb
 
 
+@contextlib.contextmanager
+def tempdir():
+  tdir = tempfile.mkdtemp(suffix='stream_logdog_test', dir=test_env.BASE_DIR)
+  try:
+    yield tdir
+  finally:
+    shutil.rmtree(tdir)
+
+
 def _translate_annotation_datagram(dg):
   """Translate annotation datagram binary data into a Python dict modeled after
   the JSONPB projection of the datagram.
 
   This is chosen because it allows for easy idiomatic equality assertions in
   test cases.
 
   Args:
     dg (str): The serialized annotation pb.Step datagram.
   """
@@ skipping 359 matching lines @@
               {u'step': {
                 u'name': u'bar',
                 u'status': u'SUCCESS',
                 u'started': u'2106-06-12T01:02:06Z',
                 u'ended': u'2106-06-12T01:02:07Z',
               }},
             ],
         },
     })
 
+  def testDumpFinalState(self):
+    self.env.argv = ['fake_program', 'arg0', 'arg1']
+    self.env.environ['foo'] = 'bar'
+    self.env.cwd = 'CWD'
+
+    # Create a StreamEngine with an update interval that will trigger each time
+    # _advance_time is called.
+    with tempdir() as tdir:
+      dump_path = os.path.join(tdir, 'dump.bin')
+      with self._new_stream_engine(
+          update_interval=datetime.timedelta(seconds=1),
+          dump_path=dump_path) as se:
+        # Initial stream state (no steps).
+        self.assertEqual(self.client.all_streams(), {
+            u'annotations': {
+                u'name': u'steps',
+                u'started': u'2106-06-12T01:02:03Z',
+                u'command': {
+                  u'commandLine': [u'fake_program', u'arg0', u'arg1'],
+                  u'cwd': u'CWD',
+                  u'environ': {u'foo': u'bar'},
+                },
+            },
+        })
+
+      with open(dump_path, 'rb') as fd:
+        step = _translate_annotation_datagram(fd.read())
+      self.assertEqual(step, {
+            u'name': u'steps',
+            u'status': u'SUCCESS',
+            u'started': u'2106-06-12T01:02:03Z',
+            u'ended': u'2106-06-12T01:02:04Z',
+            u'command': {
+              u'commandLine': [u'fake_program', u'arg0', u'arg1'],
+              u'cwd': u'CWD',
+              u'environ': {u'foo': u'bar'},
+            },
+        })
+
   def testBasicStream(self):
     self.env.argv = ['fake_program', 'arg0', 'arg1']
     self.env.environ['foo'] = 'bar'
     self.env.cwd = 'CWD'
 
     with self._new_stream_engine(name_base='test/base') as se:
       with self._step_stream(se,
           name='first step',
           cmd=['first', 'step'],
           cwd='FIRST_CWD') as step:
@@ skipping 554 matching lines @@
     sn = sn.append('#!!! stream name !!!')
     self.assertEqual(str(sn), 'base/s______stream_name____')
 
   def testAugmentInvalidStreamNameNormalizes(self):
     sn = stream_logdog._StreamName('base')
     self.assertEqual(str(sn.augment(' !!! other !!! ')), 'base_____other_____')
 
 
 if __name__ == '__main__':
   unittest.main()