report-tools/generate_report.py

##############################################################################

# Copyright (c) 2021, ARM Limited and Contributors. All rights reserved.

#

# SPDX-License-Identifier: BSD-3-Clause

##############################################################################
import re
import yaml
import argparse
import os
import logging
import time


class TCReport(object):
    """
    Class definition for objects to build report files in a
    pipeline stage
    """
    STATUS_VALUES = ["PASS", "FAIL", "SKIP"]

    def __init__(self, metadata=None, test_environments=None,
                 test_configuration=None, target=None,
                 test_suites=None, report_file=None):
        """
        Constructor for the class. Initialise the report object and loads
        an existing report(yaml) if defined.

        :param metadata: Initial metadata report object
        :param test_environments: Initial test environment object
        :param test_configuration: Initial test configuration object
        :param target: Initial target object
        :param test_suites: Initial test suites object
        :param report_file: If defined then an existing yaml report is loaded
        as the initial report object
        """
        if test_suites is None:
            test_suites = {}
        if target is None:
            target = {}
        if test_configuration is None:
            test_configuration = {'test-assets': {}}
        if test_environments is None:
            test_environments = {}
        if metadata is None:
            metadata = {}
        self._report_name = "Not-defined"
        # Define if is a new report or an existing report
        if report_file:
            # The structure of the report must follow:
            # - report name:
            #               {report properties}
            try:
                with open(report_file) as f:
                    full_report = yaml.full_load(f)
                    self._report_name, \
                    self.report = list(full_report.items())[0]
            except Exception as e:
                logging.exception(
                    f"Exception loading existing report '{report_file}'")
                raise Exception(e)
        else:
            self.report = {'metadata': metadata,
                           'test-environments': test_environments,
                           'test-config': test_configuration,
                           'target': target,
                           'test-suites': test_suites
                           }

    def dump(self, file_name):
        """
        Method that dumps the report object with the report name as key in
        a yaml format in a given file.

        :param file_name: File name to dump the yaml report
        :return: Nothing
        """
        with open(file_name, 'w') as f:
            yaml.dump({self._report_name: self.report}, f)

    @property
    def test_suites(self):
        return self.report['test-suites']

    @test_suites.setter
    def test_suites(self, value):
        self.test_suites = value

    @property
    def test_environments(self):
        return self.report['test-environments']

    @test_environments.setter
    def test_environments(self, value):
        self.test_environments = value

    @property
    def test_config(self):
        return self.report['test-config']

    @test_config.setter
    def test_config(self, value):
        self.test_config = value

    def add_test_suite(self, name: str, test_results, metadata=None):
        """
        Public method to add a test suite object to a report object.

        :param name: Unique test suite name
        :param test_results: Object with the tests results
        :param metadata: Metadata object for the test suite
        """
        if metadata is None:
            metadata = {}
        if name in self.test_suites:
            logging.error("Duplicated test suite:{}".format(name))
        else:
            self.test_suites[name] = {'test-results': test_results,
                                      'metadata': metadata}

    def add_test_environment(self, name: str, values=None):
        """
        Public method to add a test environment object to a report object.

        :param name: Name (key) of the test environment object
        :param values: Object assigned to the test environment object
        :return: Nothing
        """
        if values is None:
            values = {}
        self.test_environments[name] = values

    def add_test_asset(self, name: str, values=None):
        """
        Public method to add a test asset object to a report object.

        :param name: Name (key) of the test asset object
        :param values: Object assigned to the test asset object
        :return: Nothing
        """
        if values is None:
            values = {}
        if 'test-assets' not in self.test_config:
            self.test_config['test-assets'] = {}
        self.test_config['test-assets'][name] = values

    @staticmethod
    def process_ptest_results(lava_log_string="",
                              results_pattern=r"(?P<status>("
                                              r"PASS|FAIL|SKIP)): ("
                                              r"?P<id>.+)"):
        """
        Method that process ptest-runner results from a lava log string and
        converts them to a test results object.

        :param lava_log_string: Lava log string
        :param results_pattern: Regex used to capture the test results
        :return: Test results object
        """
        pattern = re.compile(results_pattern)
        if 'status' not in pattern.groupindex or \
            'id' not in pattern.groupindex:
            raise Exception(
                "Status and/or id must be defined in the results pattern")
        results = {}
        lines = lava_log_string.split("\n")
        it = iter(lines)
        stop_found = False
        for line in it:
            fields = line.split(" ", 1)
            if len(fields) > 1 and fields[1] == "START: ptest-runner":
                for report_line in it:
                    timestamp, *rest = report_line.split(" ", 1)
                    if not rest:
                        continue
                    if rest[0] == "STOP: ptest-runner":
                        stop_found = True
                        break
                    p = pattern.match(rest[0])
                    if p:
                        id = p.groupdict()['id'].replace(" ", "_")
                        status = p.groupdict()['status']
                        if not id:
                            print("Warning: missing 'id'")
                        elif status not in TCReport.STATUS_VALUES:
                            print("Warning: Status unknown")
                        elif id in results:
                            print("Warning: duplicated id")
                        else:
                            metadata = {k: p.groupdict()[k]
                                        for k in p.groupdict().keys()
                                        if k not in ('id', 'status')}
                            results[id] = {'status': status,
                                           'metadata': metadata}
                break
        if not stop_found:
            logger.warning("End of ptest-runner not found")
        return results

    def parse_fvp_model_version(self, lava_log_string):
        """
        Obtains the FVP model and version from a lava log string.

        :param lava_log_string: Lava log string
        :return: Tuple with FVP model and version
        """
        result = re.findall(r"opt/model/(.+) --version", lava_log_string)
        model = "" if not result else result[0]
        result = re.findall(r"Fast Models \[(.+?)\]\n", lava_log_string)
        version = "" if not result else result[0]
        self.report['target'] = {'platform': model, 'version': version}
        return model, version

    @property
    def report_name(self):
        return self._report_name

    @report_name.setter
    def report_name(self, value):
        self._report_name = value

    @property
    def metadata(self):
        return self.report['metadata']

    @metadata.setter
    def metadata(self, metadata):
        self.report['metadata'] = metadata


class KvDictAppendAction(argparse.Action):
    """
    argparse action to split an argument into KEY=VALUE form
    on the first = and append to a dictionary.
    """

    def __call__(self, parser, args, values, option_string=None):
        d = getattr(args, self.dest) or {}
        for value in values:
            try:
                (k, v) = value.split("=", 2)
            except ValueError as ex:
                raise \
                    argparse.ArgumentError(self, f"Could not parse argument '{values[0]}' as k=v format")
            d[k] = v
        setattr(args, self.dest, d)


def read_metadata(metadata_file):
    """
    Function that returns a dictionary object from a KEY=VALUE lines file.

    :param metadata_file: Filename with the KEY=VALUE pairs
    :return: Dictionary object with key and value pairs
    """
    if not metadata_file:
        return {}
    with open(metadata_file) as f:
        d = dict([line.strip().split("=", 1) for line in f])
    return d


def import_env(env_names):
    """
    Function that matches a list of regex expressions against all the
    environment variables keys and returns an object with the matched key
    and the value of the environment variable.

    :param env_names: List of regex expressions to match env keys
    :return: Object with the matched env variables
    """
    env_list = list(os.environ.keys())
    keys = []
    for expression in env_names:
        r = re.compile(expression)
        keys = keys + list(filter(r.match, env_list))
    d = {key: os.environ[key] for key in keys}
    return d


def merge_dicts(*dicts):
    """
    Function to merge a list of dictionaries.

    :param dicts: List of dictionaries
    :return: A merged dictionary
    """
    merged = {}
    for d in dicts:
        merged.update(d)
    return merged


def process_lava_log(_report, _args):
    """
    Function to adapt user arguments to process test results and add properties
    to the report object.

    :param _report: Report object
    :param _args: User arguments
    :return: Nothing
    """
    with open(_args.lava_log, "r") as f:
        lava_log = f.read()
    # Get the test results
    results = {}
    if _args.type == 'ptest-report':
        results = TCReport.process_ptest_results(lava_log,
                                                 results_pattern=_args.results_pattern)
    if _args.report_name:
        _report.report_name = _args.report_name
    _report.parse_fvp_model_version(lava_log)
    metadata = {}
    if _args.metadata_pairs or _args.metadata_env or _args.metadata_file:
        metadata = _args.metadata_pairs or import_env(
            _args.metadata_env) or read_metadata(_args.metadata_file)
    _report.add_test_suite(_args.test_suite_name, test_results=results,
                           metadata=metadata)


if __name__ == '__main__':
    # Defining logger
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=[
            logging.FileHandler("log_parser_{}.log".format(time.time())),
            logging.StreamHandler()
        ])
    """
    The main aim of this script is to be called with different options to build
    a report object that can be dumped into a yaml format
    """
    parser = argparse.ArgumentParser(description='Generic yaml report for TC')
    parser.add_argument("--report", "-r", help="Report filename")
    parser.add_argument("-f", help="Force new report", action='store_true',
                        dest='new_report')
    parser.add_argument("command", help="Command: process-results")
    group_results = parser.add_argument_group("Process results")
    group_results.add_argument('--test-suite-name', type=str,
                               help='Test suite name')
    group_results.add_argument('--lava-log', type=str, help='Lava log file')
    group_results.add_argument('--type', type=str, help='Type of report log',
                               default='ptest-report')
    group_results.add_argument('--report-name', type=str, help='Report name',
                               default="")
    group_results.add_argument('--results-pattern', type=str,
                               help='Regex pattern to extract test results',
                               default=r"(?P<status>(PASS|FAIL|SKIP)): ("
                                       r"?P<id>.+ .+) - ( "
                                       r"?P<description>.+)")
    test_env = parser.add_argument_group("Test environments")
    test_env.add_argument('--test-env-name', type=str,
                          help='Test environment type')
    test_env.add_argument("--test-env-values",
                          nargs="+",
                          action=KvDictAppendAction,
                          default={},
                          metavar="KEY=VALUE",
                          help="Set a number of key-value pairs "
                               "(do not put spaces before or after the = "
                               "sign). "
                               "If a value contains spaces, you should define "
                               "it with double quotes: "
                               'key="Value with spaces". Note that '
                               "values are always treated as strings.")
    test_env.add_argument("--test-env-env",
                          nargs="+",
                          default={},
                          help="Import environment variables values with the "
                               "given name.")
    parser.add_argument("--metadata-pairs",
                        nargs="+",
                        action=KvDictAppendAction,
                        default={},
                        metavar="KEY=VALUE",
                        help="Set a number of key-value pairs "
                             "(do not put spaces before or after the = sign). "
                             "If a value contains spaces, you should define "
                             "it with double quotes: "
                             'key="Value with spaces". Note that '
                             "values are always treated as strings.")

    test_config = parser.add_argument_group("Test config")
    test_config.add_argument('--test-asset-name', type=str,
                             help='Test asset type')
    test_config.add_argument("--test-asset-values",
                             nargs="+",
                             action=KvDictAppendAction,
                             default={},
                             metavar="KEY=VALUE",
                             help="Set a number of key-value pairs "
                                  "(do not put spaces before or after the = "
                                  "sign). "
                                  "If a value contains spaces, you should "
                                  "define "
                                  "it with double quotes: "
                                  'key="Value with spaces". Note that '
                                  "values are always treated as strings.")
    test_config.add_argument("--test-asset-env",
                             nargs="+",
                             default=None,
                             help="Import environment variables values with "
                                  "the given name.")

    parser.add_argument("--metadata-env",
                        nargs="+",
                        default=None,
                        help="Import environment variables values with the "
                             "given name.")
    parser.add_argument("--metadata-file",
                        type=str,
                        default=None,
                        help="File with key-value pairs lines i.e"
                             "key1=value1\nkey2=value2")

    args = parser.parse_args()
    report = None
    # Check if report exists (that can be overwritten) or is a new report
    if os.path.exists(args.report) and not args.new_report:
        report = TCReport(report_file=args.report)  # load existing report
    else:
        report = TCReport()

    # Possible list of commands:
    # process-results: To parse test results from stream into a test suite obj
    if args.command == "process-results":
        # Requires the test suite name and the log file, lava by the time being
        if not args.test_suite_name:
            parser.error("Test suite name required")
        elif not args.lava_log:
            parser.error("Lava log file required")
        process_lava_log(report, args)
    # set-report-metadata: Set the report's metadata
    elif args.command == "set-report-metadata":
        # Various options to load metadata into the report object
        report.metadata = merge_dicts(args.metadata_pairs,
                                      read_metadata(args.metadata_file),
                                      import_env(args.metadata_env))
    # add-test-environment: Add a test environment to the report's object
    elif args.command == "add-test-environment":
        # Various options to load environment data into the report object
        report.add_test_environment(args.test_env_name,
                                    merge_dicts(args.test_env_values,
                                                import_env(args.test_env_env)))
    # add-test-asset: Add a test asset into the report's object (test-config)
    elif args.command == "add-test-asset":
        report.add_test_asset(args.test_asset_name,
                              merge_dicts(args.test_asset_values,
                                          import_env(args.test_asset_env)))

    # Dump the report object as a yaml report
    report.dump(args.report)