123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301 |
- # 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.
- Args:
- unit (str): The unit string to be checked.
- Returns:
- 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.
- Returns:
- 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.
- Args:
- dct (Dict[str, Any]): The JSON object to be parsed as a ResultLine. MUST
- contain the strings 'description', 'measurement', and 'unit'.
- Raises:
- KeyError: If the passed in dictionary does not contain the three required
- strings that a serialized ResultLine must contain.
- Returns:
- 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.
- Args:
- line (str): The stdout line to be converted
- Returns:
- 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
- try:
- 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:
- self.name = name
- self.time = time
- self.lines = lines
- def ToJsonDict(self) -> Dict[str, Any]:
- """Converts a TestResult object to a JSON-serializable dictionary.
- Returns:
- Dict[str, Any]: The output dictionary object that can be directly
- serialized to JSON.
- """
- return {
- "name": self.name,
- "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
- object.
- Args:
- obj_dict (Dict[str, Any]): The JSON object as read from an output JSON file.
- Returns:
- 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
- Args:
- line (str): The line of output to attempt to destructure into name and time.
- Raises:
- 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.
- Returns:
- 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()
- try:
- 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)
- try:
- 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.
- Args:
- lines (List[str]): The stdout lines to be parsed into a single test result
- Returns:
- 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:
- res_lines.append(res_line)
- else:
- 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
- run.
- """
- def __init__(self, name: str, tests: List[TestResult]) -> None:
- self.name = name
- self.tests = tests
- def ToJsonDict(self) -> Dict[str, Any]:
- """Converts a TargetResult to a JSON-serializable dict.
- Returns:
- Dict[str, Any]: The TargetResult in JSON-serializable form.
- """
- return {
- "name": self.name,
- "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.
- Args:
- path (str): The location to place the serialized version of this object.
- Returns:
- 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.
- Args:
- path (str): The location of the JSON-serialized TargetResult to read.
- Returns:
- 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.
- Args:
- lines (List[str]): The entire list of lines from the standard output to be
- parsed.
- name (str): The name of the Target that generated the output. Necessary to
- be able to give the TargetResult a meaningful name.
- Returns:
- 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_lists.append(test_line_accum)
- 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
- test_line_accum.append(line)
- if line.startswith("[ OK ]"):
- read_lines = False
- test_cases = [
- TaggedTestFromLines(test_lines) for test_lines in test_line_lists
- ]
- return TargetResult(name, test_cases)
|