validate.py

#!/usr/bin/python3
import argparse
import os
import sys
import subprocess
import traceback
import yaml

run_pycodestyle = False
try:
    import pycodestyle

    run_pycodestyle = True
except ImportError as e:
    print(e)
    print("Install pycodestyle: pip3 install pycodestyle")
    sys.exit(1)

try:
    import magic
except ImportError as e:
    print(e)
    print("Install python-magic: pip3 install python-magic")
    sys.exit(1)


def print_stderr(message):
    sys.stderr.write(message)
    sys.stderr.write("\n")


def publish_result(result_message_list, args):
    if result_message_list:
        result_message = "\n".join(result_message_list)
        try:
            f = open(args.result_file, "a")
            f.write(result_message)
            f.write("\n")
            f.close()
        except IOError as e:
            print(e)
            print_stderr("Cannot write to result file: %s" % args.result_file)
        if args.verbose:
            print_stderr(result_message)
    else:
        result_message = "\n".join(args.failed_message_list)
        print_stderr(result_message)


def detect_abi():
    # Retrieve the current canonical abi from
    # automated/lib/sh-test-lib:detect_abi
    return (
        subprocess.check_output(
            ". automated/lib/sh-test-lib && detect_abi && echo $abi", shell=True
        )
        .decode("utf-8")
        .strip()
    )


def pycodestyle_check(filepath, args):
    _fmt = "%(row)d:%(col)d: %(code)s %(text)s"
    options = {"ignore": args.pycodestyle_ignore, "show_source": True}
    pycodestyle_checker = pycodestyle.StyleGuide(options)
    fchecker = pycodestyle_checker.checker_class(
        filepath, options=pycodestyle_checker.options
    )
    fchecker.check_all()
    if fchecker.report.file_errors > 0:
        result_message_list = []
        result_message_list.append("* PYCODESTYLE: [FAILED]: " + filepath)
        fchecker.report.print_statistics()
        for line_number, offset, code, text, doc in fchecker.report._deferred_print:
            result_message_list.append(
                _fmt
                % {
                    "path": filepath,
                    "row": fchecker.report.line_offset + line_number,
                    "col": offset + 1,
                    "code": code,
                    "text": text,
                }
            )
        publish_result(result_message_list, args)
        args.failed_message_list = args.failed_message_list + result_message_list
        return 1
    else:
        if args.verbose:
            message = "* PYCODESTYLE: [PASSED]: " + filepath
            print_stderr(message)
    return 0


def validate_yaml_contents(filepath, args):
    def validate_testdef_yaml(y, args):
        result_message_list = []
        if "metadata" not in y.keys():
            result_message_list.append("* METADATA [FAILED]: " + filepath)
            result_message_list.append("\tmetadata section missing")
            publish_result(result_message_list, args)
            args.failed_message_list = args.failed_message_list + result_message_list
            exit(1)
        metadata_dict = y["metadata"]
        mandatory_keys = set(
            ["name", "format", "description", "maintainer", "os", "devices"]
        )
        if not mandatory_keys.issubset(set(metadata_dict.keys())):
            result_message_list.append("* METADATA [FAILED]: " + filepath)
            result_message_list.append(
                "\tmandatory keys missing: %s"
                % mandatory_keys.difference(set(metadata_dict.keys()))
            )
            result_message_list.append(
                "\tactual keys present: %s" % metadata_dict.keys()
            )
            publish_result(result_message_list, args)
            args.failed_message_list = args.failed_message_list + result_message_list
            return 1
        for key in mandatory_keys:
            if len(metadata_dict[key]) == 0:
                result_message_list.append("* METADATA [FAILED]: " + filepath)
                result_message_list.append("\t%s has no content" % key)
                publish_result(result_message_list, args)
                args.failed_message_list = (
                    args.failed_message_list + result_message_list
                )
                return 1
        # check if name has white spaces
        if metadata_dict["name"].find(" ") > -1:
            result_message_list.append("* METADATA [FAILED]: " + filepath)
            result_message_list.append("\t'name' contains whitespace")
            publish_result(result_message_list, args)
            args.failed_message_list = args.ailed_message_list + result_message_list
            return 1
        # check 'format' value
        if metadata_dict["format"] not in [
            "Lava-Test Test Definition 1.0",
            "Manual Test Definition 1.0",
        ]:
            result_message_list.append("* METADATA [FAILED]: " + filepath)
            result_message_list.append("\t'format' has incorrect value")
            publish_result(result_message_list, args)
            args.failed_message_list = args.failed_message_list + result_message_list
            return 1

        result_message_list.append("* METADATA [PASSED]: " + filepath)
        publish_result(result_message_list, args)
        return 0

    def validate_skipgen_yaml(filepath, args):
        abi = detect_abi()
        # Run skipgen on skipgen yaml file to check for output and errors
        skips = (
            subprocess.check_output(
                "automated/bin/{}/skipgen {}".format(abi, filepath), shell=True
            )
            .decode("utf-8")
            .strip()
        )
        if len(skips.split("\n")) < 1:
            message = "* SKIPGEN [FAILED]: " + filepath + " - No skips found"
            publish_result([message], args)
            args.failed_message_list.append(message)
            return 1
        publish_result(["* SKIPGEN [PASSED]: " + filepath], args)
        return 0

    filecontent = None
    try:
        with open(filepath, "r") as f:
            filecontent = f.read()
    except FileNotFoundError:
        publish_result(
            ["* YAMLVALIDCONTENTS [PASSED]: " + filepath + " - deleted"], args
        )
        return 0
    y = yaml.load(filecontent, Loader=yaml.FullLoader)
    if "run" in y.keys():
        # test definition yaml file
        return validate_testdef_yaml(y, args)
    elif "skiplist" in y.keys():
        # skipgen yaml file
        return validate_skipgen_yaml(filepath, args)
    else:
        publish_result(
            [
                "* YAMLVALIDCONTENTS [SKIPPED]: "
                + filepath
                + " - Unknown yaml type detected"
            ],
            args,
        )
        return 0


def validate_yaml(filename, args):
    filecontent = None
    try:
        with open(filename, "r") as f:
            filecontent = f.read()
    except FileNotFoundError:
        publish_result(["* YAMLVALID [PASSED]: " + filename + " - deleted"], args)
        return 0
    try:
        yaml.load(filecontent, Loader=yaml.FullLoader)
        if args.verbose:
            message = "* YAMLVALID: [PASSED]: " + filename
            print_stderr(message)
    except yaml.YAMLError:
        message = "* YAMLVALID: [FAILED]: " + filename
        result_message_list = []
        result_message_list.append(message)
        result_message_list.append("\n\n")
        exc_type, exc_value, exc_traceback = sys.exc_info()
        for line in traceback.format_exception_only(exc_type, exc_value):
            result_message_list.append(" " + line)
        publish_result(result_message_list, args)
        args.failed_message_list = args.failed_message_list + result_message_list
        return 1
    return 0


def validate_shell(filename, ignore_options):
    ignore_string = ""
    if args.shellcheck_ignore is not None:
        # Exclude types of warnings in the following format:
        # -e CODE1,CODE2..
        ignore_string = "-e %s" % ",".join(args.shellcheck_ignore)
    if len(ignore_string) < 4:  # contains only "-e "
        ignore_string = ""
    cmd = "shellcheck %s" % ignore_string
    return validate_external(cmd, filename, "SHELLCHECK", args)


def validate_php(filename, args):
    cmd = "php -l"
    return validate_external(cmd, filename, "PHPLINT", args)


def validate_external(cmd, filename, prefix, args):
    final_cmd = "%s %s 2>&1" % (cmd, filename)
    status, output = subprocess.getstatusoutput(final_cmd)
    if status == 0:
        message = "* %s: [PASSED]: %s" % (prefix, filename)
        publish_result([message], args)
    else:
        result_message_list = []
        result_message_list.append("* %s: [FAILED]: %s" % (prefix, filename))
        result_message_list.append("* %s: [OUTPUT]:" % prefix)
        for line in output.splitlines():
            result_message_list.append(" " + line)
        publish_result(result_message_list, args)
        args.failed_message_list = args.failed_message_list + result_message_list
        return 1
    return 0


def validate_file(args, path):
    if args.verbose:
        print("Validating file: %s" % path)
    filetype = magic.from_file(path, mime=True)
    exitcode = 0
    # libmagic takes yaml as 'text/plain', so use file extension here.
    if path.endswith((".yaml", ".yml")):
        exitcode = validate_yaml(path, args)
        if exitcode == 0:
            # if yaml isn't valid there is no point in checking metadata
            exitcode = validate_yaml_contents(path, args)
    elif run_pycodestyle and filetype == "text/x-python":
        exitcode = pycodestyle_check(path, args)
    elif filetype == "text/x-php":
        exitcode = validate_php(path, args)
    elif path.endswith(".sh") or filetype == "text/x-shellscript":
        exitcode = validate_shell(path, args)
    else:
        publish_result(
            [
                "* UNKNOWN [SKIPPED]: "
                + path
                + " - Unknown file type detected: "
                + filetype
            ],
            args,
        )
        return 0
    return exitcode


def run_unit_tests(args, filelist=None):
    exitcode = 0
    if filelist is not None:
        for filename in filelist:
            tmp_exitcode = validate_file(args, filename)
            if tmp_exitcode != 0:
                exitcode = 1
    else:
        for root, dirs, files in os.walk("."):
            if not root.startswith("./.git"):
                for name in files:
                    tmp_exitcode = validate_file(args, root + "/" + name)
                    if tmp_exitcode != 0:
                        exitcode = 1
    return exitcode


def main(args):
    exitcode = 0
    if args.git_latest:
        # check if git exists
        git_status, git_result = subprocess.getstatusoutput(
            "git diff --name-only HEAD~1"
        )
        if git_status == 0:
            filelist = git_result.split()
            exitcode = run_unit_tests(args, filelist)
    elif len(args.file_path) > 0:
        exitcode = run_unit_tests(args, [args.file_path])
    else:
        exitcode = run_unit_tests(args)
    if not args.verbose:
        publish_result(None, args)
    exit(exitcode)


if __name__ == "__main__":
    failed_message_list = []
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-p",
        "--pycodestyle-ignore",
        nargs="*",
        default=["E501"],
        help="Space separated list of pycodestyle exclusions",
        dest="pycodestyle_ignore",
    )
    parser.add_argument(
        "-s",
        "--shellcheck-ignore",
        nargs="*",
        help="Space separated list of shellcheck exclusions",
        dest="shellcheck_ignore",
    )
    parser.add_argument(
        "-g",
        "--git-latest",
        action="store_true",
        default=False,
        help="If set, the script will try to evaluate files in last git \
                            commit instead of the whole repository",
        dest="git_latest",
    )
    parser.add_argument(
        "-f",
        "--file-path",
        default="",
        help="Path to the file that should be checked",
        dest="file_path",
    )
    parser.add_argument(
        "-r",
        "--result-file",
        default="build-error.txt",
        help="Path to the file that contains results in case of failure",
        dest="result_file",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        default=False,
        help="Make output more verbose",
        dest="verbose",
    )

    args = parser.parse_args()
    setattr(args, "failed_message_list", failed_message_list)
    main(args)