path: root/test-suite

#!/usr/bin/python3

import argparse
import json
import logging
import os
import shutil
import subprocess
import sys
import tempfile
import yaml


class Suite:
    def __init__(self, tmp):
        self.tmp = tmp
        self.rad = Rad(os.path.join(tmp, "dot-radicle"))
        self.rad.auth()

        self.config = Config(tmp)
        self.config.write()

    def assert_triggered(self, msg):
        assert msg["response"] == "triggered"

    def assert_success(self, msg):
        assert msg == {"response": "finished", "result": "success"}

    def assert_failure(self, msg):
        assert msg["response"] == "finished"
        assert msg["result"] == "failure"

    def assert_error(self, msg):
        assert msg["response"] == "finished"
        res = msg["result"]
        assert isinstance(res, dict)
        assert "error" in res

    def run_all_test_cases(self, tests):
        methods = self._test_methods()
        chosen = self._chosen(methods, tests)
        for meth in methods:
            if meth in chosen:
                info(f"test case {meth}")
                getattr(self, meth)()
            else:
                info(f"skipping test {meth}")

    def _test_methods(self):
        return [meth for meth in dir(self) if meth.startswith("test_")]

    def _chosen(self, methods, tests):
        if tests:
            chosen = set()
            for test in tests:
                for meth in methods:
                    if test in meth:
                        chosen.add(meth)
        else:
            chosen = set(methods)
        return chosen

    def _create_git_repo(self, repo_name):
        git = Git(os.path.join(self.tmp, repo_name))
        git.init()
        git.write("README.md", "README")
        git.commit("first commit")
        return git

    def _create_valid_native_yaml(self, git, shell):
        native = NativeYaml(shell)
        git.write(".radicle/native.yaml", native.yaml())
        git.commit("add native.yaml")

    def _get_repo_info(self, git):
        commit = git.head()
        self.rad.init(git)
        rid = self.rad.rid(git)
        return rid, commit

    def _setup(self, repo_name, shell):
        git = self._create_git_repo(repo_name)
        self._create_valid_native_yaml(git, shell)
        return self._get_repo_info(git)

    def _create_ci(self):
        return NativeCI(self.rad, self.config)

    def _create_valid_trigger(self, rid, commit):
        return Trigger(rid, commit)

    def _run_ci(self, trigger, ci, rid, commit):
        exit, resps, stderr = ci.run(trigger)
        debug(f"exit: {exit}")
        debug(f"responses: {resps}")
        return exit, resps, stderr

    def _test_case(self, repo_name, shell):
        rid, commit = self._setup(repo_name, shell)
        ci = self._create_ci()
        trigger = self._create_valid_trigger(rid, commit)
        return self._run_ci(trigger, ci, rid, commit)

    def test_happy_path(self):
        exit, resps, stderr = self._test_case("happy-path", "echo hello, world")
        assert exit == 0
        assert len(resps) == 2
        self.assert_triggered(resps[0])
        self.assert_success(resps[1])

    def test_without_config_env_var(self):
        git = self._create_git_repo("no-config")
        self._create_valid_native_yaml(git, "echo hello world")
        rid, commit = self._get_repo_info(git)
        trigger = Trigger(rid, commit)
        ci = self._create_ci()
        ci.without_config()
        exit, resps, stderr = ci.run(trigger)
        debug(f"exit: {exit}")
        debug(f"responses: {resps}")
        assert exit != 0
        assert len(resps) == 0
        assert "RADICLE_NATIVE_CI" in stderr

    def test_config_missing(self):
        git = self._create_git_repo("config-missing")
        self._create_valid_native_yaml(git, "echo hello world")
        rid, commit = self._get_repo_info(git)
        trigger = Trigger(rid, commit)
        cfg2 = Config("no-config-file")
        ci = NativeCI(self.rad, cfg2)
        exit, resps, stderr = ci.run(trigger)
        debug(f"exit: {exit}")
        debug(f"responses: {resps}")
        assert exit != 0
        assert len(resps) == 0
        assert "no-config-file" in stderr

    def test_command_fails(self):
        exit, resps, stderr = self._test_case("cmd-fails", "false")
        assert exit == 1
        assert len(resps) == 2
        self.assert_triggered(resps[0])
        self.assert_failure(resps[1])

    def test_repository_does_not_exist(self):
        rid = "rad:z3aaaaaaaaaaaaaaaaaaaaaaaaaaa"
        commit = "8d947e182b096ec009e1c9eda9e6a67f5eef83d9"

        ci = self._create_ci()
        trigger = self._create_valid_trigger(rid, commit)
        exit, resps, stderr = self._run_ci(trigger, ci, rid, commit)

        assert exit == 2

    def test_commit_does_not_exist(self):
        git = self._create_git_repo("commit-missing")
        self._create_valid_native_yaml(git, "echo hello world")
        rid, _commit = self._get_repo_info(git)
        commit = "0000000000000000000000000000000000000000"
        trigger = Trigger(rid, commit)
        ci = self._create_ci()
        exit, resps, stderr = ci.run(trigger)

        assert exit == 1
        assert len(resps) == 2
        self.assert_triggered(resps[0])
        self.assert_failure(resps[1])

    def test_no_message(self):
        git = self._create_git_repo("no-message")
        self._create_valid_native_yaml(git, "echo hello world")
        ci = self._create_ci()
        exit, resps, stderr = ci.run_without_request()

        assert exit != 0
        assert len(resps) == 0

    def test_malformed_trigger(self):
        git = self._create_git_repo("malformed-trigger")
        self._create_valid_native_yaml(git, "echo hello world")
        rid, _commit = self._get_repo_info(git)
        trigger = MalformedTrigger()
        ci = self._create_ci()
        exit, resps, stderr = ci.run(trigger)

        assert exit != 0
        assert len(resps) == 0

    def test_native_yaml_has_no_shell(self):
        exit, resps, stderr = self._test_case("no-shell", None)
        assert exit == 1
        assert len(resps) == 2
        self.assert_triggered(resps[0])
        self.assert_failure(resps[1])

    def test_native_yaml_shell_is_not_string(self):
        exit, resps, stderr = self._test_case("shell-not-string", {"foo": "bar"})
        assert exit == 1
        assert len(resps) == 2
        self.assert_triggered(resps[0])
        self.assert_failure(resps[1])

    def test_command_takes_too_long(self):
        git = self._create_git_repo("command-takes-too-long")
        self._create_valid_native_yaml(git, "sleep 5")
        rid, commit = self._get_repo_info(git)
        trigger = Trigger(rid, commit)
        ci = self._create_ci()
        exit, resps, stderr = ci.run(trigger)
        assert exit == 1
        assert len(resps) == 2
        self.assert_triggered(resps[0])
        self.assert_failure(resps[1])
        assert "124" in stderr


class Git:
    def __init__(self, path):
        self.path = path

    def filename(self, relative):
        return os.path.join(self.path, "./" + relative)

    def write(self, relative, data):
        write(self.filename(relative), data)

    def _git(self, argv):
        return run(argv, cwd=self.path).stdout

    def init(self):
        assert not os.path.exists(self.path)
        run(["git", "init", self.path])
        debug(f"created git repository at {self.path}")

    def commit(self, msg):
        run(["git", "add", "."], cwd=self.path)
        run(["git", "commit", "-m", msg], cwd=self.path)

    def head(self):
        return self._git(["git", "rev-parse", "HEAD"]).strip()

    def ls_files(self):
        out = self._git(["git", "ls-files"])
        debug(f"ls-files:\n{out}")


class Rad:
    PASSPHRASE = "xyzzy"

    def __init__(self, rad_home):
        assert rad_home is not None
        self.rad_home = rad_home

    def _rad(self, argv, env=None, cwd=None):
        if env is None:
            env = {}
        env["RAD_HOME"] = self.rad_home
        env["RAD_PASSPHRASE"] = self.PASSPHRASE
        return run(argv, env=env, cwd=cwd).stdout

    def auth(self):
        self._rad(["rad", "auth", "--alias=test-node"])
        debug(f"created Radicle node at {self.rad_home}")

    def init(self, git):
        name = os.path.basename(git.path)
        self._rad(
            [
                "rad",
                "init",
                f"--name={name}",
                f"--description=test-repo",
                f"--public",
                f"--no-confirm",
            ],
            cwd=git.path,
        )
        debug(f"added git repository {git.path} to node")

    def rid(self, git):
        return self._rad(["rad", "."], cwd=git.path).strip()


class Config:
    def __init__(self, tmp):
        self.path = os.path.join(tmp, "config.yaml")
        self.dict = {
            "state": os.path.join(tmp, "state"),
            "log": os.path.join(tmp, "node-log.txt"),
            "timeout": 2,
        }

    def write(self):
        write(self.path, self.yaml())

    def yaml(self):
        return yaml.safe_dump(self.dict, indent=4)


class NativeYaml:
    def __init__(self, shell):
        self.dict = {}
        if shell is not None:
            self.dict["shell"] = shell

    def yaml(self):
        return yaml.safe_dump(self.dict, indent=4)


class Trigger:
    def __init__(self, rid, commit):
        self.rid = rid
        self.commit = commit

    def json(self):
        return json.dumps(
            {"request": "trigger", "repo": self.rid, "commit": self.commit}
        )


class MalformedTrigger:
    def json(self):
        return json.dumps({"request": "trigger"})


class NativeCI:
    def __init__(self, rad, config):
        self.rad_home = rad.rad_home
        self.config = config.path
        self.env = {
            "RAD_HOME": self.rad_home,
            "RADICLE_NATIVE_CI": self.config,
            "RADICLE_NATIVE_CI_LOG": "debug",
        }
        self.timeout = 1

    def without_config(self):
        del self.env["RADICLE_NATIVE_CI"]

    def run(self, request):
        p = run(
            [
                "cargo",
                "run",
                "-q",
            ],
            input=request.json(),
            env=dict(self.env),
            may_fail=True,
        )
        resps = [json.loads(line.strip()) for line in p.stdout.splitlines()]
        return p.returncode, resps, p.stderr

    def run_without_request(self):
        p = run(
            [
                "cargo",
                "run",
                "-q",
            ],
            stdin=subprocess.DEVNULL,
            env=dict(self.env),
            may_fail=True,
        )
        resps = [json.loads(line.strip()) for line in p.stdout.splitlines()]
        return p.returncode, resps, p.stderr


def debug(msg):
    logging.debug(msg)


def info(msg):
    logging.info(msg)


def die(msg, exc_info=None):
    logging.critical(msg, exc_info=exc_info)
    sys.exit(1)


def run(argv, env=None, cwd=None, stdin=None, input=None, may_fail=False):
    debug(f"run {argv} with env={env}, cwd={cwd} input={input!r}")
    if env is not None:
        for name, value in os.environ.items():
            if name not in env:
                env[name] = value
    p = subprocess.run(
        argv, capture_output=True, env=env, cwd=cwd, stdin=stdin, input=input, text=True
    )

    debug(f"stdout:\n{indent(p.stdout)}")
    debug(f"stderr:\n{indent(p.stderr)}")
    if p.returncode != 0 and not may_fail:
        die(f"command failed: {argv}")

    return p


def indent(s):
    return "".join([f"    {line}\n" for line in s.splitlines()])


def write(filename, data):
    dirname = os.path.dirname(filename)
    if not os.path.exists(dirname):
        os.makedirs(dirname)
    with open(filename, "w") as f:
        f.write(data)


def main(tmp, tests):
    suite = Suite(tmp)
    suite.run_all_test_cases(tests)


p = argparse.ArgumentParser()
p.add_argument("--verbose", action="store_true")
p.add_argument("--test", action="append", nargs="?", dest="tests")
args = p.parse_args()

if args.verbose:
    level = logging.DEBUG
    print(args)
else:
    level = logging.INFO


logging.basicConfig(
    level=level,
    stream=sys.stderr,
    format="%(levelname)s %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

tmp = tempfile.mkdtemp()
debug(f"created temporary directory {tmp}")

try:
    main(tmp, args.tests)
except AssertionError as e:
    debug(f"removing temporary directory {tmp}")
    shutil.rmtree(tmp)
    die(f"ERROR: assertion failed: {e}", exc_info=True)
except Exception as e:
    debug(f"removing temporary directory {tmp}")
    shutil.rmtree(tmp)
    die(f"ERROR: uncaught exception {e}", exc_info=True)
else:
    debug(f"removing temporary directory {tmp}")
    shutil.rmtree(tmp)