blob: 4db630e686524fecd6e0a07c86d715783d850213 [file] [log] [blame]
# Copyright 2018 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import json
import logging
import os
from typing import Any, Dict, List, Tuple, Optional
def UnitStringIsValid(unit: str) -> bool:
"""Checks to make sure that a given string is in fact a recognized unit used
by the chromium perftests to report results.
unit (str): The unit string to be checked.
bool: Whether or not it is a unit.
accepted_units = [
"us/hop", "us/task", "ns/sample", "ms", "s", "count", "KB", "MB/s", "us"
return unit in accepted_units
class ResultLine(object):
"""ResultLine objects are each an individual line of output, complete with a
unit, measurement, and descriptive component.
def __init__(self, desc: str, measure: float, unit: str) -> None:
self.desc = desc
self.meas = measure
self.unit = unit
def ToJsonDict(self) -> Dict[str, Any]:
"""Converts a ResultLine into a JSON-serializable dictionary object.
Dict[str, Any]: A mapping of strings that will appear in the output JSON
object to their respective values.
return {
"description": self.desc,
"measurement": self.meas,
"unit": self.unit,
def ReadResultLineFromJson(dct: Dict[str, Any]) -> ResultLine:
"""Converts a JSON dictionary object into a ResultLine.
dct (Dict[str, Any]): The JSON object to be parsed as a ResultLine. MUST
contain the strings 'description', 'measurement', and 'unit'.
KeyError: If the passed in dictionary does not contain the three required
strings that a serialized ResultLine must contain.
ResultLine: A ResultLine object reconstituted from the JSON dictionary.
return ResultLine(dct["description"], float(dct["measurement"]), dct["unit"])
def ResultLineFromStdout(line: str) -> Optional[ResultLine]:
"""Takes a line of stdout data and attempts to parse it into a ResultLine.
line (str): The stdout line to be converted
Optional[ResultLine]: The output is Optional, because the line may be noise,
or in some way incorrectly formatted and unparseable.
if "pkgsvr" in line:
return None # Filters pkgsrv noise from Fuchsia output.
chunks = line.split()
# There should be 1 chunk for the measure, 1 for the unit, and at least one
# for the line description, so at least 3 total
if len(chunks) < 3:
logging.warning("The line {} contains too few space-separated pieces to be "
"parsed as a ResultLine".format(line))
return None
unit = chunks[-1]
if not UnitStringIsValid(unit):
logging.warning("The unit string parsed from {} was {}, which was not "
"expected".format(line, unit))
return None
measure = float(chunks[-2])
desc = " ".join(chunks[:-2])
return ResultLine(desc, measure, unit)
except ValueError as e:
logging.warning("The chunk {} could not be parsed as a valid measurement "
"because of {}".format(chunks[-2], str(e)))
return None
class TestResult(object):
"""TestResult objects comprise the smallest unit of a GTest target, and
contain the name of the individual test run, and the time that the test took
to run."""
def __init__(self, name: str, time: float, lines: List[ResultLine]) -> None: = name
self.time = time
self.lines = lines
def ToJsonDict(self) -> Dict[str, Any]:
"""Converts a TestResult object to a JSON-serializable dictionary.
Dict[str, Any]: The output dictionary object that can be directly
serialized to JSON.
return {
"time_in_ms": self.time,
"lines": [line.ToJsonDict() for line in self.lines]
def ReadTestFromJson(obj_dict: Dict[str, Any]) -> TestResult:
"""Reconstitutes a TestResult read from a JSON file back into a TestResult
obj_dict (Dict[str, Any]): The JSON object as read from an output JSON file.
TestResult: The reconstituted TestResult object.
name = obj_dict["name"]
time = obj_dict["time_in_ms"]
lines = [ReadResultLineFromJson(line) for line in obj_dict["lines"]]
return TestResult(name, time, lines)
def ExtractTestInfo(line: str) -> Tuple[str, float]:
"""Deconstructs a line starting with OK, stating that the test finished
successfully, and isolates the timing measurement as well as a descriptive
string for the test
line (str): The line of output to attempt to destructure into name and time.
Exception: In the event that it couldn't split on '(', because then it
find the necessary timing measurement.
Exception: in the event that it cannot find the ')' character in the output,
because then it isn't capable of isolating the timing measurement.
Tuple[str, float]: A tuple of the test name, and the amount of time it took
to run.
# Trim off the [ OK ] part of the line
trimmed = line.lstrip("[ OK ]").strip()
test_name, rest = trimmed.split("(") # Isolate the measurement
except Exception as e:
err_text = "Could not extract the case name from {} because of error {}"\
.format(trimmed, str(e))
raise Exception(err_text)
measure, _ = rest.split(")", 1)[0].split()
except Exception as e:
err_text = "Could not extract measure and units from {}\
because of error {}".format(rest, str(e))
raise Exception(err_text)
return test_name.strip(), float(measure)
def TaggedTestFromLines(lines: List[str]) -> TestResult:
"""Takes a chunk of lines gathered together from the stdout of a test process
and collects it all into a single test result, including the set of
ResultLines inside of the TestResult.
lines (List[str]): The stdout lines to be parsed into a single test result
TestResult: The final parsed TestResult from the input
test_name, time = ExtractTestInfo(lines[-1])
res_lines = []
for line in lines[:-1]:
res_line = ResultLineFromStdout(line)
if res_line:
logging.warning("Couldn't parse line {} into a ResultLine".format(line))
return TestResult(test_name, time, res_lines)
class TargetResult(object):
"""TargetResult objects contain the entire set of TestSuite objects that are
invoked by a single test target, such as base:base_perftests and the like.
They also include the name of the target, and the time it took the target to
def __init__(self, name: str, tests: List[TestResult]) -> None: = name
self.tests = tests
def ToJsonDict(self) -> Dict[str, Any]:
"""Converts a TargetResult to a JSON-serializable dict.
Dict[str, Any]: The TargetResult in JSON-serializable form.
return {
"tests": [test.ToJsonDict() for test in self.tests]
def WriteToJson(self, path: str) -> None:
"""Writes this TargetResult object as a JSON file at the given file path.
path (str): The location to place the serialized version of this object.
None: It'd return IO (), but this Python.
with open(path, "w") as outfile:
json.dump(self.ToJsonDict(), outfile, indent=2)
def ReadTargetFromJson(path: str) -> Optional[TargetResult]:
"""Takes a file path and attempts to parse a TargetResult from it.
path (str): The location of the JSON-serialized TargetResult to read.
Optional[TargetResult]: Again, technically should be wrapped in an IO,
but Python.
with open(path, "r") as json_file:
dct = json.load(json_file)
return TargetResult(
dct["name"], [ReadTestFromJson(test_dct) for test_dct in dct["tests"]])
def TargetResultFromStdout(lines: List[str], name: str) -> TargetResult:
"""TargetResultFromStdout attempts to associate GTest names to the lines of
output that they produce. Example input looks something like the following:
[ RUN ] TestNameFoo
INFO measurement units
[ OK ] TestNameFoo (measurement units)
Unfortunately, Because the results of output from perftest targets is not
necessarily consistent between test targets, this makes a best-effort to parse
as much information from them as possible.
lines (List[str]): The entire list of lines from the standard output to be
name (str): The name of the Target that generated the output. Necessary to
be able to give the TargetResult a meaningful name.
TargetResult: The TargetResult object parsed from the input lines.
test_line_lists = [] # type: List[List[str]]
test_line_accum = [] # type: List[str]
read_lines = False
for line in lines:
# We're starting a test suite
if line.startswith("[ RUN ]"):
read_lines = True
# We have a prior suite that needs to be added
if len(test_line_accum) > 0:
test_line_accum = []
elif read_lines:
# We don't actually care about the data in the RUN line, just its
# presence. the OK line contains the same info, as well as the total test
# run time
if line.startswith("[ OK ]"):
read_lines = False
test_cases = [
TaggedTestFromLines(test_lines) for test_lines in test_line_lists
return TargetResult(name, test_cases)