89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
|
"""Base for MoreThanTest test cases."""
|
||
|
|
|
||
|
|
from aiohttp import ClientSession, CookieJar
|
||
|
|
from asyncio import create_subprocess_exec
|
||
|
|
from pathlib import Path
|
||
|
|
from socket import socket
|
||
|
|
from unittest import IsolatedAsyncioTestCase
|
||
|
|
|
||
|
|
LOCALHOST = "127.0.0.1"
|
||
|
|
|
||
|
|
class Server:
|
||
|
|
"""Setup and run servers."""
|
||
|
|
|
||
|
|
def __init__(self, *args):
|
||
|
|
"""Initialize class"""
|
||
|
|
app = Path.cwd().joinpath("target", "release", "morethantext")
|
||
|
|
addr = "127.0.0.1"
|
||
|
|
port = 3000
|
||
|
|
if args:
|
||
|
|
self.cmd = list(args)
|
||
|
|
self.cmd.insert(0, app)
|
||
|
|
get_port = False
|
||
|
|
get_addr = False
|
||
|
|
for item in args:
|
||
|
|
if get_port:
|
||
|
|
port = item
|
||
|
|
get_port = False
|
||
|
|
if get_addr:
|
||
|
|
addr = item
|
||
|
|
get_addr = False
|
||
|
|
if item == "-a" or item == "--address":
|
||
|
|
get_addr = True
|
||
|
|
if item == "-p" or item == "--port":
|
||
|
|
get_port = True
|
||
|
|
else:
|
||
|
|
self.cmd = [app]
|
||
|
|
self.server = None
|
||
|
|
self.host = f"http://{addr}:{port}"
|
||
|
|
|
||
|
|
async def create(self):
|
||
|
|
"""Cerate the server"""
|
||
|
|
self.server = await create_subprocess_exec(*self.cmd)
|
||
|
|
|
||
|
|
async def destroy(self):
|
||
|
|
"""destroy servers"""
|
||
|
|
self.server.terminate()
|
||
|
|
await self.server.wait()
|
||
|
|
|
||
|
|
|
||
|
|
class MTTClusterTC(IsolatedAsyncioTestCase):
|
||
|
|
"""Test case for MoreThanTText."""
|
||
|
|
|
||
|
|
async def asyncSetUp(self):
|
||
|
|
"""Test setup"""
|
||
|
|
self.servers = []
|
||
|
|
self.jar = CookieJar(unsafe=True)
|
||
|
|
self.session = ClientSession(cookie_jar=self.jar)
|
||
|
|
|
||
|
|
async def asyncTearDown(self):
|
||
|
|
"""Test tear down."""
|
||
|
|
await self.session.close()
|
||
|
|
for server in self.servers:
|
||
|
|
await server.destroy()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
async def get_port():
|
||
|
|
"""Retrieve an unused port."""
|
||
|
|
sock = socket()
|
||
|
|
sock.bind((LOCALHOST, 0))
|
||
|
|
port = sock.getsockname()[1]
|
||
|
|
sock.close()
|
||
|
|
return port
|
||
|
|
|
||
|
|
async def create_server_with_flags(self, *args):
|
||
|
|
"""Create a single server with flags."""
|
||
|
|
server = Server(*args)
|
||
|
|
await server.create()
|
||
|
|
self.servers.append(server)
|
||
|
|
|
||
|
|
async def create_server(self):
|
||
|
|
port = await self.get_port()
|
||
|
|
await self.create_server_with_flags("-p", str(port))
|
||
|
|
|
||
|
|
async def run_tests(self, uri, func):
|
||
|
|
"""Run the tests on each server."""
|
||
|
|
for server in self.servers:
|
||
|
|
async with self.session.get(f"{server.host}{uri}") as response:
|
||
|
|
func(response)
|