diff --git a/examples/authelia/tests.json b/examples/authelia/tests.json new file mode 100644 index 000000000..877336620 --- /dev/null +++ b/examples/authelia/tests.json @@ -0,0 +1,22 @@ +{ + "name": "authelia", + "kinds": [ + "docker", + "autoconf", + "swarm", + "kubernetes" + ], + "timeout": 60, + "tests": [ + { + "type": "string", + "url": "https://TEST_DOMAIN1_1 + "string": "authelia" + }, + { + "type": "string", + "url": "https://TEST_DOMAIN1_2 + "string": "authelia" + } + ] +} \ No newline at end of file diff --git a/tests/new/DockerTest.py b/tests/new/DockerTest.py new file mode 100644 index 000000000..b15325671 --- /dev/null +++ b/tests/new/DockerTest.py @@ -0,0 +1,33 @@ +from Test import Test +from os.path import isdir, join +from os import chown, walk +from traceback import format_exc + +class DockerTest(Test) : + + def __init__(self, name, timeout, tests) : + super().__init__(name, "docker", timeout, tests) + + def init() : + try : + super().init() + for root, dirs, files in walk("/tmp/bw-data") : + for name in dirs + files : + chown(join(root, name), 101, 101) + except : + self._log("exception while running DockerTest.init()\n" + format_exc(), error=True) + return False + return True + + def _setup_test(self) : + try : + super()._setup_test() + # TODO : + except : + self._log("exception while running DockerTest._setup_test()\n" + format_exc(), error=True) + return False + return True + + + def _cleanup_test(self) : + pass \ No newline at end of file diff --git a/tests/new/Test.py b/tests/new/Test.py new file mode 100644 index 000000000..c56e30dc9 --- /dev/null +++ b/tests/new/Test.py @@ -0,0 +1,111 @@ +from abc import ABC, abstractmethod +from sys import stderr +from time import time, sleep +from requests import get +from traceback import format_exc +from shutil import rmtree +from os.path import isdir, join +from os import mkdir, makedirs, walk + +class Test(ABC) : + + def __init__(self, name, kind, timeout, tests) : + self._name = name + self.__kind = kind + self.__timeout = timeout + self.__tests = tests + self._log("instiantiated with " + str(len(tests)) + " tests and timeout of " + str(timeout) + "s") + + def _log(self, msg, error=False) : + when = datetime.datetime.today().strftime("[%Y-%m-%d %H:%M:%S]") + what = self._name + " - " + self.__kind + " - " + msg + if error : + print(when + " " + what, flush=True, file=stderr) + else : + print(when + " " + what, flush=True) + + # Class method + # called once before running all the different tests for a given integration + # must be override if specific actions needs to be done + def init() : + try : + if not isdir("/tmp/bw-data") : + mkdir("/tmp/bw-data") + chmod("/tmp/bw-data", 0o777) + rm_dirs = ["configs", "plugins", "www"] + for rm_dir in rm_dirs : + if isdir(rm_dir) : + rmtree("/tmp/bw-data/" + rm_dir, ignore_errors=True) + except : + self._log("exception while running Test.init()\n" + format_exc(), error=True) + return False + return True + + # called before starting the tests + # must be override if specific actions needs to be done + def _setup_test(self) : + try : + if isdir("/tmp/tests/" + self._name) : + rmtree("/tmp/tests/" + self._name) + makedirs("/tmp/tests/" + self._name, exist_ok=True) + except : + self._log("exception while running Test._setup_test()\n" + format_exc(), error=True) + return False + return True + + # called after running the tests + # must be override if specific actions needs to be done + def _cleanup_test(self) : + try : + rmtree("/tmp/tests/" + self._name) + except : + self._log("exception while running Test._cleanup_test()\n" + format_exc(), error=True) + return False + return True + + # run all the tests + def run_tests(self) : + if not self._setup_test() : + return False + start = time() + while time() < start + self.__timeout : + all_ok = True + for test in self.__tests : + ok = self.__run_test(test) + if not ok : + all_ok = False + break + if all_ok : + elapsed = str(int(time() - start)) + self._log("success (" + elapsed + "/" + str(self.__timeout) + "s)") + return self._cleanup_test() + self._log("tests no ok, retrying in 1s ...", error=True) + sleep(1) + self._log("failed (timeout = " + str(self.__timeout) + "s)", error=True) + self._cleanup_test() + return False + + # run a single test + def __run_test(self, test) : + try : + if test["type"] == "string" : + r = get(test["url"], timeout=5) + return test["string"] in r.text + except : + self._log("exception while running test of type " + test["type"] + " on URL " + test["url"] + "\n" + format_exc(), error=True) + raise("unknow test type " + test["type"]) + + def _replace_in_file(path, old, new) : + with open(path, "r") as f : + content = f.read() + content = content.replace(old, new) + with open(path, "w") as f : + f.write(content) + + def _rename(path, old, new) : + for root, dirs, files in walk(path) : + for name in dirs + files : + full_path = join(root, name) + if old in full_path : + new_path = full_path.replace(old, new) + rename(full_path, new_path) \ No newline at end of file