path: root/check

#!/usr/bin/env python3

import argparse
import glob
import json
import os
import sys
import time

from subprocess import PIPE, DEVNULL, STDOUT


class Runcmd:
    """Run external commands in various ways"""

    def __init__(self, verbose, progress, offline):
        self._verbose = verbose
        self._progress = progress
        self.offline = offline
        # Deliberately chosen because it's 12:45 / 13:45 offset from UTC
        # As such it ought to show any TZ related errors if we're lucky.
        self._env = {"TZ": "NZ-CHAT"}

    def _write_msg(self, msg):
        sys.stdout.write(f"{msg}\n")
        sys.stdout.flush()

    def setenv(self, key, value):
        self._env[key] = value

    def msg(self, msg):
        if self._verbose:
            self._write_msg(msg)

    def title(self, title):
        if self._verbose:
            self._write_msg("")
            self._write_msg("=" * len(title))
            self._write_msg(title)
            self._write_msg("=" * len(title))
        elif self._progress:
            self._write_msg(title)

    def runcmd_unchecked(self, argv, **kwargs):
        """Run a command (generic version)

        Return a subcommand.CompletedProcess. It's the caller's duty
        check that the command succeeded.

        All actual execution of other programs happens via this method.
        However, only methods of this class should ever call this
        method.
        """

        # Import "run" here so that no other part of the code can see the
        # symbol.
        from subprocess import run

        self.msg(f"RUN: {argv} {kwargs}")

        if not self._verbose:
            if "stdout" not in kwargs:
                kwargs["stdout"] = PIPE
            if "stderr" not in kwargs:
                kwargs["stderr"] = STDOUT

        assert "key" not in kwargs
        env = dict(os.environ)
        env.update(self._env)

        return run(argv, env=env, **kwargs)

    def runcmd(self, argv, **kwargs):
        """Run a command (generic version)

        If the command fails, terminate the program. On success, return
        a subprocess.CompletedProcess.
        """

        p = self.runcmd_unchecked(argv, **kwargs)
        if p.returncode != 0:
            sys.stdout.write((p.stdout or b"").decode("UTF-8"))
            sys.stderr.write((p.stderr or b"").decode("UTF-8"))
            sys.stderr.write(f"Command {argv} failed\n")
            sys.exit(p.returncode)
        return p

    def runcmd_maybe(self, argv, **kwargs):
        """Run a command it's availbe, or quietly do no nothing"""
        if self.got_command(argv[0]):
            self.runcmd(argv, **kwargs)

    def got_command(self, name):
        """Is a command of a given name available?"""
        p = self.runcmd_unchecked(["which", name], stdout=DEVNULL)
        return p.returncode == 0

    def cargo(self, args, **kwargs):
        """Run cargo with arguments."""
        return self.runcmd(["cargo"] + args, **kwargs)

    def cargo_maybe(self, args, **kwargs):
        """Run cargo if the desired subcommand is available"""
        if self.got_cargo(args[0]):
            return self.runcmd(["cargo"] + args, **kwargs)

    def got_cargo(self, subcommand):
        """Is a cargo subcommand available?"""
        p = self.runcmd(["cargo", "--list"], check=True, stdout=PIPE)
        lines = [
            line.split()[0]
            for line in p.stdout.decode("UTF-8").splitlines()
            if line.strip()
        ]
        return subcommand in lines

    def codegen(self, md, template, output, **kwargs):
        """Run the Subplot code generator and the test program it produces"""
        self.cargo(
            [
                "run",
                "--package=subplot",
                "--bin=subplot",
                "--",
                f"--resources={os.path.abspath('share')}",
                "codegen",
                f"--template={template}",
                md,
                f"--output={output}",
            ],
            **kwargs,
        )

    def docgen(self, md, template, output, **kwargs):
        """Run the Subplot document generator"""

        # GitLab CI, and other CI engines, runs tests under Docker, which uses
        # an overlay file system, which at least sometimes only has full-second
        # time stamps in file meta data, and sometimes that seems to happen
        # when the kernel needs to flush its inode cache, and so it's not very
        # deterministic. This leads to occasional test failures. To prevent
        # such flaky tests, we wait for a second to make sure the time stamp of
        # the output will be newer than the input has.
        #
        # This is an ugly kluge. It would be possible to check if the sleep is
        # needed, but that kind of code is going to be tricky. Best avoid
        # tricky code in test suites. Test code should be obviously correct.
        time.sleep(1)

        self.cargo(
            [
                "run",
                "--package=subplot",
                "--bin=subplot",
                "--",
                f"--resources={os.path.abspath('share')}",
                "docgen",
                f"--template={template}",
                md,
                f"--output={output}",
            ],
            **kwargs,
        )

    def libdocgen(self, bindings, output):
        self.cargo(
            [
                "run",
                "--package=subplot",
                "--bin=subplot",
                "--",
                f"--resources={os.path.abspath('share')}",
                "libdocgen",
                "--output",
                output,
                bindings,
            ]
        )

    def get_templates(self, filename, strict):
        args = [
            "run",
            "--quiet",
            "--package=subplot",
            "--bin=subplot",
            "--",
            f"--resources={os.path.abspath('share')}",
            "metadata",
            "-o",
            "json",
        ]
        if not strict:
            args += ["--merciful"]
        args += [filename]
        metadata = self.cargo(
            args,
            stdout=PIPE,
            stderr=PIPE,
        ).stdout.decode("UTF-8")
        metadata = json.loads(metadata)
        impls = metadata.get("impls", {})
        if not impls:
            sys.exit(f"{filename} does not specify a template")
        impl_names = [name for name in impls.keys()]
        return impl_names


def find_files(pattern, pred):
    """Find files recursively, if they are accepted by a predicate function"""
    return [f for f in glob.glob(pattern, recursive=True) if pred(f)]


def check_python(r):
    """Run all checks for Python code"""
    r.title("checking Python code")

    # Find all Python files anywhere, except those we know aren't proper Python.
    py = find_files(
        "**/*.py",
        lambda f: os.path.basename(f) not in ("template.py", "test.py")
        and "test-outputs" not in f,
    )

    # Test with flake8 if available. Flake8 finds Python files itself.
    r.runcmd_maybe(["flake8", "--config=flake8.ini", "check"] + py)

    # Check formatting with Black. We need to provide the files to Python
    # ourselves.
    # r.runcmd_maybe(["black", "--check"] + py)

    # Find and run unit tests.
    tests = find_files("**/*_tests.py", lambda f: True)
    for test in tests:
        dirname = os.path.dirname(test)
        test = os.path.basename(test)
        r.runcmd(["python3", test], cwd=dirname)


def check_shell(r):
    """Run all checks for shell code"""
    r.title("checking shell code")

    # Find all shell files anywhere, except generated test programs.
    sh = find_files(
        "**/*.sh",
        lambda f: os.path.basename(f) != "test.sh" and "test-outputs" not in f,
    )

    r.runcmd_maybe(["shellcheck"] + sh)


def check_rust(r, strict=False, sloppy=False):
    """Run all checks for Rust code"""
    r.title("checking Rust code")

    argv = ["cargo", "build", "--workspace", "--all-targets"]
    if r.offline:
        argv.append("--offline")
    r.runcmd(argv)

    if r.got_cargo("clippy") and not sloppy:
        argv = [
            "cargo",
            "clippy",
            "--workspace",
            "--all-targets",
        ]
        if strict:
            argv += [
                "--",
                "-Dwarnings",
            ]
        r.runcmd(argv)
    elif strict:
        sys.exit("Strict Rust checks specified, but clippy was not found")

    r.runcmd(["cargo", "test", "--workspace"])
    if not sloppy:
        r.runcmd(["cargo", "fmt", "--", "--check"])


def check_subplots(r, strict=False):
    """Run all Subplots and generate documents for them"""
    output = os.path.abspath("test-outputs")
    os.makedirs(output, exist_ok=True)

    subplots = find_files(
        "**/*.subplot",
        lambda f: f == f.lower() and "subplotlib" not in f and "test-outputs" not in f,
    )
    if r.offline:
        r.msg("Only testing subplot.subplot due to --offline")
        subplots = ["subplot.subplot"]
    for subplot0 in subplots:
        r.title(f"checking subplot {subplot0}")

        dirname = os.path.dirname(subplot0) or "."
        subplot = os.path.basename(subplot0)
        base, _ = os.path.splitext(subplot)

        doc_template = None

        for template in r.get_templates(subplot0, strict):
            if doc_template is None:
                doc_template = template
            if template == "python":
                test_py = os.path.join(output, f"test-{base}.py")
                test_log = os.path.join(output, f"test-{base}.log")

                # Remove test log from previous run, if any.
                if os.path.exists(test_log):
                    os.remove(test_log)

                bindir = get_bin_dir(r)

                r.codegen(subplot, "python", test_py, cwd=dirname)
                p = r.runcmd_unchecked(
                    [
                        "python3",
                        test_py,
                        "--log",
                        test_log,
                        f"--env=SUBPLOT_DIR={bindir}",
                    ],
                    cwd=dirname,
                )
                if p.returncode != 0:
                    if os.path.exists(test_log):
                        tail(test_log)
                    sys.exit(1)
            elif template == "bash":
                test_sh = os.path.join(output, f"test-{base}.sh")
                r.codegen(subplot, "bash", test_sh, cwd=dirname)
                r.runcmd(["bash", "-x", test_sh], cwd=dirname)
            elif template == "rust":
                r.msg(f"Ignoring Rust template in {subplot0}")
            else:
                sys.exit(f"unknown template {template} in {subplot0}")

        base = os.path.basename(subplot)
        base, _ = os.path.splitext(subplot)
        base = os.path.join(output, base)
        html = base + ".html"
        r.docgen(subplot, doc_template, html, cwd=dirname)
        r.runcmd(["tidy", "-errors", html], cwd=dirname)


def tail(filename, numlines=100):
    lines = []
    with open(filename) as f:
        for line in f.readlines():
            lines.append(line)
            lines = lines[-numlines:]

    print(f"last {len(lines)} of {filename}:")
    for line in lines:
        print(f"    {line.rstrip()}")


def check_tooling(r):
    """Check build environment for tooling the test suite needs"""
    commands = [
        "bash",
        "cargo",
        "dot",
        "plantuml",
        "rustc",
        "rustfmt",
        "tidy",
    ]
    for command in commands:
        if not r.got_command(command):
            sys.exit(f"can't find {command}, which is needed for test suite")

    if not r.got_command("daemonize") and not r.got_command("/usr/sbin/daemonize"):
        sys.exit(
            "can't find daemonize in PATH or in /usr/sbin, but it's needed for test suite"
        )


def check_doc(r):
    docs = os.path.join("test-outputs", "libdocs")
    if not os.path.exists(docs):
        os.mkdir(docs)

    bindings = []
    for dirname, _, basenames in os.walk("share"):
        dirname = dirname[len("share/") :]
        bindings += [
            os.path.join(dirname, x)
            for x in basenames
            if x != "template.yaml" and x.endswith(".yaml")
        ]

    for filename in bindings:
        md = os.path.splitext(os.path.basename(filename))[0] + ".md"
        md = os.path.join(docs, md)
        r.libdocgen(filename, md)


def parse_args():
    """Parse command line arguments to this script"""
    p = argparse.ArgumentParser()
    p.add_argument("-v", dest="verbose", action="store_true", help="be verbose")
    p.add_argument(
        "-p", dest="progress", action="store_true", help="print some progress output"
    )
    p.add_argument(
        "--strict", action="store_true", help="don't allow compiler warnings"
    )
    p.add_argument(
        "--sloppy", action="store_true", help="don't check formatting or with clippy"
    )
    p.add_argument(
        "--offline", action="store_true", help="only run tests that can be run offline"
    )

    all_whats = ["tooling", "python", "shell", "rust", "subplots", "doc"]
    p.add_argument(
        "what", nargs="*", default=all_whats, help=f"what to test: {all_whats}"
    )
    return p.parse_args()


def get_bin_dir(r):
    p = r.runcmd(
        ["cargo", "metadata", "--format-version=1", "--frozen", "--no-deps"],
        check=True,
        stdout=PIPE,
    )
    obj = json.loads(p.stdout)
    return os.path.join(obj["target_directory"], "debug")


def main():
    """Main program"""
    args = parse_args()

    r = Runcmd(args.verbose, args.progress, args.offline)
    r.setenv("PYTHONDONTWRITEBYTECODE", "1")

    for what in args.what:
        if what == "python":
            check_python(r)
        elif what == "shell":
            check_shell(r)
        elif what == "rust":
            check_rust(r, strict=args.strict, sloppy=args.sloppy)
        elif what == "subplots":
            check_subplots(r, strict=args.strict)
        elif what == "tooling":
            check_tooling(r)
        elif what == "doc":
            check_doc(r)
        else:
            sys.exit(f"Unknown test {what}")

    sys.stdout.write("Everything seems to be in order.\n")


main()