merge with HEAD

Author: google-yfyang

test/color_runner.py

@@ -35,37 +35,13 @@ def _restoreStdout(self):
 class ProgressMixin:
   test_count = 0
   progress_counter = 0
-  fail_count = 0
-
-  def addError(self, test, err):
-    self.fail_count += 1
-    super().addError(test, err)
-
-  def addFailure(self, test, err):
-    self.fail_count += 1
-    super().addFailure(test, err)
-
-  def addUnexpectedSuccess(self, test):
-    self.fail_count += 1
-    super().addUnexpectedSuccess(test)
-
-  def writeProgressPrefix(self):
-    if self.showAll:
-      base_progress = f'[{self.progress_counter}/{self.test_count}'
-      rtn = len(base_progress)
-      progress = with_color(CYAN, base_progress)
-      if self.fail_count:
-        fail_str = f'-{self.fail_count}'
-        progress += with_color(RED, fail_str)
-        rtn += len(fail_str)
-      progress += with_color(CYAN, '] ')
-      self.stream.write(progress)
-      return rtn + 2
-    return 0
 
   def startTest(self, test):
+    assert self.test_count > 0
     self.progress_counter += 1
-    self.writeProgressPrefix()
+    if self.showAll:
+      progress = f'[{self.progress_counter}/{self.test_count}] '
+      self.stream.write(with_color(CYAN, progress))
     super().startTest(test)
 
 
@@ -86,9 +62,6 @@ class ColorTextRunner(unittest.TextTestRunner):
   """Subclass of TextTestRunner that uses ColorTextResult"""
   resultclass = ColorTextResult # type: ignore
 
-  def __init__(self, *args, **kwargs):
-    super().__init__(*args, verbosity=2, **kwargs)
-
   def _makeResult(self):
     result = super()._makeResult()
     result.test_count = self.test_count

test/parallel_testsuite.py

@@ -115,6 +115,37 @@ def addTest(self, test):
     super().addTest(test)
     test.is_parallel = True
 
+  def printOneResult(self, res):
+    self.progress_counter += 1
+    progress = f'[{self.progress_counter}/{self.num_tests}] '
+
+    if res.test_result == 'success':
+      msg = 'ok'
+      color = GREEN
+    elif res.test_result == 'errored':
+      msg = 'ERROR'
+      color = RED
+    elif res.test_result == 'failed':
+      msg = 'FAIL'
+      color = RED
+    elif res.test_result == 'skipped':
+      reason = res.skipped[0][1]
+      msg = f"skipped '{reason}'"
+      color = CYAN
+    elif res.test_result == 'unexpected success':
+      msg = 'unexpected success'
+      color = RED
+    elif res.test_result == 'expected failure':
+      color = RED
+      msg = 'expected failure'
+    else:
+      assert False, f'unhandled test result {res.test_result}'
+
+    if res.test_result != 'skipped':
+      msg += f' ({res.elapsed:.2f}s)'
+
+    errlog(f'{with_color(CYAN, progress)}{res.test} ... {with_color(color, msg)}')
+
   def run(self, result):
     # The 'spawn' method is used on windows and it can be useful to set this on
     # all platforms when debugging multiprocessing issues.  Without this we

test/runner.py

@@ -429,16 +429,12 @@ def run_tests(options, suites):
     testRunner = xmlrunner.XMLTestRunner(output=output, verbosity=2,
                                          failfast=options.failfast)
     print('Writing XML test output to ' + os.path.abspath(output.name))
-  elif options.ansi and not options.verbose:
-    # When not in verbose mode and ansi color output is available use our nice single-line
+  elif options.buffer and options.ansi and not options.verbose:
+    # When buffering is enabled and ansi color output is available use our nice single-line
     # result display.
-    testRunner = SingleLineTestRunner(failfast=options.failfast)
+    testRunner = SingleLineTestRunner(verbosity=2, failfast=options.failfast)
   else:
-    if not options.ansi:
-      print('using verbose test runner (ANSI not avilable)')
-    else:
-      print('using verbose test runner (verbose output requested)')
-    testRunner = ColorTextRunner(failfast=options.failfast)
+    testRunner = ColorTextRunner(verbosity=2, failfast=options.failfast)
 
   total_core_time = 0
   run_start_time = time.perf_counter()
@@ -476,9 +472,7 @@ def parse_args():
                            'test.  Implies --cores=1.  Defaults to true when running a single test')
   parser.add_argument('--no-clean', action='store_true',
                       help='Do not clean the temporary directory before each test run')
-  parser.add_argument('--verbose', '-v', action='count', default=0,
-                      help="Show test stdout and stderr, and don't use the single-line test reporting. "
-                           'Specifying `-v` twice will enable test framework logging (i.e. EMTEST_VERBOSE)')
+  parser.add_argument('--verbose', '-v', action='store_true')
   # TODO: Replace with BooleanOptionalAction once we can depend on python3.9
   parser.add_argument('--ansi', action='store_true', default=None)
   parser.add_argument('--no-ansi', action='store_false', dest='ansi', default=None)

test/single_line_runner.py

@@ -6,7 +6,7 @@
 import shutil
 import unittest
 
-from color_runner import ColorTextResult, ColorTextRunner
+from color_runner import BufferingMixin, ColorTextRunner
 
 from tools.colored_logger import CYAN, GREEN, RED, with_color
 
@@ -20,47 +20,65 @@ def term_width():
   return shutil.get_terminal_size()[0]
 
 
-class SingleLineTestResult(ColorTextResult):
+class SingleLineTestResult(BufferingMixin, unittest.TextTestResult):
   """Similar to the standard TextTestResult but uses ANSI escape codes
   for color output and reusing a single line on the terminal.
   """
 
-  def writeStatus(self, test, msg, color, line_pos):
-    # Because the message can include the skip reason (which can be very long sometimes), truncate
-    # it to a reasonable length to avoid exceeding line length.
-    if len(msg) > 30:
-      msg = msg[:30]
-    # Format the line so that it fix within the terminal width, unless its less then min_len
+  def __init__(self, *args, **kwargs):
+    super().__init__(*args, **kwargs)
+    self.progress_counter = 0
+
+  def writeStatusLine(self, line):
+    clearline(self._original_stderr)
+    self._original_stderr.write(line)
+    self._original_stderr.flush()
+
+  def updateStatus(self, test, msg, color):
+    progress = f'[{self.progress_counter}/{self.test_count}] '
+    # Format the line so that it fix within the terminal width, unless it's less then min_len
     # in which case there is not much we can do, and we just overflow the line.
-    min_len = line_pos + len(msg) + 5
+    min_len = len(progress) + len(msg) + 5
     test_name = str(test)
     if term_width() > min_len:
       max_name = term_width() - min_len
       test_name = test_name[:max_name]
-    line = f'{test_name} ... {with_color(color, msg)}'
-    self._original_stderr.write(line)
-
-  def _write_status(self, test, status):
-    clearline(self._original_stderr)
-    pos = self.writeProgressPrefix()
-    # Add some color to the status message
-    if status == 'ok':
-      color = GREEN
-    elif status.isupper():
-      color = RED
-      # Use a newline when a test fails, so you can see a list of failures while
-      # the other tests are still running
-      status += '\n'
-    else:
-      color = CYAN
-    self.writeStatus(test, status, color, pos)
-    self._original_stderr.flush()
+    line = f'{with_color(CYAN, progress)}{test_name} ... {with_color(color, msg)}'
+    self.writeStatusLine(line)
 
   def startTest(self, test):
     self.progress_counter += 1
-    # We explictly don't call TextTestResult.startTest here since we delay all printing
-    # of results until `_write_status`
+    assert self.test_count > 0
+    # Note: We explicitly do not use `super()` here but instead call `unittest.TestResult`. i.e.
+    # we skip the superclass (since we don't want its specific behaviour) and instead call its
+    # superclass.
     unittest.TestResult.startTest(self, test)
+    if self.progress_counter == 1:
+      self.updateStatus(test, '', GREEN)
+
+  def addSuccess(self, test):
+    unittest.TestResult.addSuccess(self, test)
+    self.updateStatus(test, 'ok', GREEN)
+
+  def addFailure(self, test, err):
+    unittest.TestResult.addFailure(self, test, err)
+    self.updateStatus(test, 'FAIL', RED)
+
+  def addError(self, test, err):
+    unittest.TestResult.addError(self, test, err)
+    self.updateStatus(test, 'ERROR', RED)
+
+  def addExpectedFailure(self, test, err):
+    unittest.TestResult.addExpectedFailure(self, test, err)
+    self.updateStatus(test, 'expected failure', RED)
+
+  def addUnexpectedSuccess(self, test, err):
+    unittest.TestResult.addUnexpectedSuccess(self, test, err)
+    self.updateStatus(test, 'UNEXPECTED SUCCESS', RED)
+
+  def addSkip(self, test, reason):
+    unittest.TestResult.addSkip(self, test, reason)
+    self.updateStatus(test, f"skipped '{reason}'", CYAN)
 
   def printErrors(self):
     # All tests have been run at this point so print a final newline