"""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()