DXR is a code search and navigation tool aimed at making sense of large projects. It supports full-text and regex searches as well as structural queries.

Mercurial (c68fe15a81fc)

VCS Links

Line Code
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164
# 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/.

from __future__ import absolute_import, print_function

import json
import threading
from collections import defaultdict

from mozlog.formatters import TbplFormatter
from mozrunner.utils import get_stack_fixer_function


class ReftestFormatter(TbplFormatter):
    """
    Formatter designed to preserve the legacy "tbpl" format in reftest.

    This is needed for both the reftest-analyzer and mozharness log parsing.
    We can change this format when both reftest-analyzer and mozharness have
    been changed to read structured logs.
    """

    def __call__(self, data):
        if 'component' in data and data['component'] == 'mozleak':
            # Output from mozleak requires that no prefix be added
            # so that mozharness will pick up these failures.
            return "%s\n" % data['message']

        formatted = TbplFormatter.__call__(self, data)

        if formatted is None:
            return
        if data['action'] == 'process_output':
            return formatted
        return 'REFTEST %s' % formatted

    def log(self, data):
        prefix = "%s |" % data['level'].upper()
        return "%s %s\n" % (prefix, data['message'])

    def _format_status(self, data):
        extra = data.get('extra', {})
        status = data['status']

        status_msg = "TEST-"
        if 'expected' in data:
            status_msg += "UNEXPECTED-%s" % status
        else:
            if status not in ("PASS", "SKIP"):
                status_msg += "KNOWN-"
            status_msg += status
            if extra.get('status_msg') == 'Random':
                status_msg += "(EXPECTED RANDOM)"
        return status_msg

    def test_status(self, data):
        extra = data.get('extra', {})
        test = data['test']

        status_msg = self._format_status(data)
        output_text = "%s | %s | %s" % (status_msg, test, data.get("subtest", "unknown test"))
        if data.get('message'):
            output_text += " | %s" % data['message']

        if "reftest_screenshots" in extra:
            screenshots = extra["reftest_screenshots"]
            image_1 = screenshots[0]["screenshot"]

            if len(screenshots) == 3:
                image_2 = screenshots[2]["screenshot"]
                output_text += ("\nREFTEST   IMAGE 1 (TEST): data:image/png;base64,%s\n"
                                "REFTEST   IMAGE 2 (REFERENCE): data:image/png;base64,%s") % (
                                image_1, image_2)
            elif len(screenshots) == 1:
                output_text += "\nREFTEST   IMAGE: data:image/png;base64,%s" % image_1

        return output_text + "\n"

    def test_end(self, data):
        status = data['status']
        test = data['test']

        output_text = ""
        if status != "OK":
            status_msg = self._format_status(data)
            output_text = "%s | %s | %s" % (status_msg, test, data.get("message", ""))

        if output_text:
            output_text += "\nREFTEST "
        output_text += "TEST-END | %s" % test
        return "%s\n" % output_text

    def process_output(self, data):
        return "%s\n" % data["data"]

    def suite_end(self, data):
        lines = []
        summary = data['extra']['results']
        summary['success'] = summary['Pass'] + summary['LoadOnly']
        lines.append("Successful: %(success)s (%(Pass)s pass, %(LoadOnly)s load only)" %
                     summary)
        summary['unexpected'] = (summary['Exception'] + summary['FailedLoad'] +
                                 summary['UnexpectedFail'] + summary['UnexpectedPass'] +
                                 summary['AssertionUnexpected'] +
                                 summary['AssertionUnexpectedFixed'])
        lines.append(("Unexpected: %(unexpected)s (%(UnexpectedFail)s unexpected fail, "
                      "%(UnexpectedPass)s unexpected pass, "
                      "%(AssertionUnexpected)s unexpected asserts, "
                      "%(FailedLoad)s failed load, "
                      "%(Exception)s exception)") % summary)
        summary['known'] = (summary['KnownFail'] + summary['AssertionKnown'] +
                            summary['Random'] + summary['Skip'] + summary['Slow'])
        lines.append(("Known problems: %(known)s (" +
                      "%(KnownFail)s known fail, " +
                      "%(AssertionKnown)s known asserts, " +
                      "%(Random)s random, " +
                      "%(Skip)s skipped, " +
                      "%(Slow)s slow)") % summary)
        lines = ["REFTEST INFO | %s" % s for s in lines]
        lines.append("REFTEST SUITE-END | Shutdown")
        return "INFO | Result summary:\n{}\n".format('\n'.join(lines))


class OutputHandler(object):
    """Process the output of a process during a test run and translate
    raw data logged from reftest.js to an appropriate structured log action,
    where applicable.
    """

    def __init__(self, log, utilityPath, symbolsPath=None):
        self.stack_fixer_function = get_stack_fixer_function(utilityPath, symbolsPath)
        self.log = log
        self.proc_name = None
        self.results = defaultdict(int)

    def __call__(self, line):
        # need to return processed messages to appease remoteautomation.py
        if not line.strip():
            return []
        line = line.decode('utf-8', errors='replace')

        try:
            data = json.loads(line)
        except ValueError:
            self.verbatim(line)
            return [line]

        if isinstance(data, dict) and 'action' in data:
            if data['action'] == 'results':
                for k, v in data['results'].items():
                    self.results[k] += v
            else:
                self.log.log_raw(data)
        else:
            self.verbatim(json.dumps(data))

        return [data]

    def verbatim(self, line):
        if self.stack_fixer_function:
            line = self.stack_fixer_function(line)
        name = self.proc_name or threading.current_thread().name
        self.log.process_output(name, line)