# 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 time
from functools import reduce
from mozterm import Terminal
from ..handlers import SummaryHandler
from . import base
from .process import strstatus
from .tbplformatter import TbplFormatter
color_dict = {
"log_test_status_fail": "red",
"log_process_output": "blue",
"log_test_status_pass": "green",
"log_test_status_unexpected_fail": "red",
"log_test_status_known_intermittent": "yellow",
"time": "cyan",
"action": "yellow",
"pid": "cyan",
"heading": "bold_yellow",
"sub_heading": "yellow",
"error": "red",
"warning": "yellow",
"bold": "bold",
"grey": "grey",
"normal": "normal",
"bright_black": "bright_black",
}
DEFAULT = "\x1b(B\x1b[m"
def format_seconds(total):
"""Format number of seconds to MM:SS.DD form."""
minutes, seconds = divmod(total, 60)
return "%2d:%05.2f" % (minutes, seconds)
class TerminalColors(object):
def __init__(self, term, color_dict):
for key, value in color_dict.items():
attribute = getattr(term, value)
# In Blessed, these attributes aren't always callable. We can assume
# that if they're not, they're just the raw ANSI Escape Sequences.
# This TerminalColors class is basically just a lookup table for
# what function to call to format/color an input string a certain way.
# So if the attribute above is a callable, we can just proceed, but
# if it's not, we need to create our own function that prepends the
# raw ANSI Escape Sequences to the input string, so that everything
# has the same behavior. We append DEFAULT to reset to no formatting
# at the end of our string, to prevent text that comes afterwards
# from inheriting the prepended formatting.
if not callable(attribute):
def apply_formatting(text):
return attribute + text + DEFAULT
attribute = apply_formatting
setattr(self, key, attribute)
class MachFormatter(base.BaseFormatter):
def __init__(
self,
start_time=None,
write_interval=False,
write_times=True,
terminal=None,
disable_colors=False,
summary_on_shutdown=False,
verbose=False,
enable_screenshot=False,
**kwargs
):
super(MachFormatter, self).__init__(**kwargs)
if start_time is None:
start_time = time.time()
start_time = int(start_time * 1000)
self.start_time = start_time
self.write_interval = write_interval
self.write_times = write_times
self.status_buffer = {}
self.has_unexpected = {}
self.last_time = None
self.color_formatter = TerminalColors(
Terminal(disable_styling=disable_colors), color_dict
)
self.verbose = verbose
self._known_pids = set()
self.tbpl_formatter = None
self.enable_screenshot = enable_screenshot
self.summary = SummaryHandler()
self.summary_on_shutdown = summary_on_shutdown
message_handlers = {
"colors": {
"on": self._enable_colors,
"off": self._disable_colors,
},
"summary_on_shutdown": {
"on": self._enable_summary_on_shutdown,
"off": self._disable_summary_on_shutdown,
},
}
for topic, handlers in message_handlers.items():
self.message_handler.register_message_handlers(topic, handlers)
def __call__(self, data):
self.summary(data)
s = super(MachFormatter, self).__call__(data)
if s is None:
return
time = self.color_formatter.time(format_seconds(self._time(data)))
return "%s %s\n" % (time, s)
def _enable_colors(self):
self.disable_colors = False
def _disable_colors(self):
self.disable_colors = True
def _enable_summary_on_shutdown(self):
self.summary_on_shutdown = True
def _disable_summary_on_shutdown(self):
self.summary_on_shutdown = False
def _get_test_id(self, data):
test_id = data.get("test")
if isinstance(test_id, list):
test_id = tuple(test_id)
return test_id
def _get_file_name(self, test_id):
if isinstance(test_id, str):
return test_id
if isinstance(test_id, tuple):
return "".join(test_id)
assert False, "unexpected test_id"
def suite_start(self, data):
num_tests = reduce(lambda x, y: x + len(y), data["tests"].values(), 0)
action = self.color_formatter.action(data["action"].upper())
name = ""
if "name" in data:
name = " %s -" % (data["name"],)
return "%s:%s running %i tests" % (action, name, num_tests)
def suite_end(self, data):
action = self.color_formatter.action(data["action"].upper())
rv = [action]
if not self.summary_on_shutdown:
rv.append(
self._format_suite_summary(
self.summary.current_suite, self.summary.current
)
)
return "\n".join(rv)
def _format_expected(self, status, expected, known_intermittent=[]):
if status == expected:
color = self.color_formatter.log_test_status_pass
if expected not in ("PASS", "OK"):
color = self.color_formatter.log_test_status_fail
status = "EXPECTED-%s" % status
else:
if status in known_intermittent:
color = self.color_formatter.log_test_status_known_intermittent
status = "KNOWN-INTERMITTENT-%s" % status
else:
color = self.color_formatter.log_test_status_fail
if status in ("PASS", "OK"):
status = "UNEXPECTED-%s" % status
return color(status)
def _format_status(self, test, data):
name = data.get("subtest", test)
rv = "%s %s" % (
self._format_expected(
data["status"],
data.get("expected", data["status"]),
data.get("known_intermittent", []),
),
name,
)
if "message" in data:
rv += " - %s" % data["message"]
if "stack" in data:
rv += self._format_stack(data["stack"])
return rv
def _format_stack(self, stack):
return "\n%s\n" % self.color_formatter.bright_black(stack.strip("\n"))
def _format_suite_summary(self, suite, summary):
count = summary["counts"]
logs = summary["unexpected_logs"]
intermittent_logs = summary["intermittent_logs"]
harness_errors = summary["harness_errors"]
rv = [
"",
self.color_formatter.sub_heading(suite),
self.color_formatter.sub_heading("~" * len(suite)),
]
# Format check counts
checks = self.summary.aggregate("count", count)
rv.append(
"Ran {} checks ({})".format(
sum(checks.values()),
", ".join(
["{} {}s".format(v, k) for k, v in sorted(checks.items()) if v]
),
)
)
# Format expected counts
checks = self.summary.aggregate("expected", count, include_skip=False)
intermittent_checks = self.summary.aggregate(
"known_intermittent", count, include_skip=False
)
intermittents = sum(intermittent_checks.values())
known = (
" ({} known intermittents)".format(intermittents) if intermittents else ""
)
rv.append("Expected results: {}{}".format(sum(checks.values()), known))
# Format skip counts
skip_tests = count["test"]["expected"]["skip"]
skip_subtests = count["subtest"]["expected"]["skip"]
if skip_tests:
skipped = "Skipped: {} tests".format(skip_tests)
if skip_subtests:
skipped = "{}, {} subtests".format(skipped, skip_subtests)
rv.append(skipped)
# Format unexpected counts
checks = self.summary.aggregate("unexpected", count)
unexpected_count = sum(checks.values())
rv.append("Unexpected results: {}".format(unexpected_count))
if unexpected_count:
for key in ("test", "subtest", "assert"):
if not count[key]["unexpected"]:
continue
status_str = ", ".join(
[
"{} {}".format(n, s)
for s, n in sorted(count[key]["unexpected"].items())
]
)
rv.append(
" {}: {} ({})".format(
key, sum(count[key]["unexpected"].values()), status_str
)
)
# Format intermittents
if intermittents > 0:
heading = "Known Intermittent Results"
rv.extend(
[
"",
self.color_formatter.heading(heading),
self.color_formatter.heading("-" * len(heading)),
]
)
if count["subtest"]["count"]:
for test_id, results in intermittent_logs.items():
test = self._get_file_name(test_id)
rv.append(self.color_formatter.bold(test))
for data in results:
rv.append(" %s" % self._format_status(test, data).rstrip())
else:
for test_id, results in intermittent_logs.items():
test = self._get_file_name(test_id)
for data in results:
assert "subtest" not in data
rv.append(self._format_status(test, data).rstrip())
# Format status
testfailed = any(
count[key]["unexpected"] for key in ("test", "subtest", "assert")
)
if not testfailed and not harness_errors:
rv.append(self.color_formatter.log_test_status_pass("OK"))
else:
# Format test failures
heading = "Unexpected Results"
rv.extend(
[
"",
self.color_formatter.heading(heading),
self.color_formatter.heading("-" * len(heading)),
]
)
if count["subtest"]["count"]:
for test_id, results in logs.items():
test = self._get_file_name(test_id)
rv.append(self.color_formatter.bold(test))
for data in results:
rv.append(" %s" % self._format_status(test, data).rstrip())
else:
for test_id, results in logs.items():
test = self._get_file_name(test_id)
for data in results:
assert "subtest" not in data
rv.append(self._format_status(test, data).rstrip())
# Format harness errors
if harness_errors:
for data in harness_errors:
rv.append(self.log(data))
return "\n".join(rv)
def test_start(self, data):
action = self.color_formatter.action(data["action"].upper())
return "%s: %s" % (action, self._get_test_id(data))
def test_end(self, data):
subtests = self._get_subtest_data(data)
if "expected" in data and data["status"] not in data.get(
"known_intermittent", []
):
parent_unexpected = True
expected_str = ", expected %s" % data["expected"]
else:
parent_unexpected = False
expected_str = ""
has_screenshots = "reftest_screenshots" in data.get("extra", {})
test = self._get_test_id(data)
# Reset the counts to 0
self.status_buffer[test] = {"count": 0, "unexpected": 0, "pass": 0}
self.has_unexpected[test] = bool(subtests["unexpected"])
if subtests["count"] != 0:
rv = "Test %s%s. Subtests passed %i/%i. Unexpected %s" % (
data["status"],
expected_str,
subtests["pass"],
subtests["count"],
subtests["unexpected"],
)
else:
rv = "%s%s" % (data["status"], expected_str)
unexpected = self.summary.current["unexpected_logs"].get(data["test"])
if unexpected:
if len(unexpected) == 1 and parent_unexpected:
message = unexpected[0].get("message", "")
if message:
rv += " - %s" % message
if "stack" in data:
rv += self._format_stack(data["stack"])
elif not self.verbose:
rv += "\n"
for d in unexpected:
rv += self._format_status(data["test"], d)
intermittents = self.summary.current["intermittent_logs"].get(data["test"])
if intermittents:
rv += "\n"
for d in intermittents:
rv += self._format_status(data["test"], d)
if "expected" not in data and not bool(subtests["unexpected"]):
color = self.color_formatter.log_test_status_pass
else:
color = self.color_formatter.log_test_status_unexpected_fail
action = color(data["action"].upper())
rv = "%s: %s" % (action, rv)
if has_screenshots and self.enable_screenshot:
if self.tbpl_formatter is None:
self.tbpl_formatter = TbplFormatter()
# Create TBPL-like output that can be pasted into the reftest analyser
rv = "\n".join((rv, self.tbpl_formatter.test_end(data)))
return rv
def valgrind_error(self, data):
rv = " " + data["primary"] + "\n"
for line in data["secondary"]:
rv = rv + line + "\n"
return rv
def lsan_leak(self, data):
allowed = data.get("allowed_match")
if allowed:
prefix = self.color_formatter.log_test_status_fail("FAIL")
else:
prefix = self.color_formatter.log_test_status_unexpected_fail(
"UNEXPECTED-FAIL"
)
return "%s LeakSanitizer: leak at %s" % (prefix, ", ".join(data["frames"]))
def lsan_summary(self, data):
allowed = data.get("allowed", False)
if allowed:
prefix = self.color_formatter.warning("WARNING")
else:
prefix = self.color_formatter.error("ERROR")
return (
"%s | LeakSanitizer | "
"SUMMARY: AddressSanitizer: %d byte(s) leaked in %d allocation(s)."
% (prefix, data["bytes"], data["allocations"])
)
def mozleak_object(self, data):
data_log = data.copy()
data_log["level"] = "INFO"
data_log["message"] = "leakcheck: %s leaked %d %s" % (
data["process"],
data["bytes"],
data["name"],
)
return self.log(data_log)
def mozleak_total(self, data):
if data["bytes"] is None:
# We didn't see a line with name 'TOTAL'
if data.get("induced_crash", False):
data_log = data.copy()
data_log["level"] = "INFO"
data_log["message"] = (
"leakcheck: %s deliberate crash and thus no leak log\n"
% data["process"]
)
return self.log(data_log)
if data.get("ignore_missing", False):
return (
"%s ignoring missing output line for total leaks\n"
% data["process"]
)
status = self.color_formatter.log_test_status_pass("FAIL")
return "%s leakcheck: " "%s missing output line for total leaks!\n" % (
status,
data["process"],
)
if data["bytes"] == 0:
return "%s leakcheck: %s no leaks detected!\n" % (
self.color_formatter.log_test_status_pass("PASS"),
data["process"],
)
message = "leakcheck: %s %d bytes leaked\n" % (data["process"], data["bytes"])
# data["bytes"] will include any expected leaks, so it can be off
# by a few thousand bytes.
failure = data["bytes"] > data["threshold"]
status = (
self.color_formatter.log_test_status_unexpected_fail("UNEXPECTED-FAIL")
if failure
else self.color_formatter.log_test_status_fail("FAIL")
)
return "%s %s\n" % (status, message)
def test_status(self, data):
test = self._get_test_id(data)
if test not in self.status_buffer:
self.status_buffer[test] = {"count": 0, "unexpected": 0, "pass": 0}
self.status_buffer[test]["count"] += 1
if data["status"] == "PASS":
self.status_buffer[test]["pass"] += 1
if "expected" in data and data["status"] not in data.get(
"known_intermittent", []
):
self.status_buffer[test]["unexpected"] += 1
if self.verbose:
return self._format_status(test, data).rstrip("\n")
def assertion_count(self, data):
if data["min_expected"] <= data["count"] <= data["max_expected"]:
return
if data["min_expected"] != data["max_expected"]:
expected = "%i to %i" % (data["min_expected"], data["max_expected"])
else:
expected = "%i" % data["min_expected"]
action = self.color_formatter.log_test_status_fail("ASSERT")
return "%s: Assertion count %i, expected %s assertions\n" % (
action,
data["count"],
expected,
)
def process_output(self, data):
rv = []
pid = data["process"]
if pid.isdigit():
pid = "pid:%s" % pid
pid = self.color_formatter.pid(pid)
if "command" in data and data["process"] not in self._known_pids:
self._known_pids.add(data["process"])
rv.append("%s Full command: %s" % (pid, data["command"]))
rv.append("%s %s" % (pid, data["data"]))
return "\n".join(rv)
def crash(self, data):
test = self._get_test_id(data)
if data.get("stackwalk_returncode", 0) != 0 and not data.get(
"stackwalk_stderr"
):
success = True
else:
success = False
rv = [
"pid:%s. Process type: %s. Test:%s. Minidump analysed:%s. Signature:[%s]"
% (
data.get("pid", "unknown"),
data.get("process_type", None),
test,
success,
data["signature"],
)
]
if data.get("java_stack"):
rv.append("Java exception: %s" % data["java_stack"])
else:
if data.get("reason"):
rv.append("Mozilla crash reason: %s" % data["reason"])
if data.get("minidump_path"):
rv.append("Crash dump filename: %s" % data["minidump_path"])
if data.get("stackwalk_returncode", 0) != 0:
rv.append(
"minidump-stackwalk exited with return code %d"
% data["stackwalk_returncode"]
)
if data.get("stackwalk_stderr"):
rv.append("stderr from minidump-stackwalk:")
rv.append(data["stackwalk_stderr"])
elif data.get("stackwalk_stdout"):
rv.append(data["stackwalk_stdout"])
if data.get("stackwalk_errors"):
rv.extend(data.get("stackwalk_errors"))
rv = "\n".join(rv)
if not rv[-1] == "\n":
rv += "\n"
action = self.color_formatter.action(data["action"].upper())
return "%s: %s" % (action, rv)
def process_start(self, data):
rv = "Started process `%s`" % data["process"]
desc = data.get("command")
if desc:
rv = "%s (%s)" % (rv, desc)
return rv
def process_exit(self, data):
return "%s: %s" % (data["process"], strstatus(data["exitcode"]))
def log(self, data):
level = data.get("level").upper()
if level in ("CRITICAL", "ERROR"):
level = self.color_formatter.error(level)
elif level == "WARNING":
level = self.color_formatter.warning(level)
elif level == "INFO":
level = self.color_formatter.log_process_output(level)
if data.get("component"):
rv = " ".join([data["component"], level, data["message"]])
else:
rv = "%s %s" % (level, data["message"])
if "stack" in data:
rv += "\n%s" % data["stack"]
return rv
def lint(self, data):
fmt = (
"{path} {c1}{lineno}{column} {c2}{level}{normal} {message}"
" {c1}{rule}({linter}){normal}"
)
message = fmt.format(
path=data["path"],
normal=self.color_formatter.normal,
c1=self.color_formatter.grey,
c2=self.color_formatter.error
if data["level"] == "error"
else (self.color_formatter.log_test_status_fail),
lineno=str(data["lineno"]),
column=(":" + str(data["column"])) if data.get("column") else "",
level=data["level"],
message=data["message"],
rule="{} ".format(data["rule"]) if data.get("rule") else "",
linter=data["linter"].lower() if data.get("linter") else "",
)
return message
def shutdown(self, data):
if not self.summary_on_shutdown:
return
heading = "Overall Summary"
rv = [
"",
self.color_formatter.heading(heading),
self.color_formatter.heading("=" * len(heading)),
]
for suite, summary in self.summary:
rv.append(self._format_suite_summary(suite, summary))
return "\n".join(rv)
def _get_subtest_data(self, data):
test = self._get_test_id(data)
return self.status_buffer.get(test, {"count": 0, "unexpected": 0, "pass": 0})
def _time(self, data):
entry_time = data["time"]
if self.write_interval and self.last_time is not None:
t = entry_time - self.last_time
self.last_time = entry_time
else:
t = entry_time - self.start_time
return t / 1000.0
Codebase Browser
chromium