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 pathlib import Path
from socket import socket


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

    def __init__(self):
        """Initialization of a server."""
        self.env = {}
        self.process = None
        self.loop = get_event_loop()

    @property
    def address(self):
        """Get the server address."""
        address = "127.0.0.1"
        if "MMT_ADDRESS" in self.env:
            address = self.env["MMT_ADDRESS"]
        return address

    @address.setter
    def address(self, addr):
        """Sets the server address."""
        self.env["MTT_ADDRESS"] = addr

    @property
    def port(self):
        """Get the port of the server."""
        port = "9090"
        if "MTT_PORT" in self.env:
            port = self.env["MTT_PORT"]
        return port

    @port.setter
    def port(self, num):
        """Set the port for the server."""
        self.env["MTT_PORT"] = str(num)

    async def __start(self):
        """async start of the server."""
        print(self.env)
        self.process = await create_subprocess_exec(
            Path.cwd().joinpath("target", "release", "morethantext_web"), env=self.env
        )
        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()

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