Add initial postprocess unit test thingy.

recipe_engine/third_party/expect_tests/handle_test.py

 # Copyright 2014 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.
 
 import collections
 import os
 import sys
 import time
 
 from cStringIO import StringIO
 
 from .type_definitions import DirSeen, Handler, Failure
 from .serialize import GetCurrentData, DiffData, NonExistant
 
 
 Missing = collections.namedtuple('Missing', 'test log_lines')
 Fail = collections.namedtuple('Fail', 'test diff log_lines')
 Pass = collections.namedtuple('Pass', 'test')
 
 
+class FailChecks(collections.namedtuple('FailChecks', 'test checks')):
+  def format(self, indent):
+    return (' '*indent+'\n').join([c.format(indent+2) for c in self.checks])
+
+
 class TestHandler(Handler):
   """Run the tests."""
 
   @classmethod
   def gen_stage_loop(cls, _opts, tests, put_next_stage, put_result_stage):
     dirs_seen = set()
     for test in tests:
       subtests = test.tests
       for subtest in subtests:
         if subtest.expect_dir not in dirs_seen:
           put_result_stage(DirSeen(subtest.expect_dir))
           dirs_seen.add(subtest.expect_dir)
       put_next_stage(test)
 
   @classmethod
   def run_stage_loop(cls, _opts, results, put_next_stage):
     for test, result, log_lines in results:
       current, _ = GetCurrentData(test)
       if current is NonExistant:
         put_next_stage(Missing(test, log_lines))
       else:
         diff = DiffData(current, result.data)
         if not diff:
-          put_next_stage(Pass(test))
+          failed_checks = [check for check in result.checks if not check.passed]
+          if failed_checks:
+            put_next_stage(FailChecks(test, failed_checks))
+          else:
+            put_next_stage(Pass(test))
         else:
           put_next_stage(Fail(test, diff, log_lines))
 
   class ResultStageHandler(Handler.ResultStageHandler):
     def __init__(self, *args):
       super(TestHandler.ResultStageHandler, self).__init__(*args)
       self.dirs_seen = set()
       self.files_expected = collections.defaultdict(set)
       self.err_out = StringIO()
       self.start = time.time()
@@ skipping 25 matching lines @@
 
     def handle_DirSeen(self, dirseen):
       self.dirs_seen.add(dirseen.dir)
 
     def _handle_record(self, test):
       self.num_tests += 1
       if test.expect_path() is not None:
         head, tail = os.path.split(test.expect_path())
         self.files_expected[head].add(tail)
 
+    def handle_FailChecks(self, fc):
+      self._handle_record(fc.test)
+      self._emit('C', fc.test, 'FAIL CHECK')
+      self._add_result(fc.format(2), fc.test, 'FAIL CHECK', 'failed checks')
+      return Failure()
+
     def handle_Pass(self, p):
       self._handle_record(p.test)
       if not self.opts.quiet:
         self._emit('.', p.test, 'ok')
 
     def handle_Fail(self, fail):
       self._handle_record(fail.test)
       self._emit('F', fail.test, 'FAIL')
       self._add_result('\n'.join(fail.diff), fail.test, 'FAIL', 'failures',
                        fail.log_lines)
@@ skipping 44 matching lines @@
             self.num_tests, time.time() - self.start)
         print
       if aborted:
         print 'ABORTED'
       elif self.errors:
         print 'FAILED (%s)' % (', '.join('%s=%d' % i
                                          for i in self.errors.iteritems()))
       elif not self.opts.quiet:
         print 'OK'