morethantext-web/tests/step_defs/server.py

"""Fixture for setting up a single server."""

from asyncio import create_subprocess_exec, get_event_loop, sleep
from json import dumps
from pathlib import Path
from random import choices
from shutil import rmtree
from socket import gethostname, gethostbyname, socket
from string import ascii_lowercase


class Server:
    """Setup a single server."""

    def __init__(self):
        """Initialization of a server."""
        self.datadir = Path.cwd().joinpath(
            "tests", "data", "".join(choices(ascii_lowercase, k=5))
        )
        self.datadir.mkdir(parents=True)
        self.settings = {}
        self.use_config = True
        # self.env = {}
        self.process = None
        self.loop = get_event_loop()

    @property
    def base_url(self):
        """Returns the base URL for the server."""
        address = "127.0.0.1"
        port = 9090
        if "address" in self.settings:
            address = self.settings["address"]
        if "port" in self.settings:
            port = self.settings["port"]
        return f"{address}:{port}"

    async def __start(self):
        """async start of the server."""
        env = None
        if self.settings:
            if self.use_config:
                with open(
                    self.datadir.joinpath("morethantext.json"), "w", encoding="utf-8"
                ) as cfg:
                    cfg.write(dumps(self.settings))
            else:
                env = {}
                for key, value in self.settings.items():
                    env[f"MTT_{key.upper()}"] = str(value)
        self.process = await create_subprocess_exec(
            Path.cwd().joinpath("target", "release", "morethantext_web"),
            env=env,
            cwd=self.datadir,
        )
        await sleep(1)

    async def __stop(self):
        """async stop of the server."""
        if self.process is not None and self.process.returncode is None:
            self.process.terminate()
            await self.process.wait()

    def start(self):
        """Start the server."""
        self.loop.run_until_complete(self.__start())

    def stop(self):
        """Stop the server."""
        self.loop.run_until_complete(self.__stop())

    def destroy(self):
        """Removes the server instance."""
        self.stop()
        rmtree(self.datadir, ignore_errors=True)

    def set_safe_port(self):
        """Set the server port to something not being used."""
        port = 9090
        while port == 9090:
            sock = socket()
            if "address" in self.settings:
                address = self.settings["address"]
            else:
                address = "127.0.0.1"
            sock.bind((address, 0))
            port = sock.getsockname()[1]
            sock.close()
        self.settings["port"] = port

    def set_to_host_ip(self):
        """Set the server to use something other than localhost."""
        hostname = gethostname()
        self.settings["address"] = gethostbyname(hostname)
        self.set_safe_port()