99 lines
2.8 KiB
Python
99 lines
2.8 KiB
Python
"""Base for MoreThanTest test cases."""
|
|
|
|
from asyncio import create_subprocess_exec, sleep
|
|
from pathlib import Path
|
|
from socket import socket
|
|
from unittest import IsolatedAsyncioTestCase
|
|
from aiohttp import ClientSession
|
|
|
|
|
|
LOCALHOST = "127.0.0.1"
|
|
SESSION_KEY = "sessionid"
|
|
HOST = "example.com"
|
|
|
|
|
|
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 in ("-a", "--address"):
|
|
get_addr = True
|
|
if item in ("-p", "--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)
|
|
await sleep(1)
|
|
|
|
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.cookies = {}
|
|
self.session = ClientSession()
|
|
|
|
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):
|
|
"""Create a server on a random port."""
|
|
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}", cookies=self.cookies
|
|
) as response:
|
|
if SESSION_KEY in response.cookies:
|
|
self.cookies[SESSION_KEY] = response.cookies[SESSION_KEY].value
|
|
func(response)
|