memgraph/tests/qa/continuous_integration

#!/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

log = logging.getLogger(__name__)

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=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 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"]:
        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"]
    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)


def generate_remarkup(data):
    """
    :param data: Tabular data to convert to remarkup.

    :return: Remarkup formatted status string.
    """
    ret = "==== Quality assurance status: ====\n\n"
    ret += "<table>\n"
    for row in data:
        ret += "  <tr>\n"
        for item in row:
            if row == data[0]:
                fmt = "    <th>{}</th>\n"
            else:
                fmt = "    <td>{}</td>\n"
            ret += fmt.format(item)
        ret += "  </tr>\n"
    ret += "</table>\n"
    return ret


if __name__ == "__main__":
    # logger config
    logging.basicConfig(level=logging.INFO)

    # run suites
    for suite in suites:
        log.info("Starting suite '{}' scenarios.".format(suite.name))
        subprocess.run(["./run",
                        "--test-suite", suite.test_suite,
                        "--name", suite.name,
                        # "--verbose",
                        "--memgraph-params", suite.memgraph_params],
                       check = False)

    # measurements
    measurements = ""

    # status table headers
    status_data = [["Suite", "Scenarios"]]

    # 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 = generate_status(
                suite.name, suite_result_path, required = suite.mandatory)

        if suite.mandatory and suite_passed != suite_total:
            mandatory_fails.append(suite.name)

        status_data.append([suite.name, suite_status])
        measurements += generate_measurements(suite.name, suite_result_path)

    # create status message
    qa_status_message = generate_remarkup(status_data)

    # 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))