tor-browser

The Tor Browser
git clone https://git.dasho.dev/tor-browser.git
Log | Files | Refs | README | LICENSE

reports.py (4662B)

      1 # This Source Code Form is subject to the terms of the Mozilla Public
      2 # License, v. 2.0. If a copy of the MPL was not distributed with this
      3 # file, # You can obtain one at http://mozilla.org/MPL/2.0/.
      4 
      5 import functools
      6 import itertools
      7 import json
      8 import os
      9 import sys
     10 import tempfile
     11 from os import path
     12 
     13 _EMPTY_REPORT = {
     14    "tests": 0,
     15    "failures": 0,
     16    "disabled": 0,
     17    "errors": 0,
     18    "testsuites": {},
     19 }
     20 
     21 
     22 def merge_gtest_reports(test_reports):
     23    """
     24    Logically merges json test reports matching [this
     25    schema](https://google.github.io/googletest/advanced.html#generating-a-json-report).
     26 
     27    It is assumed that each test will appear in at most one report (rather than
     28    trying to search and merge each test).
     29 
     30    Arguments:
     31    * test_reports - an iterator of python-native data (likely loaded from GTest JSON files).
     32    """
     33    INTEGER_FIELDS = ["tests", "failures", "disabled", "errors"]
     34    TESTSUITE_INTEGER_FIELDS = ["tests", "failures", "disabled"]
     35 
     36    def merge_testsuite(target, suite):
     37        for field in TESTSUITE_INTEGER_FIELDS:
     38            if field in suite:
     39                target[field] += suite[field]
     40        # We assume that each test will appear in at most one report,
     41        # so just extend the list of tests.
     42        target["testsuite"].extend(suite["testsuite"])
     43 
     44    def merge_one(current, report):
     45        for field in INTEGER_FIELDS:
     46            if field in report:
     47                current[field] += report[field]
     48        for suite in report["testsuites"]:
     49            name = suite["name"]
     50            if name in current["testsuites"]:
     51                merge_testsuite(current["testsuites"][name], suite)
     52            else:
     53                current["testsuites"][name] = suite
     54                for field in TESTSUITE_INTEGER_FIELDS:
     55                    current["testsuites"][name].setdefault(field, 0)
     56        return current
     57 
     58    merged = functools.reduce(merge_one, test_reports, _EMPTY_REPORT)
     59    # We had testsuites as a dict for fast lookup when merging, change
     60    # it back to a list to match the schema.
     61    merged["testsuites"] = list(merged["testsuites"].values())
     62 
     63    return merged
     64 
     65 
     66 class AggregatedGTestReport(dict):
     67    """
     68    An aggregated gtest report (stored as a `dict`)
     69 
     70    This should be used as a context manager to manage the lifetime of
     71    temporary storage for reports. If no exception occurs, when the context
     72    exits the reports will be merged into this dictionary. Thus, the context
     73    must not be exited before the outputs are written (e.g., by gtest processes
     74    completing).
     75 
     76    When merging results, it is assumed that each test will appear in at most
     77    one report (rather than trying to search and merge each test).
     78    """
     79 
     80    __slots__ = ["result_dir"]
     81 
     82    def __init__(self):
     83        tmpdir_kwargs = {}
     84        if sys.version_info >= (3, 10):
     85            tmpdir_kwargs["ignore_cleanup_errors"] = True
     86        self.result_dir = tempfile.TemporaryDirectory(**tmpdir_kwargs)
     87        super().__init__()
     88        self.reset()
     89 
     90    def __enter__(self):
     91        self.result_dir.__enter__()
     92        return self
     93 
     94    def __exit__(self, *exc_info):
     95        # Only collect reports if no exception occurred
     96        if exc_info[0] is None:
     97            d = self.result_dir.name
     98            result_files = filter(
     99                lambda f: path.isfile(f), map(lambda f: path.join(d, f), os.listdir(d))
    100            )
    101 
    102            def json_from_file(file):
    103                with open(file) as f:
    104                    return json.load(f)
    105 
    106            self.update(
    107                merge_gtest_reports(
    108                    itertools.chain([self], map(json_from_file, result_files))
    109                )
    110            )
    111        self.result_dir.__exit__(*exc_info)
    112 
    113    def reset(self):
    114        """Clear all results."""
    115        self.clear()
    116        self.update({
    117            "tests": 0,
    118            "failures": 0,
    119            "disabled": 0,
    120            "errors": 0,
    121            "testsuites": [],
    122        })
    123 
    124    def gtest_output(self, job_id):
    125        """
    126        Create a gtest output string with the given job id (to differentiate
    127        outputs).
    128        """
    129        # Replace `/` with `_` in job_id to prevent nested directories (job_id
    130        # may be a suite name, which may have slashes for parameterized test
    131        # suites).
    132        return f"json:{self.result_dir.name}/{job_id.replace('/', '_')}.json"
    133 
    134    def set_output_in_env(self, env, job_id):
    135        """
    136        Sets an environment variable mapping appropriate with the output for
    137        the given job id.
    138 
    139        Returns the env.
    140        """
    141        env["GTEST_OUTPUT"] = self.gtest_output(job_id)
    142        return env