Parse the test executable output on the fly.

run_test_cases.py

 #!/usr/bin/env python
 # Copyright (c) 2012 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.
 
 """Runs each test cases as a single shard, single process execution.
 
 Similar to sharding_supervisor.py but finer grained. It runs each test case
 individually instead of running per shard. Runs multiple instances in parallel.
 """
@@ skipping 725 matching lines @@
           test_case_data['duration'] = float(match.group(1)) / 1000.
         test_case_data['returncode'] = int(line.startswith(FAILED_PREFIX))
         yield test_case_data
         test_case = None
         test_case_data = None
 
   if test_case_data:
     assert not terminated_early
     # This means the last one likely crashed.
     test_case_data['crashed'] = True
+    test_case_data['duration'] = 0
     test_case_data['returncode'] = 1
     yield test_case_data
   elif terminated_early:
     for _ in lines:
       # Exhaust the generator.
       pass
 
   # If test_cases is not empty, these test cases were not run.
   for t in test_cases:
     yield {
@@ skipping 27 matching lines @@
         # that ran if the startup cost was above 10ms.
         if startup_duration > 0.01 and data_ran:
           distributed_duration = startup_duration / len(data_ran)
           for i in data_ran:
             i['duration'] += distributed_duration
         i['returncode'] = returncode or i['returncode']
       break
   return data
 
 
+def convert_to_lines(generator):
+  """Turn input coming from a generator into lines.
+
+  It is Windows-friendly.
+  """
+  accumulator = ''
+  for data in generator:
+    items = (accumulator + data).splitlines(True)
+    for item in items[:-1]:
+      yield item
+    if items[-1].endswith(('\r', '\n')):
+      yield items[-1]
+      accumulator = ''
+    else:
+      accumulator = items[-1]
+  if accumulator:
+    yield accumulator
+
+
 def chromium_filter_tests(data):
-  """Removes funky PRE_ chromium-specific tests."""
-  return [d for d in data if chromium_valid(d['test_case'], False, True)]
+  """Returns a generator that removes funky PRE_ chromium-specific tests."""
+  return (d for d in data if chromium_valid(d['test_case'], False, True))
 
 
 class Runner(object):
   """Immutable settings to run many test cases in a loop."""
   def __init__(
       self, cmd, cwd_dir, timeout, progress, retries, decider, verbose,
       add_task, add_serial_task):
     self.cmd = cmd[:]
     self.cwd_dir = cwd_dir
     self.timeout = timeout
@@ skipping 21 matching lines @@
     if '--gtest_print_time' not in cmd:
       cmd.append('--gtest_print_time')
 
     # TODO(maruel): Use a distribution model.
     timeout = self.timeout * len(test_cases)
 
     if self.verbose > 1:
       self.progress.update_item('Starting command %s with a timeout of %ss' %
                                 (cmd, timeout), False, False)
 
-    output, _, returncode, duration = call_with_timeout(
+    # TODO(maruel): Differentiate between soft and hard timeouts.
+    proc = Popen(
         cmd,
-        timeout,
         cwd=self.cwd_dir,
+        stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
         env=self.env)
-
-    if self.verbose > 1:
-      self.progress.update_item('Command %s finished after %ss' % (cmd,
-                                                                   duration),
-                                False, False)
-
+    # Create a pipeline of generators.
+    gen_lines = convert_to_lines(data for _, data in proc.yield_any(timeout))
     # It needs to be valid utf-8 otherwise it can't be stored.
     # TODO(maruel): Be more intelligent than decoding to ascii.
-    utf8_output = output.decode('ascii', 'ignore').encode('utf-8')
+    gen_lines_utf8 = (
+      line.decode('ascii', 'ignore').encode('utf-8') for line in gen_lines)
+    gen_test_cases = process_output(gen_lines_utf8, test_cases)
+    gen_test_cases_filtered = chromium_filter_tests(gen_test_cases)
 
-    data = process_output(utf8_output.splitlines(True), test_cases)
-    data = normalize_testing_time(data, duration, returncode)
-    data = chromium_filter_tests(data)
-
-    if sys.platform == 'win32':
-      output = output.replace('\r\n', '\n')
-
-    for i in data:
+    data = []
+    for i in gen_test_cases_filtered:
+      if i['duration'] is None:
+        continue
+      # A new test_case completed.
+      data.append(i)
       self.decider.got_result(i['returncode'] == 0)
       need_to_retry = i['returncode'] != 0 and try_count < self.retries
       if try_count:
         line = '%s (%.2fs) - retry #%d' % (
             i['test_case'], i['duration'] or 0, try_count)
       else:
         line = '%s (%.2fs)' % (i['test_case'], i['duration'] or 0)
       if self.verbose or i['returncode'] != 0 or try_count > 0:
         # Print output in one of three cases:
         #   --verbose was specified.
         #   The test failed.
         #   The wasn't the first attempt (this is needed so the test parser can
         #       detect that a test has been successfully retried).
         if i['output']:
           line += '\n' + i['output']
       self.progress.update_item(line, True, need_to_retry)
 
       if need_to_retry:
         if try_count + 1 < self.retries:
           # The test failed and needs to be retried normally.
           # Leave a buffer of ~40 test cases before retrying.
           priority += 40
           self.add_task(
               priority, self.map, priority, [i['test_case']], try_count + 1)
         else:
           # This test only has one retry left, so the final retry should be
           # done serially.
           self.add_serial_task(
               priority, self.map, priority, [i['test_case']], try_count + 1)
-    return data
+
+    if self.verbose > 1:
+      self.progress.update_item(
+          'Command %s finished after %ss' % (cmd, proc.duration()),
+          False, False)
+
+    return normalize_testing_time(data, proc.duration(), proc.returncode)
 
 
 def get_test_cases(
     cmd, cwd, whitelist, blacklist, index, shards, seed, disabled, fails, flaky,
     manual):
   """Returns the filtered list of test cases.
 
   This is done synchronously.
   """
   try:
@@ skipping 670 matching lines @@
         options.gtest_output,
         result_file,
         options.verbose)
   except Failure as e:
     print >> sys.stderr, e.args[0]
     return 1
 
 
 if __name__ == '__main__':
   sys.exit(main(sys.argv[1:]))