Recognize crashed tests as failures in sharding_supervisor

tools/sharding_supervisor/sharding_supervisor.py

 #!/usr/bin/env python
 # Copyright (c) 2011 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.
 
 """Shards a given test suite and runs the shards in parallel.
 
 ShardingSupervisor is called to process the command line options and creates
 the specified number of worker threads. These threads then run each shard of
 the test in a separate process and report on the results. When all the shards
@@ skipping 63 matching lines @@
 
 
 class ShardRunner(threading.Thread):
   """Worker thread that manages a single shard at a time.
 
   Attributes:
     supervisor: The ShardingSupervisor that this worker reports to.
     counter: Called to get the next shard index to run.
   """
 
-  def __init__(self, supervisor, counter, test_fail, test_timeout):
+  def __init__(self, supervisor, counter, test_start, test_ok, test_fail,
+               test_timeout):
     """Inits ShardRunner with a supervisor and counter."""
     threading.Thread.__init__(self)
     self.supervisor = supervisor
     self.counter = counter
+    self.test_start = test_start
+    self.test_ok = test_ok
     self.test_fail = test_fail
     self.test_timeout = test_timeout
+    self.current_test = ""
 
-  def SearchForFailure(self, regex, prefix, line, description):
-    results = regex.search(line)
+  def ReportFailure(self, description, prefix, test_name):
+    log_line = "%s: %s%s\n" % (description, prefix, test_name)
+    self.supervisor.LogLineFailure(log_line)
+
+  def ProcessLine(self, prefix, line):
+    results = self.test_start.search(line)
     if results:
-      log_line = "%s: %s%s\n" % (description, prefix, results.group(1))
-      self.supervisor.LogLineFailure(log_line)
-      return True
-    return False
+      if self.current_test:
+        self.ReportFailure("CRASHED", prefix, results.group(1))
+      self.current_test = results.group(1)
+      return
+
+    results = self.test_ok.search(line)
+    if results:
+      self.current_test = ""
+      return
+
+    results = self.test_fail.search(line)
+    if results:
+      self.ReportFailure("FAILED", prefix, results.group(1))
+      self.current_test = ""
+      return
+
+    results = self.test_timeout.search(line)

[Paweł Hajdan Jr., 2011/08/15 17:33:17]

Don't rely on that line to be present. It's fragile.

[clee, 2011/08/15 18:07:09]

It's what gtest_command uses to detect timeouts. There isn't a '[  FAILED  ]'
line for timeouts, so this seems to be the only way to detect one. Or can
timeouts and crashes be grouped together (as 'incomplete' perhaps?), in which
case the current check for 'crashed' would suffice? Based on past experience,
would there be an advantage in having timeouts and crashes be two separate
categories? wdyt? 

[Paweł Hajdan Jr., 2011/08/16 17:36:07]

And guess what? It's broken.
It's not really useful to trying to distinguish between crashes and hangs IMO.

I think that a more reliable solution would be to detect that you see one test
starting, but you never see it finishing: you either encounter EOF or another
test starting. That is a pretty good sign something bad has happened.

+    if results:
+      self.ReportFailure("TIMEOUT", prefix, results.group(1))
+      self.current_test = ""
 
   def run(self):
     """Runs shards and outputs the results.
 
     Gets the next shard index from the supervisor, runs it in a subprocess,
     and collects the output. Each line is prefixed with 'N>', where N is the
     current shard index.
     """
     while True:
       try:
         index = self.counter.get_nowait()
       except Queue.Empty:
         break
       prefix = "%i>" % index
       chars = StringIO()
       shard_running = True
       shard = RunShard(
           self.supervisor.test, self.supervisor.num_shards, index,
           self.supervisor.gtest_args, subprocess.PIPE, subprocess.STDOUT)
       while shard_running:
         char = shard.stdout.read(1)
         if not char and shard.poll() is not None:
           shard_running = False
         chars.write(char)
         if char == "\n" or not shard_running:
           line = chars.getvalue()
           if not line and not shard_running:
             break
-          if not self.SearchForFailure(
-              self.test_fail, prefix, line, "FAILED"):
-            self.SearchForFailure(self.test_timeout, prefix, line, "TIMEOUT")
+          self.ProcessLine(prefix, line)
           line = prefix + line
           self.supervisor.LogOutputLine(index, line)
           chars.close()
           chars = StringIO()
       self.supervisor.ShardIndexCompleted(index)
       if shard.returncode != 0:
         self.supervisor.LogShardFailure(index)
 
 
 class ShardingSupervisor(object):
@@ skipping 38 matching lines @@
     """
 
     # Regular expressions for parsing GTest logs. Test names look like
     #   SomeTestCase.SomeTest
     #   SomeName/SomeTestCase.SomeTest/1
     # This regex also matches SomeName.SomeTest/1 and
     # SomeName/SomeTestCase.SomeTest, which should be harmless.
     test_name_regex = r"((\w+/)?\w+\.\w+(/\d+)?)"
 
     # Regex for filtering out ANSI escape codes when using color.
-    ansi_code_regex = r"(?:\x1b\[.*?[a-zA-Z])?"
+    ansi_regex = r"(?:\x1b\[.*?[a-zA-Z])?"
 
+    test_start = re.compile(
+        ansi_regex + "\[\s+RUN\s+\] " + ansi_regex + test_name_regex)

[M-A Ruel, 2011/08/15 15:46:08]

Use r"".

[clee, 2011/08/15 18:07:09]

Done.

+    test_ok = re.compile(
+        ansi_regex + "\[\s+OK\s+\] " + ansi_regex + test_name_regex)
     test_fail = re.compile(
-        ansi_code_regex + "\[\s+FAILED\s+\] " + ansi_code_regex +
-        test_name_regex)
+        ansi_regex + "\[\s+FAILED\s+\] " + ansi_regex + test_name_regex)
     test_timeout = re.compile(
         "Test timeout \([0-9]+ ms\) exceeded for " + test_name_regex)
 
     workers = []
     counter = Queue.Queue()
     for i in range(self.num_shards):
       counter.put(i)
 
-    # Disable stdout buffering to read shard output one character at a time.
-    # This allows for shard crashes that do not end with a "\n".
-    sys.stdout = os.fdopen(sys.stdout.fileno(), "w", 0)
-
     for i in range(self.num_runs):
-      worker = ShardRunner(self, counter, test_fail, test_timeout)
+      worker = ShardRunner(
+          self, counter, test_start, test_ok, test_fail, test_timeout)
       worker.start()
       workers.append(worker)
     if self.reorder:
       self.WaitForShards()
     else:
       for worker in workers:
         worker.join()
 
-    # Re-enable stdout buffering.
-    sys.stdout = sys.__stdout__
-
-    return self.PrintSummary()
+    return self.PrintSummary(self.failure_log)
 
   def LogLineFailure(self, line):
     """Saves a line in the failure log to be printed at the end.
 
     Args:
       line: The line to save in the failure_log.
     """
     if line not in self.failure_log:
       self.failure_log.append(line)
 
@@ skipping 15 matching lines @@
 
   def LogOutputLine(self, index, line):
     if self.reorder:
       self.shard_output[index].put(line)
     else:
       sys.stdout.write(line)
 
   def ShardIndexCompleted(self, index):
     self.shard_output[index].put(self.SHARD_COMPLETED)
 
-  def PrintSummary(self):
+  def PrintSummary(self, failed_tests):
     """Prints a summary of the test results.
 
     If any shards had failing tests, the list is sorted and printed. Then all
     the lines that indicate a test failure are reproduced.
 
     Returns:
       The number of shards that had failing tests.
     """
     sys.stderr.write("\n")
     num_failed = len(self.failed_shards)
     if num_failed > 0:
       self.failed_shards.sort()
-      if self.color:
-        sys.stderr.write("\x1b[1;5;31m")
-      sys.stderr.write("SHARDS THAT FAILED: %s\n" % str(self.failed_shards))
+      self.WriteText(sys.stderr,
+                     "FAILED SHARDS: %s\n" % str(self.failed_shards),
+                     "\x1b[1;5;31m")
     else:
-      if self.color:
-        sys.stderr.write("\x1b[1;5;32m")
-      sys.stderr.write("ALL SHARDS PASSED!\n")
-    if self.failure_log:
-      if self.color:
-        sys.stderr.write("\x1b[1;5;31m")
-      sys.stderr.write("TESTS THAT DID NOT PASS:\n")
-      if self.color:
-        sys.stderr.write("\x1b[m")
-      for line in self.failure_log:
+      self.WriteText(sys.stderr, "ALL SHARDS PASSED!\n", "\x1b[1;5;32m")
+    if failed_tests:
+      self.WriteText(sys.stderr, "FAILED TESTS:\n", "\x1b[1;5;31m")
+      for line in failed_tests:
         sys.stderr.write(line)
     if self.color:
       sys.stderr.write("\x1b[m")
     return num_failed
 
+  def WriteText(self, pipe, text, ansi):
+    if self.color:
+      pipe.write(ansi)
+    pipe.write(text)
+    if self.color:
+      pipe.write("\x1b[m")
+
 
 def main():
   parser = optparse.OptionParser(usage=SS_USAGE)
   parser.add_option(
       "-n", "--shards_per_core", type="int", default=SS_DEFAULT_SHARDS_PER_CORE,
       help="number of shards to generate per CPU")
   parser.add_option(
       "-r", "--runs_per_core", type="int", default=SS_DEFAULT_RUNS_PER_CORE,
       help="number of shards to run in parallel per CPU")
   parser.add_option(
@@ skipping 47 matching lines @@
 
   # shard and run the whole test
   ss = ShardingSupervisor(args[0], num_shards, num_runs, options.color,
                           options.reorder, gtest_args)
   return ss.ShardTest()
 
 
 if __name__ == "__main__":
   sys.exit(main())