bunkerized-nginx/tests/LinuxTest.py

from Test import Test
from os.path import isdir, isfile
from os import getenv, mkdir, chmod
from shutil import rmtree
from traceback import format_exc
from subprocess import run
from time import sleep
from logger import setup_logger


class LinuxTest(Test):
    def __init__(self, name, timeout, tests, distro):
        super().__init__(name, "linux", timeout, tests)
        self._domains = {
            r"www\.example\.com": getenv("TEST_DOMAIN1"),
            r"auth\.example\.com": getenv("TEST_DOMAIN1"),
            r"app1\.example\.com": getenv("TEST_DOMAIN1_1"),
            r"app2\.example\.com": getenv("TEST_DOMAIN1_2"),
            r"app3\.example\.com": getenv("TEST_DOMAIN1_3"),
        }
        if not distro in ("ubuntu", "debian", "fedora", "centos"):
            raise Exception(f"unknown distro {distro}")
        self.__distro = distro
        self.__logger = setup_logger("Linux_test", getenv("LOGLEVEL", "INFO"))

    @staticmethod
    def init(distro):
        try:
            if not Test.init():
                return False
            # TODO : find the nginx uid/gid on Docker images
            proc = run("sudo chown -R root:root /tmp/bw-data", shell=True)
            if proc.returncode != 0:
                raise Exception("chown failed (autoconf stack)")
            if isdir("/tmp/linux"):
                rmtree("/tmp/linux")
            mkdir("/tmp/linux")
            chmod("/tmp/linux", 0o0777)
            cmd = f"docker run -p 80:80 -p 443:443 --rm --name linux-{distro} -d --tmpfs /tmp --tmpfs /run --tmpfs /run/lock -v /sys/fs/cgroup:/sys/fs/cgroup:ro bw-{distro}"
            proc = run(cmd, shell=True)
            if proc.returncode != 0:
                raise Exception("docker run failed (linux stack)")
            if distro in ("ubuntu", "debian"):
                cmd = "apt install -y /opt/\$(ls /opt | grep deb)"
            elif distro in ("centos", "fedora"):
                cmd = "dnf install -y /opt/\$(ls /opt | grep rpm)"
            proc = LinuxTest.docker_exec(distro, cmd)
            if proc.returncode != 0:
                raise Exception("docker exec apt install failed (linux stack)")
            proc = LinuxTest.docker_exec(distro, "systemctl start bunkerweb")
            if proc.returncode != 0:
                raise Exception("docker exec systemctl start failed (linux stack)")
            cp_dirs = {
                "/tmp/bw-data/letsencrypt": "/etc/letsencrypt",
                "/tmp/bw-data/cache": "/var/cache/bunkerweb",
            }
            for src, dst in cp_dirs.items():
                proc = LinuxTest.docker_cp(distro, src, dst)
                if proc.returncode != 0:
                    raise Exception(f"docker cp failed for {src} (linux stack)")
                proc = LinuxTest.docker_exec(distro, f"chown -R nginx:nginx {dst}/*")
                if proc.returncode != 0:
                    raise Exception(
                        f"docker exec failed for directory {src} (linux stack)"
                    )

            if distro in ("ubuntu", "debian"):
                LinuxTest.docker_exec(
                    distro,
                    "DEBIAN_FRONTEND=noninteractive apt-get install -y php-fpm unzip",
                )
                if distro == "ubuntu":
                    LinuxTest.docker_cp(
                        distro,
                        "./tests/www-deb.conf",
                        "/etc/php/8.1/fpm/pool.d/www.conf",
                    )
                    LinuxTest.docker_exec(
                        distro, "systemctl stop php8.1-fpm ; systemctl start php8.1-fpm"
                    )
                elif distro == "debian":
                    LinuxTest.docker_cp(
                        distro,
                        "./tests/www-deb.conf",
                        "/etc/php/7.4/fpm/pool.d/www.conf",
                    )
                    LinuxTest.docker_exec(
                        distro, "systemctl stop php7.4-fpm ; systemctl start php7.4-fpm"
                    )
            elif distro in ("centos", "fedora"):
                LinuxTest.docker_exec(distro, "dnf install -y php-fpm unzip")
                LinuxTest.docker_cp(
                    distro, "./tests/www-rpm.conf", "/etc/php-fpm.d/www.conf"
                )
                LinuxTest.docker_exec(
                    distro,
                    "mkdir /run/php ; chmod 777 /run/php ; systemctl stop php-fpm ; systemctl start php-fpm",
                )
            sleep(60)
        except:
            setup_logger("Linux_test", getenv("LOGLEVEL", "INFO")).error(
                f"exception while running LinuxTest.init()\n{format_exc()}",
            )
            return False
        return True

    @staticmethod
    def end(distro):
        ret = True
        try:
            if not Test.end():
                return False
            proc = run(f"docker kill linux-{distro}", shell=True)
            if proc.returncode != 0:
                ret = False
        except:
            setup_logger("Linux_test", getenv("LOGLEVEL", "INFO")).error(
                f"exception while running LinuxTest.end()\n{format_exc()}"
            )
            return False
        return ret

    def _setup_test(self):
        try:
            super()._setup_test()
            test = f"/tmp/tests/{self._name}"
            for ex_domain, test_domain in self._domains.items():
                Test.replace_in_files(test, ex_domain, test_domain)
                Test.rename(test, ex_domain, test_domain)
            Test.replace_in_files(test, "example.com", getenv("ROOT_DOMAIN"))
            proc = LinuxTest.docker_cp(self.__distro, test, f"/opt/{self._name}")
            if proc.returncode != 0:
                raise Exception("docker cp failed (test)")
            setup = test + "/setup-linux.sh"
            if isfile(setup):
                proc = LinuxTest.docker_exec(
                    self.__distro, f"cd /opt/{self._name} && ./setup-linux.sh"
                )
                if proc.returncode != 0:
                    raise Exception("docker exec setup failed (test)")
            proc = LinuxTest.docker_exec(
                self.__distro, f"cp /opt/{self._name}/variables.env /etc/bunkerweb/"
            )
            if proc.returncode != 0:
                raise Exception("docker exec cp variables.env failed (test)")
            proc = LinuxTest.docker_exec(
                self.__distro, "systemctl stop bunkerweb ; systemctl start bunkerweb"
            )
            if proc.returncode != 0:
                raise Exception("docker exec systemctl restart failed (linux stack)")
        except:
            self.__logger.error(
                f"exception while running LinuxTest._setup_test()\n{format_exc()}",
            )
            self._debug_fail()
            self._cleanup_test()
            return False
        return True

    def _cleanup_test(self):
        try:
            proc = LinuxTest.docker_exec(
                self.__distro,
                f"cd /opt/{self._name} ; ./cleanup-linux.sh ; rm -rf /etc/bunkerweb/configs/* ; rm -rf /etc/bunkerweb/plugins/*",
            )
            if proc.returncode != 0:
                raise Exception("docker exec rm failed (cleanup)")
            super()._cleanup_test()
        except:
            self.__logger.error(
                f"exception while running LinuxTest._cleanup_test()\n{format_exc()}",
            )
            return False
        return True

    def _debug_fail(self):
        LinuxTest.docker_exec(
            self.__distro,
            "cat /var/log/nginx/access.log ; cat /var/log/nginx/error.log ; journalctl -u bunkerweb --no-pager",
        )

    def docker_exec(distro, cmd_linux):
        return run(
            f'docker exec linux-{distro} /bin/bash -c "{cmd_linux}"',
            shell=True,
        )

    def docker_cp(distro, src, dst):
        return run(f"sudo docker cp {src} linux-{distro}:{dst}", shell=True)