#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Continuous integration toolkit. The purpose of this script is to generate everything which is needed for the CI environment. List of responsibilities: * execute default suites * terminate execution if any of internal scenarios fails * creates the report file that is needed by the Apollo plugin to post the status on Phabricator. (.quality_assurance_status) """ import os import sys import json import logging import subprocess from argparse import ArgumentParser log = logging.getLogger(__name__) SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__)) class Test: """ Class used to store basic information about a single test suite @attribute name: string, name of the test_suite (must be unique) @attribute test_suite: string, test_suite within tck_engine/tests which contains tck tests @attribute memgraph_params string, any command line arguments that should be passed to memgraph before evaluating tests from this suite @attribute mandatory bool, True if this suite is obligatory for continuous integration to pass """ def __init__(self, name, test_suite, memgraph_params, mandatory): self.name = name self.test_suite = test_suite self.memgraph_params = memgraph_params self.mandatory = mandatory # Constants suites = [ Test( name="memgraph_V1", test_suite="memgraph_V1", memgraph_params="", mandatory=True ), Test( name="memgraph_V1_POD", test_suite="memgraph_V1", memgraph_params="--properties-on-disk=x,y,z,w,k,v,a,b,c,d,e,f,r,t,o,prop,age,name,surname,location", mandatory=True ), Test( name="openCypher_M09", test_suite="openCypher_M09", memgraph_params="", mandatory=False ), ] results_folder = os.path.join("tck_engine", "results") suite_suffix = "memgraph-{}.json" qa_status_path = ".quality_assurance_status" measurements_path = ".apollo_measurements" def parse_args(): """ Parse command line arguments """ argp = ArgumentParser(description=__doc__) argp.add_argument("--distributed", action="store_true") return argp.parse_args() def get_newest_path(folder, suffix): """ :param folder: Scanned folder. :param suffix: File suffix. :return: Path to the newest file in the folder with the specified suffix. """ name_list = sorted(filter(lambda x: x.endswith(suffix), os.listdir(folder))) if len(name_list) <= 0: sys.exit("Unable to find any file with suffix %s in folder %s!" % (suffix, folder)) return os.path.join(folder, name_list.pop()) def generate_measurements(suite, result_path): """ :param suite: Test suite name. :param result_path: File path with json status report. :return: Measurements string. """ with open(result_path) as f: result = json.load(f) ret = "" for i in ["total", "passed", "restarts"]: ret += "{}.{} {}\n".format(suite, i, result[i]) return ret def generate_status(suite, result_path, required=False): """ :param suite: Test suite name. :param result_path: File path with json status report. :param required: Adds status ticks to the message if required. :return: Status string. """ with open(result_path) as f: result = json.load(f) total = result["total"] passed = result["passed"] restarts = result["restarts"] ratio = passed / total msg = "{} / {} //({:.2%})//".format(passed, total, ratio) if required: if passed == total: msg += " {icon check color=green}" else: msg += " {icon times color=red}" return (msg, passed, total, restarts) def generate_remarkup(data, distributed=False): """ :param data: Tabular data to convert to remarkup. :return: Remarkup formatted status string. """ extra_desc = "distributed " if distributed else "" ret = "==== Quality assurance {}status: ====\n\n".format(extra_desc) ret += "\n" for row in data: ret += " \n" for item in row: if row == data[0]: fmt = " \n" else: fmt = " \n" ret += fmt.format(item) ret += " \n" ret += "
{}{}
\n" return ret if __name__ == "__main__": args = parse_args() distributed = [] # Tests are not mandatory for distributed if args.distributed: distributed = ["--distributed"] for suite in suites: suite.mandatory = False # Logger config logging.basicConfig(level=logging.INFO) venv_python = os.path.join(SCRIPT_DIR, "ve3", "bin", "python3") exec_dir = os.path.realpath(os.path.join(SCRIPT_DIR, "tck_engine")) tests_dir = os.path.realpath(os.path.join(exec_dir, "tests")) # Run suites for suite in suites: log.info("Starting suite '{}' scenarios.".format(suite.name)) test = os.path.realpath(os.path.join(tests_dir, suite.test_suite)) cmd = [venv_python, "-u", os.path.join(exec_dir, "test_executor.py"), "--root", test, "--test-name", "{}".format(suite.name), "--db", "memgraph", "--memgraph-params", "\"{}\"".format(suite.memgraph_params)] + distributed subprocess.run(cmd, check=False) # Measurements measurements = "" # Status table headers status_data = [["Suite", "Scenarios", "Restarts"]] # List of mandatory suites that have failed mandatory_fails = [] for suite in suites: # Get data files for test suite suite_result_path = get_newest_path(results_folder, suite_suffix.format(suite.name)) log.info("Memgraph result path is {}".format(suite_result_path)) # Read scenarios suite_status, suite_passed, suite_total, suite_restarts = \ generate_status(suite.name, suite_result_path, required=suite.mandatory) if suite.mandatory and suite_passed != suite_total or \ not args.distributed and suite_restarts > 0: mandatory_fails.append(suite.name) status_data.append([suite.name, suite_status, suite_restarts]) measurements += generate_measurements(suite.name, suite_result_path) # Create status message qa_status_message = generate_remarkup(status_data, args.distributed) # Create the report file with open(qa_status_path, "w") as f: f.write(qa_status_message) # Create the measurements file with open(measurements_path, "w") as f: f.write(measurements) log.info("Status is generated in %s" % qa_status_path) log.info("Measurements are generated in %s" % measurements_path) if mandatory_fails != []: sys.exit("Some mandatory tests have failed -- %s" % str(mandatory_fails))