morethantext/release_tests/support/mttserver.py

"""Starts and runs releasse version of MoreThanText server."""

from asyncio import create_subprocess_exec
from socket import AF_INET, SOCK_STREAM, socket
from pathlib import Path

EXECUTABLE = Path.cwd().joinpath("target", "release", "morethantext")


class MTTServer:
    """Runs a MoreThanText server."""

    def __init__(self, *args):
        """Initialization"""
        self.port = 3000
        self.address = "127.0.0.1"
        self.cmd = [EXECUTABLE]
        for item in list(args):
            self.cmd.append(str(item))
        try:
            self.port = self.cmd[self.cmd.index("-p") + 1]
        except ValueError:
            pass
        try:
            self.port = self.cmd[self.cmd.index("--port") + 1]
        except ValueError:
            pass
        try:
            self.address = self.cmd[self.cmd.index("-a") + 1]
        except ValueError:
            pass
        try:
            self.address = self.cmd[self.cmd.index("--address") + 1]
        except ValueError:
            pass
        self.server = None

    @property
    def baseurl(self):
        """URL to server"""
        return f"http://{self.address}:{self.port}"

    async def start(self):
        """Starts the server."""
        self.server = await create_subprocess_exec(*self.cmd)
        # delays the return untul the serverr is responding.
        with socket(AF_INET, SOCK_STREAM) as soc:
            while soc.connect_ex((self.address, int(self.port))) != 0:
                continue

    async def stop(self):
        """Stops the server."""
        if self.server:
            if not self.server.returncode:
                self.server.terminate()
                await self.server.wait()

    async def cleanup(self):
        """Clean up."""
        await self.stop()