--- a/testing/mozbase/mozlog/mozlog/commandline.py
+++ b/testing/mozbase/mozlog/mozlog/commandline.py
@@ -14,16 +14,17 @@ from .structuredlog import StructuredLog
log_formatters = {
'raw': (formatters.JSONFormatter, "Raw structured log messages"),
'unittest': (formatters.UnittestFormatter, "Unittest style output"),
'xunit': (formatters.XUnitFormatter, "xUnit compatible XML"),
'html': (formatters.HTMLFormatter, "HTML report"),
'mach': (formatters.MachFormatter, "Human-readable output"),
'tbpl': (formatters.TbplFormatter, "TBPL style log format"),
+ 'grouped': (formatters.GroupingFormatter, "Grouped summary of test results"),
'errorsummary': (formatters.ErrorSummaryFormatter, argparse.SUPPRESS),
}
TEXT_FORMATTERS = ('raw', 'mach')
"""a subset of formatters for non test harnesses related applications"""
def level_filter_wrapper(formatter, level):
return handlers.LogLevelFilter(formatter, level)
new file mode 100644
--- /dev/null
+++ b/testing/mozbase/mozlog/mozlog/formatters/grouping.py
@@ -0,0 +1,285 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+import collections
+import os
+import sys
+import subprocess
+
+from . import base
+
+DEFAULT_MOVE_UP_CODE = u"\x1b[A"
+DEFAULT_CLEAR_EOL_CODE = u"\x1b[K"
+
+
+class GroupingFormatter(base.BaseFormatter):
+ """Formatter designed to produce unexpected test results grouped
+ together in a readable format."""
+ def __init__(self):
+ self.number_of_tests = 0
+ self.completed_tests = 0
+ self.need_to_erase_last_line = False
+ self.current_display = ""
+ self.running_tests = {}
+ self.test_output = collections.defaultdict(str)
+ self.subtest_failures = collections.defaultdict(list)
+ self.test_failure_text = ""
+ self.tests_with_failing_subtests = []
+ self.interactive = os.isatty(sys.stdout.fileno())
+
+ # TODO(mrobinson, 8313): We need to add support for Windows terminals here.
+ if self.interactive:
+ self.line_width = int(subprocess.check_output(['stty', 'size']).split()[1])
+ self.move_up, self.clear_eol = self.get_move_up_and_clear_eol_codes()
+
+ self.expected = {
+ 'OK': 0,
+ 'PASS': 0,
+ 'FAIL': 0,
+ 'ERROR': 0,
+ 'TIMEOUT': 0,
+ 'SKIP': 0,
+ 'CRASH': 0,
+ }
+
+ self.unexpected_tests = {
+ 'OK': [],
+ 'PASS': [],
+ 'FAIL': [],
+ 'ERROR': [],
+ 'TIMEOUT': [],
+ 'CRASH': [],
+ }
+
+ def get_move_up_and_clear_eol_codes(self):
+ try:
+ import blessings
+ except ImportError:
+ return DEFAULT_MOVE_UP_CODE, DEFAULT_CLEAR_EOL_CODE
+
+ try:
+ self.terminal = blessings.Terminal()
+ return self.terminal.move_up, self.terminal.clear_eol
+ except Exception as exception:
+ sys.stderr.write("GroupingFormatter: Could not get terminal "
+ "control characters: %s\n" % exception)
+ return DEFAULT_MOVE_UP_CODE, DEFAULT_CLEAR_EOL_CODE
+
+ def text_to_erase_display(self):
+ if not self.interactive or not self.current_display:
+ return ""
+ return ((self.move_up + self.clear_eol) *
+ self.current_display.count('\n'))
+
+ def generate_output(self, text=None, new_display=None):
+ if not self.interactive:
+ return text
+
+ output = self.text_to_erase_display()
+ if text:
+ output += text
+ if new_display is not None:
+ self.current_display = new_display
+ return output + self.current_display
+
+ def build_status_line(self):
+ if self.number_of_tests == 0:
+ new_display = " [%i] " % self.completed_tests
+ else:
+ new_display = " [%i/%i] " % (self.completed_tests, self.number_of_tests)
+
+ if self.running_tests:
+ indent = " " * len(new_display)
+ if self.interactive:
+ max_width = self.line_width - len(new_display)
+ else:
+ max_width = sys.maxsize
+ return new_display + ("\n%s running: " % indent).join(
+ val[:max_width] for val in self.running_tests.values()) + "\n"
+ else:
+ return ""
+
+ def suite_start(self, data):
+ self.number_of_tests = len(data["tests"])
+ self.start_time = data["time"]
+
+ if self.number_of_tests == 0:
+ return "Running tests in %s\n\n" % data[u'source']
+ else:
+ return "Running %i tests in %s\n\n" % (self.number_of_tests, data[u'source'])
+
+ def test_start(self, data):
+ self.running_tests[data['thread']] = data['test']
+ return self.generate_output(text=None,
+ new_display=self.build_status_line())
+
+ def wrap_and_indent_lines(self, lines, indent):
+ assert(len(lines) > 0)
+
+ output = indent + u"\u25B6 %s\n" % lines[0]
+ for line in lines[1:-1]:
+ output += indent + u"\u2502 %s\n" % line
+ if len(lines) > 1:
+ output += indent + u"\u2514 %s\n" % lines[-1]
+ return output
+
+ def get_lines_for_unexpected_result(self,
+ test_name,
+ status,
+ expected,
+ message,
+ stack):
+ # Test names sometimes contain control characters, which we want
+ # to be printed in their raw form, and not their interpreted form.
+ test_name = test_name.encode('unicode-escape')
+
+ if expected:
+ expected_text = u" [expected %s]" % expected
+ else:
+ expected_text = u""
+
+ lines = [u"%s%s %s" % (status, expected_text, test_name)]
+ if message:
+ lines.append(u" \u2192 %s" % message)
+ if stack:
+ lines.append("")
+ lines += [stackline for stackline in stack.splitlines()]
+ return lines
+
+ def get_output_for_unexpected_subtests(self, test_name, unexpected_subtests):
+ if not unexpected_subtests:
+ return ""
+
+ def add_subtest_failure(lines, subtest, stack=None):
+ lines += self.get_lines_for_unexpected_result(
+ subtest.get('subtest', None),
+ subtest.get('status', None),
+ subtest.get('expected', None),
+ subtest.get('message', None),
+ stack)
+
+ def make_subtests_failure(test_name, subtests, stack=None):
+ lines = [u"Unexpected subtest result in %s:" % test_name]
+ for subtest in subtests[:-1]:
+ add_subtest_failure(lines, subtest, None)
+ add_subtest_failure(lines, subtests[-1], stack)
+ return self.wrap_and_indent_lines(lines, " ") + "\n"
+
+ # Organize the failures by stack trace so we don't print the same stack trace
+ # more than once. They are really tall and we don't want to flood the screen
+ # with duplicate information.
+ output = ""
+ failures_by_stack = collections.defaultdict(list)
+ for failure in unexpected_subtests:
+ # Print stackless results first. They are all separate.
+ if 'stack' not in failure:
+ output += make_subtests_failure(test_name, [failure], None)
+ else:
+ failures_by_stack[failure['stack']].append(failure)
+
+ for (stack, failures) in failures_by_stack.iteritems():
+ output += make_subtests_failure(test_name, failures, stack)
+ return output
+
+ def test_end(self, data):
+ self.completed_tests += 1
+ test_status = data["status"]
+ test_name = data["test"]
+ had_unexpected_test_result = "expected" in data
+ subtest_failures = self.subtest_failures.pop(test_name, [])
+
+ del self.running_tests[data['thread']]
+ new_display = self.build_status_line()
+
+ if not had_unexpected_test_result and not subtest_failures:
+ self.expected[test_status] += 1
+ if self.interactive:
+ return self.generate_output(text=None, new_display=new_display)
+ else:
+ return self.generate_output(text=" %s\n" % test_name,
+ new_display=new_display)
+
+ # If the test crashed or timed out, we also include any process output,
+ # because there is a good chance that the test produced a stack trace
+ # or other error messages.
+ if test_status in ("CRASH", "TIMEOUT"):
+ stack = self.test_output[test_name] + data.get('stack', "")
+ else:
+ stack = data.get('stack', None)
+
+ output = ""
+ if had_unexpected_test_result:
+ self.unexpected_tests[test_status].append(data)
+ lines = self.get_lines_for_unexpected_result(
+ test_name,
+ test_status,
+ data.get('expected', None),
+ data.get('message', None),
+ stack)
+ output += self.wrap_and_indent_lines(lines, " ") + "\n"
+
+ if subtest_failures:
+ self.tests_with_failing_subtests.append(test_name)
+ output += self.get_output_for_unexpected_subtests(test_name,
+ subtest_failures)
+ self.test_failure_text += output
+
+ return self.generate_output(text=output, new_display=new_display)
+
+ def test_status(self, data):
+ if "expected" in data:
+ self.subtest_failures[data["test"]].append(data)
+
+ def suite_end(self, data):
+ self.end_time = data["time"]
+
+ if not self.interactive:
+ output = u"\n"
+ else:
+ output = ""
+
+ output += u"Ran %i tests finished in %.1f seconds.\n" % (
+ self.completed_tests, (self.end_time - self.start_time) / 1000)
+ output += u" \u2022 %i ran as expected. %i tests skipped.\n" % (
+ sum(self.expected.values()), self.expected['SKIP'])
+
+ def text_for_unexpected_list(text, section):
+ tests = self.unexpected_tests[section]
+ if not tests:
+ return u""
+ return u" \u2022 %i tests %s\n" % (len(tests), text)
+
+ output += text_for_unexpected_list(u"crashed unexpectedly", 'CRASH')
+ output += text_for_unexpected_list(u"had errors unexpectedly", 'ERROR')
+ output += text_for_unexpected_list(u"failed unexpectedly", 'FAIL')
+ output += text_for_unexpected_list(u"timed out unexpectedly", 'TIMEOUT')
+ output += text_for_unexpected_list(u"passed unexpectedly", 'PASS')
+ output += text_for_unexpected_list(u"unexpectedly okay", 'OK')
+
+ num_with_failing_subtests = len(self.tests_with_failing_subtests)
+ if num_with_failing_subtests:
+ output += (u" \u2022 %i tests had unexpected subtest results\n"
+ % num_with_failing_subtests)
+ output += "\n"
+
+ # Repeat failing test output, so that it is easier to find, since the
+ # non-interactive version prints all the test names.
+ if not self.interactive and self.test_failure_text:
+ output += u"Tests with unexpected results:\n" + self.test_failure_text
+
+ return self.generate_output(text=output, new_display="")
+
+ def process_output(self, data):
+ if data['thread'] not in self.running_tests:
+ return
+ test_name = self.running_tests[data['thread']]
+ self.test_output[test_name] += data['data'] + "\n"
+
+ def log(self, data):
+ # We are logging messages that begin with STDERR, because that is how exceptions
+ # in this formatter are indicated.
+ if data['message'].startswith('STDERR'):
+ return self.generate_output(text=data['message'] + "\n")
+
+ if data['level'] in ('CRITICAL', 'ERROR'):
+ return self.generate_output(text=data['message'] + "\n")