92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""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.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
|
|
address = self.base_url.split(":")[0]
|
|
while port == 9090:
|
|
sock = socket()
|
|
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()
|