Add report parser for Pipeline tests
diff --git a/report-tools/generate_report.py b/report-tools/generate_report.py
new file mode 100644
index 0000000..7c70050
--- /dev/null
+++ b/report-tools/generate_report.py
@@ -0,0 +1,456 @@
+##############################################################################
+
+# 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)