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 argparse
import atexit
import copy
import os
import sys
import json
import subprocess
import tempfile
import time
import yaml


SCRIPT_DIR = os.path.dirname(os.path.realpath(__file__))
TESTS_DIR = os.path.join(SCRIPT_DIR, "tests")
BASE_DIR = os.path.normpath(os.path.join(SCRIPT_DIR, "..", ".."))
BUILD_DIR = os.path.join(BASE_DIR, "build_release")


def wait_for_server(port, delay=0.01):
    cmd = ["nc", "-z", "-w", "1", "127.0.0.1", str(port)]
    count = 0
    while subprocess.call(cmd) != 0:
        time.sleep(0.01)
        if count > 20 / 0.01:
            print("Could not wait for server on port", port, "to startup!")
            sys.exit(1)
        count += 1
    time.sleep(delay)


def generate_measurements(suite, result_path):
    if not os.path.exists(result_path):
        return ""
    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):
    if not os.path.exists(result_path):
        return ("Internal error!", 0, 1)
    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, distributed=False):
    extra_desc = "distributed " if distributed else ""
    ret = "==== Quality assurance {}status: ====\n\n".format(extra_desc)
    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


class MemgraphRunner():
    def __init__(self, build_directory):
        self.build_directory = build_directory
        self.proc_mg = None
        self.args = []

    def start(self, args):
        if args == self.args and self.is_running():
            return

        self.stop()
        self.args = copy.deepcopy(args)

        self.durability_directory = tempfile.TemporaryDirectory()
        memgraph_binary = os.path.join(self.build_directory, "memgraph")
        args_mg = [memgraph_binary, "--durability-directory",
                   self.durability_directory.name]
        self.proc_mg = subprocess.Popen(args_mg + self.args)
        wait_for_server(7687, 1)
        assert self.is_running(), "The Memgraph process died!"

    def is_running(self):
        if self.proc_mg is None:
            return False
        if self.proc_mg.poll() is not None:
            return False
        return True

    def stop(self):
        if not self.is_running():
            return
        self.proc_mg.terminate()
        code = self.proc_mg.wait()
        assert code == 0, "The Memgraph process exited with non-zero!"


class MemgraphDistributedRunner():
    def __init__(self, build_directory, cluster_size):
        self.build_directory = build_directory
        self.cluster_size = cluster_size
        self.procs = []
        self.durability_directories = []
        self.args = []

    def start(self, args):
        if args == self.args and self.is_running():
            return

        self.stop()
        self.args = copy.deepcopy(args)

        memgraph_binary = os.path.join(self.build_directory,
                                       "memgraph_distributed")

        self.procs = []
        self.durability_directories = []
        for i in range(self.cluster_size):
            durability_directory = tempfile.TemporaryDirectory()
            self.durability_directories.append(durability_directory)

            args_mg = [memgraph_binary]
            if i == 0:
                args_mg.extend(["--master", "--master-port", "10000"])
            else:
                args_mg.extend(["--worker", "--worker-id", str(i),
                                "--worker-port", str(10000 + i),
                                "--master-port", "10000"])
            args_mg.extend(["--durability-directory",
                            durability_directory.name])

            proc_mg = subprocess.Popen(args_mg + self.args)
            self.procs.append(proc_mg)

            wait_for_server(10000 + i, 1)

        wait_for_server(7687, 1)
        assert self.is_running(), "The Memgraph cluster died!"

    def is_running(self):
        if len(self.procs) == 0:
            return False
        for i, proc in enumerate(self.procs):
            code = proc.poll()
            if code is not None:
                if code != 0:
                    print("Memgraph node", i, "exited with non-zero!")
                return False
        return True

    def stop(self):
        if len(self.procs) == 0:
            return
        self.procs[0].terminate()
        died = False
        for i, proc in enumerate(self.procs):
            code = proc.wait()
            if code != 0:
                print("Memgraph node", i, "exited with non-zero!")
                died = True
        assert not died, "The Memgraph cluster died!"


def main():
    # Parse args
    argp = argparse.ArgumentParser()
    argp.add_argument("--build-directory", default=BUILD_DIR)
    argp.add_argument("--cluster-size", default=3, type=int)
    argp.add_argument("--distributed", action="store_true")
    args = argp.parse_args()

    # Load tests from config file
    with open(os.path.join(TESTS_DIR, "config.yaml")) as f:
        suites = yaml.load(f)

    # Tests are not mandatory for distributed
    if args.distributed:
        for suite in suites:
            suite["must_pass"] = False

    # venv used to run the qa engine
    venv_python = os.path.join(SCRIPT_DIR, "ve3", "bin", "python3")

    # Temporary directory for suite results
    output_dir = tempfile.TemporaryDirectory()

    # Memgraph runner
    if args.distributed:
        memgraph = MemgraphDistributedRunner(args.build_directory,
                                             args.cluster_size)
    else:
        memgraph = MemgraphRunner(args.build_directory)

    @atexit.register
    def cleanup():
        memgraph.stop()

    # Results storage
    measurements = ""
    status_data = [["Suite", "Scenarios"]]
    mandatory_fails = []

    # Run suites
    for suite in suites:
        print("Starting suite '{}' scenarios.".format(suite["name"]))

        params = []
        if "properties_on_disk" in suite:
            params = ["--properties-on-disk=" + suite["properties_on_disk"]]
        memgraph.start(params)

        suite["stats_file"] = os.path.join(output_dir.name,
                                           suite["name"] + ".json")
        cmd = [venv_python, "-u",
               os.path.join(SCRIPT_DIR, "qa.py"),
               "--stats-file", suite["stats_file"],
               suite["test_suite"]]

        # The exit code isn't checked here because the `behave` framework
        # returns a non-zero exit code when some tests fail.
        subprocess.run(cmd)

        suite_status, suite_passed, suite_total = \
            generate_status(suite["name"], suite["stats_file"],
                            suite["must_pass"])

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

        if suite["must_pass"] and suite_passed != suite_total:
            mandatory_fails.append(suite["name"])
            break

    # Create status message
    qa_status_message = generate_remarkup(status_data, args.distributed)

    # Create the report file
    qa_status_path = os.path.join(SCRIPT_DIR, ".quality_assurance_status")
    with open(qa_status_path, "w") as f:
        f.write(qa_status_message)

    # Create the measurements file
    measurements_path = os.path.join(SCRIPT_DIR, ".apollo_measurements")
    with open(measurements_path, "w") as f:
        f.write(measurements)

    print("Status is generated in %s" % qa_status_path)
    print("Measurements are generated in %s" % measurements_path)

    # Check if tests failed
    if mandatory_fails != []:
        sys.exit("Some tests that must pass have failed -- %s"
                 % str(mandatory_fails))


if __name__ == "__main__":
    main()