Refactored tests to include clusters.

This commit is contained in:
Jeff Baskin 2024-03-04 11:21:42 -05:00
parent ec3f033ff6
commit a493fbf387
3 changed files with 119 additions and 55 deletions

88
test/mtt_tc.py Normal file
View File

@ -0,0 +1,88 @@
"""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)

View File

@ -1,66 +1,42 @@
"""Tests for single server boot ups."""
import aiohttp
from asyncio import create_subprocess_exec
from pathlib import Path
from .mtt_tc import MTTClusterTC
from socket import gethostbyname, gethostname
from unittest import IsolatedAsyncioTestCase
class Servers:
"""Setup and run servers."""
def __init__(self):
"""Initialize class"""
self.servers = []
async def create_server(self, *args):
"""Create servers"""
app = Path.cwd().joinpath("target", "release", "morethantext")
if args:
cmd = list(args)
cmd.insert(0, app)
else:
cmd = [app]
self.servers.append(await create_subprocess_exec(*cmd))
async def destroy_servers(self):
"""destroy servers"""
for server in self.servers:
server.terminate()
await server.wait()
class SingleBootTC(IsolatedAsyncioTestCase):
"""Test single server boot up."""
async def asyncSetUp(self):
"""Test Case setup"""
self.servers = Servers()
async def asyncTearDown(self):
await self.servers.destroy_servers()
class BootUpTC(MTTClusterTC):
"""Single server boot tests."""
async def test_default_boot(self):
"Are the default settings available?"""
await self.servers.create_server()
async with aiohttp.ClientSession() as session:
async with session.get("http://localhost:3000") as response:
"""Does the server default boot on http://localhost:3000?"""
await self.create_server_with_flags()
def tests(response):
"""Response tests."""
self.assertEqual(response.status, 200)
await self.run_tests("/", tests)
async def test_alt_port_boot(self):
"""Can the server boot off on alternate port?"""
port = 9025
await self.servers.create_server("-p", str(port))
async with aiohttp.ClientSession() as session:
async with session.get(f"http://localhost:{port}") as response:
await self.create_server_with_flags("-p", str(port))
def tests(response):
"""Response tests."""
self.assertEqual(response.status, 200)
await self.run_tests("/", tests)
async def test_alt_address_boot(self):
"""Can it boot off an alternate address?"""
addr = gethostbyname(gethostname())
await self.servers.create_server("-a", addr)
async with aiohttp.ClientSession() as session:
async with session.get(f"http://{addr}:3000") as response:
await self.create_server_with_flags("-a", addr)
def tests(response):
"""Response tests."""
self.assertEqual(response.status, 200)
await self.run_tests("/", tests)
async def test_for_session_id(self):
await self.create_server()
def tests(response):
"""Response tests."""
self.assertEqual(response.status, 200)
await self.run_tests("/", tests)
self.assertEqual(len(self.jar), 1, "There should be a session id.")