tor-browser

output.py (6384B)

---

# 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 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:
   """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 write(self, data):
       return self.__call__(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)